Kafka消费者关闭自动提交后未提交事件未在下次poll返回问题

你混淆了Kafka消费者的两个核心概念：消费者本地消费位置（position） 和 Broker端存储的已提交偏移量（committed offset）：

• enable.auto.commit=false 仅关闭了自动向Broker提交偏移量的逻辑，完全不影响消费者客户端内部维护的本地消费位置的前移。只要你调用poll()成功拉取到消息，客户端的本地消费位置就会自动向后移动，和你是否主动调用commitSync()没有任何关系。
• 只有当消费者发生重平衡、进程重启、或者你主动调用seek系列方法重置位置时，消费者才会读取Broker端存储的已提交偏移量，作为拉取消息的起始位置。
• 你代码中的pause()、resume()方法仅控制是否暂停拉取新消息，不会修改本地消费位置，所以暂停恢复后，消费者还是从之前已经移动到Event4的位置继续拉取，自然不会返回Event3。

核心调整逻辑 #

消费Event3失败时，除了暂停消费者，还需要主动调用seek()方法把本地消费位置重置回Event3的偏移量，这样恢复拉取后就会重新返回Event3。

修改后的关键代码示例 #

for (ConsumerRecord<String, String> record : records) {
    System.out.println(eventsCount + ":" + record.value());
    if (record.value().equals("Event3")) {
        System.out.println("consumer is pausing...... for about " + retryDelay + " seconds " + sdf.format(new Date()));
        // 新增：将当前分区的消费位置重置为当前消息的偏移量
        TopicPartition currentTp = new TopicPartition(record.topic(), record.partition());
        kafkaConsumer.seek(currentTp, record.offset());
        kafkaConsumer.pause(topicPartitions);
        pausedAt = new Date();      
        break;
    }else {
        kafkaConsumer.commitSync();
    }
}

额外优化建议 #

• 空参的commitSync()会默认提交所有已拉取消息的最大偏移量，当前你配置了max.poll.records=1不会有问题，如果后续调整该参数，建议手动指定要提交的偏移量，避免意外跳变。
• 对于多次消费失败的消息，可以引入死信队列存储，避免单个异常消息长期阻塞整个分区的消费。

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