Apache Kafka关闭自动提交后消费异常未重复消费全部消息原因排查

这是个非常典型的Kafka Consumer位移机制误解问题，我来帮你拆解清楚为什么第二次循环没有重新读取全部10条消息：

核心原因：Kafka Consumer的内存位移自动更新机制 #

当你设置auto.commit = false时，虽然不会自动提交位移到Kafka的位移主题（__consumer_offsets），但Consumer在调用poll()方法拉取到消息后，会自动更新内存中的本地位移指针——这个动作和你是否手动提交位移完全无关！

具体到你的场景：

• 首次循环调用poll()，一次性拉取了key为0-9的全部10条消息
• 此时Consumer内存里的本地位移已经被更新到这一批消息的最后位置（也就是key=9对应的位移+1）
• 当处理到key=2触发异常时，你确实没有提交位移到Broker，但内存里的位移已经被悄悄更新了
• 第二次循环调用poll()时，Consumer会直接使用内存中已更新的位移去Broker拉取消息，而不是你预期的“上次提交的位移”，所以自然不会重新拉取0-9的消息

验证这个结论的方法 #

你可以在代码里打印两个关键位移值：

• 调用poll()前，通过consumer.position(TopicPartition)获取当前的本地位移
• 调用poll()后，再次打印这个值，你会发现它已经跳到了这批消息的末尾

解决方法 #

要实现“异常时重新读取未提交的消息”，你需要手动管理位移的回滚：

1. 在拉取消息前记录起始位移：每次调用poll()前，先获取当前的已提交位移（通过consumer.committed(TopicPartition)），并保存下来
2. 异常时重置位移：当触发异常时，调用consumer.seek(TopicPartition, 已保存的提交位移)，把本地位移重置到上次提交的位置
3. 正常处理完消息后再手动提交：确保只有当所有消息都处理成功时，才调用consumer.commitSync()或consumer.commitAsync()提交位移

举个简化的代码示例：

boolean forceError = true;
TopicPartition tp = new TopicPartition("your_topic", 0);

while (true) {
    // 拉取前获取已提交的位移
    OffsetAndMetadata committedOffset = consumer.committed(tp);
    long startOffset = committedOffset != null ? committedOffset.offset() : 0;

    ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(1000));
    try {
        for (ConsumerRecord<String, String> record : records) {
            System.out.println("Processing key: " + record.key());
            if (forceError && record.key().equals("2")) {
                forceError = false;
                throw new RuntimeException("Force error for key=2");
            }
        }
        // 所有消息处理成功，提交位移
        consumer.commitSync();
    } catch (Exception e) {
        // 异常时重置位移到拉取前的已提交位置
        consumer.seek(tp, startOffset);
        System.err.println("Error occurred, resetting offset to: " + startOffset);
    }
}

这样修改后，首次异常时会把位移重置到0，第二次循环就会重新拉取全部10条消息了。

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