Confluent Kafka手动偏移量提交不符合预期问题排查

环境与配置 #

• 使用confluent_kafka 2.0.2版本测试手动偏移提交代码
• Kafka环境：confluentinc/cp-kafka:7.3.0，单分区主题test-topic2
• Consumer核心配置：enable.auto.offset_store: False、enable.auto.commit: False

测试代码 #

from confluent_kafka import Consumer, TopicPartition


LOCAL_TEST_TOPIC = "test-topic2"
LOCAL_TEST_PRODUCER_CONFIG = {
    'bootstrap.servers': '0.0.0.0:9092',
    'group.id': "tail-grouptest",
    'enable.auto.offset.store': False,
    'enable.auto.commit': False,
}

consumer = Consumer(LOCAL_TEST_PRODUCER_CONFIG)
topics = [LOCAL_TEST_TOPIC, ]


def msg_process(msg):
    if not msg:
        print("EMPTY MESSAGE")
        return
    print(msg.value().decode('utf-8'))


def basic_consume_loop(consumer, topics):
    try:
        consumer.subscribe(topics)
        topic = consumer.list_topics(topic='test-topic2')
        partitions = [TopicPartition('test-topic2', partition)
                      for partition in list(topic.topics['test-topic2'].partitions.keys())]

        # msg1
        msg1 = consumer.poll()
        msg_process(msg1)
        consumer.store_offsets(message=msg1)
        consumer.commit(asynchronous=False)
        print(consumer.position(partitions))

        # msg2
        msg2 = consumer.poll()
        msg_process(msg2)
        consumer.store_offsets(message=msg2)
        consumer.commit(asynchronous=False)
        print(consumer.position(partitions))

        # msg3
        msg3 = consumer.poll()
        msg_process(msg3)

        # msg4
        msg4 = consumer.poll()
        msg_process(msg4)

        # msg5
        msg5 = consumer.poll()
        msg_process(msg5)
        print(consumer.position(partitions))

        # msg6
        msg6 = consumer.poll()
        msg_process(msg6)
        print(consumer.position(partitions))

        print("set back to msg3")
        # back to msg3
        consumer.store_offsets(message=msg3)
        print("commit again")
        consumer.commit(message=msg3, asynchronous=False)
        print(consumer.position(partitions))

        msg_new = consumer.poll()
        msg_process(msg_new)

    finally:
        # Close down consumer to commit final offsets.
        consumer.close()


def main():
    basic_consume_loop(consumer, topics)


if __name__ == '__main__':
    main()

发送的测试消息 #

通过控制台生产者发送以下消息：

1
2
3
4
5
6
7

脚本运行结果 #

1
[TopicPartition{topic=test-topic2,partition=0,offset=1,error=None}]
2
[TopicPartition{topic=test-topic2,partition=0,offset=2,error=None}]
3
4
5
[TopicPartition{topic=test-topic2,partition=0,offset=5,error=None}]
6
[TopicPartition{topic=test-topic2,partition=0,offset=6,error=None}]
set back to msg3
commit again
[TopicPartition{topic=test-topic2,partition=0,offset=6,error=None}]
7

问题描述 #

已配置禁用自动偏移存储和自动提交，但每次调用poll()后偏移量仍自动递增。预期在将偏移量回滚到msg3后，后续poll()会重复消费msg3，但实际偏移量持续递增，consumer继续拉取新消息。请问代码存在什么问题？

解答 #

核心问题分析 #

1. consumer.position()返回的是本地消费位置
   你看到的position()输出是consumer本地维护的「下一次要拉取的偏移量」，即使你提交了旧偏移量到broker，只要已经通过poll()拉取过后续消息，本地消费位置已经更新，不会自动回退。

2. 未主动重置本地消费位置
   commit()只是把偏移量同步到Kafka的__consumer_offsets主题，但不会修改当前consumer实例的本地消费位置。后续poll()依然会从本地记录的位置继续拉取，而非从broker读取已提交的旧偏移量。

修复方案 #

要实现重复消费msg3及后续消息，需要在提交旧偏移量后，主动调用consumer.seek()重置本地消费位置：

修改代码中回滚偏移量的部分：

print("set back to msg3")
# back to msg3
# 提交偏移量到broker
consumer.commit(message=msg3, asynchronous=False)
# 主动重置本地消费位置到msg3的偏移量（msg.offset()是当前消息的偏移量，seek后下一次poll会拉取该位置的消息）
consumer.seek(TopicPartition('test-topic2', 0, msg3.offset()))
print(consumer.position(partitions))

msg_new = consumer.poll()
msg_process(msg_new)

补充说明 #

• store_offsets()仅用于更新本地偏移量缓存，当enable.auto.offset_store=False时需手动调用，但如果后续要重置位置，这一步可省略，直接通过commit()+seek()完成。
• seek()是直接修改consumer本地的消费位置，决定下一次poll()的拉取起点；而commit()仅负责将偏移量持久化到broker，仅在consumer重启或分区重新分配时才会生效。

内容的提问来源于stack exchange，提问作者WindMGC