enable_auto_commit=False，但仍然会自动提交偏移量答案

【问题标题】：enable_auto_commit=False, but still offsets are committed automaticallyenable_auto_commit=False，但仍然会自动提交偏移量
【发布时间】：2023-01-13 07:46:36
【问题描述】：

我正在使用kafka-python==2.0.2，并禁用了auto_commit，但如果我不通过代码提交，偏移量会自动提交

在下面的代码中，即使我注释掉self.consumer.commit_async(callback= ....，偏移量仍然会被提交

class KafkaMessageConsumer:
    def __init__(self, bootstrap_servers: str, topic: str, group_id: str, offset_reset_strategy: str):
        self.bootstrap_servers: str = bootstrap_servers
        self.topic: str = topic
        self.group_id: str = group_id
        self.consumer: KafkaConsumer = KafkaConsumer(topic, bootstrap_servers=bootstrap_servers, group_id=group_id,
                                                     enable_auto_commit=False, auto_offset_reset=offset_reset_strategy)

    def consume_messages(self, consumer_poll_timeout: int, max_poll_records: int,
                         message_handler: MessageHandlerImpl = MessageHandlerImpl()):
        try:
            while True:
                try:
                    msg_pack = self.consumer.poll(timeout_ms=consumer_poll_timeout, max_records=max_poll_records)
                    if bool(msg_pack):
                        for topic_partition, messages in msg_pack.items():
                            message_handler.process_messages(messages)

                        self.consumer.commit_async(callback=(lambda offsets, response: log.error(
                            f"Error while committing offset in async due to: {response}", exc_info=True) if isinstance(
                            response, Exception) else log.debug(f"Successfully committed offsets: {offsets}")))
                except Exception as e:
                    log.error(f"Error while consuming/processing message due to: {e}", exc_info=True)

        finally:
            log.error("Something went wrong, closing consumer...........")
            self.consumer.close()

这是禁用自动提交和手动提交的正确方法吗？

【问题讨论】：

您使用什么来检查偏移量是否已提交？
我期待，如果我不手动提交偏移量，那么在下一次迭代调用 poll() 应该返回相同的记录，但它不是

标签： python apache-kafka kafka-consumer-api kafka-python

【解决方案1】：

poll() 总是向前移动，无论是否提交。它只是不将偏移量写回 Kafka.

如果你重启你的应用,然后它会一次又一次地获取相同的数据吗？

【讨论】：