为什么在kafka python中，消费者最后提交的消息在消费者重启后再次被消费？

Question

我将python kafka consumer的< code>auto_commit设置为< code>False，我正在手动提交消息。然而，重启后，消费者再次消费来自每个分区的最后一条消息。只有最后一个，不能再多。

这就是kafka消费群体所展示的：

TOPIC    PARTITION  CURRENT-OFFSET  LOG-END-OFFSET  LAG
my-topic 0          0               1               1
my-topic 1          3               4               1

不知道为什么会显示滞后，whu当前偏移设置为最后一条消息而不是下一条？当我提交偏移量3时，当前偏移量不应该移动到4吗？

我提交我使用的每条消息，但是在重启时，它总是再次使用最后一条消息。

编辑:这是我使用的代码:

self.subscriber = kafka.KafkaConsumer(self.consumer_topic, 
    client_id=self.consumer_name, group_id=group_id,                                              
    bootstrap_servers=self.consumer_bootstrap_server,                                         
    consumer_timeout_ms=timeout_ms, enable_auto_commit=False)

for record in self.subscriber:
    offset = CommittableOffset(record.topic, record.partition, record.offset)
    # process message
    partition = TopicPartition(record.topic, record.partition)
    offset = OffsetAndMetadata(record.offset, None)

    self.subscriber.commit({partition:offset})
    

Answer 1

事实证明，python kafka库的工作方式与我习惯的Java /Scala库有点不同。在Java /Scala库中，当我提交消息时，实际上提交的是消息偏移量1。在kafka-python库中，我必须自己将偏移量1添加到偏移量中。