spring kafka 手动提交（manual_immediate）下未调用 acknowledge() 时仍跳过消息，根本原因在于消费者仅在启动或再平衡时读取集群保存的 offset；运行中 offset 始终由内存维护，需主动 seek 回退才能重消费。
spring kafka 手动提交（manual_immediate）下未调用 acknowledge() 时仍跳过消息，根本原因在于消费者仅在启动或再平衡时读取集群保存的 offset；运行中 offset 始终由内存维护，需主动 seek 回退才能重消费。
在 Spring Kafka 中启用 MANUAL_IMMEDIATE AckMode 并禁用自动提交（enable-auto-commit=false），本意是实现“成功才提交、失败则重试”的精确控制。但许多开发者发现：即使不调用 Acknowledgment.acknowledge()，消费者依然自动前进到下一条消息——这并非 Bug，而是 Kafka 消费者协议与 Spring 封装逻辑共同作用的结果。
核心原理：Offset 管理分两层
-
集群层（Broker）：仅在消费者首次加入 group、发生 Rebalance 或显式 commit 时写入/读取
__consumer_offsets主题； -
客户端层（内存）：每次 poll 返回消息后，KafkaConsumer 内部会自动推进其本地 position（即下一个待拉取的 offset）。该 position 与是否调用
acknowledge()无关，只与 poll 行为绑定。
因此，acknowledge() 的作用不是阻止 position 前进，而是触发向 broker 提交当前已处理到的 offset。若不提交，下次重启或 rebalance 时，消费者将从上次成功 commit 的 offset 开始消费（而非内存 position），从而导致“丢失”未提交的消息。
正确实现消息重试的两种方式
✅ 方式一：内存内重试（推荐用于瞬时失败）
在监听器内捕获异常后，不提交 offset，并主动 seek 回当前消息 offset，使下一次 poll 重新拉取同一条消息：
@KafkaListener(topics = "test-topic", groupId = "test")
public void listen(String data, Acknowledgment ack, Consumer, ?> consumer) {
try {
// 业务逻辑（可能失败）
process(data);
ack.acknowledge(); // 成功：提交 offset
} catch (Exception e) {
// 失败：seek 回当前消息 offset，强制重试
long currentOffset = ((ConsumerRecord, ?>) ThreadLocalRecord.get()).offset();
String topic = ((ConsumerRecord, ?>) ThreadLocalRecord.get()).topic();
int partition = ((ConsumerRecord, ?>) ThreadLocalRecord.get()).partition();
consumer.seek(new TopicPartition(topic, partition), currentOffset);
// 注意：此处不调用 ack.acknowledge()
}
}
⚠️ 注意：
ConsumerRecord需通过@Header(KafkaHeaders.RAW_RECORD)或自定义RecordInterceptor获取；生产环境建议使用@KafkaListener的ContainerProperties注册ErrorHandlingDeserializer+DefaultAfterRollbackProcessor实现更健壮的重试。
✅ 方式二：借助 Spring Kafka 的重试与死信队列（DLQ）
配置 SeekToCurrentErrorHandler，它会在异常时自动 seek 并重试（支持最大重试次数 + 指数退避）：
@Bean
public ConcurrentKafkaListenerContainerFactory, ?> kafkaListenerContainerFactory(
ConsumerFactory<object object> consumerFactory) {
ConcurrentKafkaListenerContainerFactory<object object> factory =
new ConcurrentKafkaListenerContainerFactory();
factory.setConsumerFactory(consumerFactory);
factory.setConcurrency(1);
factory.getContainerProperties().setAckMode(ContainerProperties.AckMode.MANUAL_IMMEDIATE);
factory.getContainerProperties().setSyncCommits(true);
// 自动重试 3 次，间隔 1s，失败后发往 dlq-topic
SeekToCurrentErrorHandler errorHandler = new SeekToCurrentErrorHandler(
new DeadLetterPublishingRecoverer(kafkaTemplate,
(r, e) -> new TopicPartition("dlq-topic", r.partition())),
new FixedBackOff(1000L, 3L)
);
factory.setCommonErrorHandler(errorHandler);
return factory;
}</object></object>
关键配置验证清单
- ✅
spring.kafka.consumer.enable-auto-commit=false（禁用自动提交） - ✅
spring.kafka.consumer.max-poll-records=1（单条处理，避免批量干扰） - ✅
containerProps.setAckMode(MANUAL_IMMEDIATE)（立即响应 acknowledge 调用） - ✅
containerProps.setSyncCommits(true)（确保 commit 同步阻塞，避免异步丢失） - ❌ 不要依赖“不调用 acknowledge() 就能卡住 offset”——这是常见误解。
总结
MANUAL_IMMEDIATE 模式下，acknowledge() 控制的是 commit 行为，而非 position 移动行为。要实现可靠重试，必须显式 seek() 或使用 Spring Kafka 内置的错误处理器。忽略这一机制将导致消息“静默丢失”。始终记住：Kafka 消费者的可靠性不来自“不提交”，而来自“可控的 seek + 显式 commit”。
