如何在服务宕机或异常时让Kafka Consumer重放/获取相同消息？

算力阿峰

2026-7-20

问题描述

我想实现当服务宕机或发生异常时，Kafka Consumer能重放/获取相同消息。目前我的理解是：当设置auto commit为false、ack mode为manual immediate时，若发生异常则不向broker发送ack，使用相同consumer group id重启消费者后就能重新读取消息。

我的@KafkaListener代码如下：

@KafkaListener(topics = "testtopic", groupId = "testgroupID")
public void listenGroupFoo(String message, Acknowledgment acknowledgment,
                           @Header(KafkaHeaders.OFFSET) int offsets,
                           @Header(value = KafkaHeaders.RECEIVED_MESSAGE_KEY, required =false) Integer key,
                           @Payload String payload,
                           @Header(KafkaHeaders.RECEIVED_PARTITION_ID) int partition,
                           @Header(KafkaHeaders.RECEIVED_TOPIC) String topic,
                           @Header(KafkaHeaders.RECEIVED_TIMESTAMP) long ts
                           ) {
                
                 //performing some process like insert to DB or 
                 calling some downstream process, once this processes is done 
                 I would like to inform kafka broker that I'm done with processing 
                 by implementing  acknowledgment.acknowledge(); for each individual message
                 one by one
}

ConsumerFactory配置：

enable.auto.commit = false
auto.offset.reset = earliest

ConcurrentKafkaListenerContainerFactory配置：

factory.getContainerProperties().setAckMode(ContainerProperties.AckMode.MANUAL_IMMEDIATE);

请问该配置是否能实现预期的消息重放功能？

回答

你的配置完全可以实现预期的消息重放功能，核心逻辑和配置的匹配度很高，具体原因如下：

enable.auto.commit = false：关闭自动提交偏移量，避免Kafka后台自动确认消息已处理，确保只有手动触发确认后，偏移量才会提交到Broker。
AckMode.MANUAL_IMMEDIATE：调用acknowledgment.acknowledge()会立即将当前消息的偏移量提交给Broker；如果处理过程中服务宕机或抛出异常，这条消息的偏移量不会被提交。
auto.offset.reset = earliest：消费者重启时，若对应分区无有效提交偏移量，会从分区起始位置重新拉取消息，保证未被确认的消息能被重新消费。

另外需要注意几个细节：

务必在业务逻辑完全执行成功（比如DB插入完成、下游调用返回成功）后，再调用acknowledge()提交偏移量，避免出现业务失败但偏移量已提交的情况。
重启消费者时必须使用相同的groupId，否则Kafka会将其视为新消费组，无法复用之前的偏移量记录。
建议给业务逻辑增加幂等性处理，避免因偏移量提交失败导致的重复消费引发数据异常。

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