Kafka消费后如何提交消息offset？kafka-python包offset未提交怎么办

• 自动提交未触发就退出进程：你设置的enable_auto_commit=True+auto_commit_interval_ms=1000是异步自动提交，需要间隔1秒才会在后台触发提交动作，但你的代码拉取并处理完消息后直接结束进程，还没到提交的时间点，offset根本没有成功上报到Kafka broker，所以每次重启都会因为没有已提交的offset，触发auto_offset_reset='earliest'的规则，从头拉取所有消息。
• 自动提交的触发时机限制：kafka-python的自动提交逻辑只会在调用poll()方法时触发提交操作，你的代码只调用了一次poll()，处理完就退出，没有下一次poll()触发提交，也会导致本地更新的offset没有同步到broker。
• 偶发拉不到消息的原因：第一次启动消费组时，消费组需要和broker完成协调、分区分配等操作，你设置的poll(timeout_ms=500)超时时间过短，可能协调还没完成就超时返回空结果。

生产环境建议使用手动提交模式，避免自动提交异步丢提交、或者提交了但消息处理失败导致的消息丢失问题。

方案1：处理完消息后手动提交offset（最稳妥） #

直接关闭自动提交，处理完所有消息后手动同步提交offset，适配短任务场景：

from kafka import KafkaConsumer

consumer = KafkaConsumer(
    'quickstart-events',
    bootstrap_servers=['localhost:9092'],
    auto_offset_reset='earliest',
    enable_auto_commit=False, # 关闭自动提交
    group_id='my-group'
)

msg_pack = consumer.poll(max_records=10, timeout_ms=3000) # 加长超时时间避免协调阶段返回空
for tp,messages in msg_pack.items():
    for message in messages:
        print("%s:%d:%d: key=%s value=%s" % (tp.topic, tp.partition,
                                             message.offset, message.key,
                                             message.value))
# 处理完所有消息后手动提交offset
consumer.commit()
consumer.close()

方案2：保留自动提交，增加等待/触发逻辑 #

如果要保留自动提交，处理完消息后等待足够时间确保提交完成再退出：

from kafka import KafkaConsumer
import time

consumer = KafkaConsumer(
    'quickstart-events',
    bootstrap_servers=['localhost:9092'],
    auto_offset_reset='earliest',
    enable_auto_commit=True,
    group_id='my-group',
    auto_commit_interval_ms=1000,
)

msg_pack = consumer.poll(max_records=10,timeout_ms=3000,update_offsets=True)
for tp,messages in msg_pack.items():
    for message in messages:
        print("%s:%d:%d: key=%s value=%s" % (tp.topic, tp.partition,
                                             message.offset, message.key,
                                             message.value))
# 等待自动提交完成再退出
time.sleep(2)
consumer.close()

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