Spring Boot Kafka消费者重启后未提交消息丢失问题求助

核心问题原因 #

你遇到的问题本质是Spring Kafka默认会在消费者容器停止时自动提交当前已消费的偏移量，哪怕你已经设置了enable.auto.commit=false和MANUAL_IMMEDIATE的手动确认模式。当应用意外停止或正常关闭时，容器会自动把那些已经消费但还没手动确认的偏移量提交到Kafka，导致重启后消费者直接从下一个偏移量开始消费。

解决步骤 #

1. 禁用容器停止时的自动提交 #

这是最关键的一步，需要显式关闭Spring Kafka容器在停止时的自动提交行为：

• 如果用Spring Boot配置文件（application.properties/application.yml）：

spring.kafka.listener.commit-offsets-on-stop=false

• 如果用Java配置类：

@Bean
public ConcurrentKafkaListenerContainerFactory<String, String> kafkaListenerContainerFactory() {
    ConcurrentKafkaListenerContainerFactory<String, String> factory = new ConcurrentKafkaListenerContainerFactory<>();
    factory.setConsumerFactory(consumerFactory());
    ContainerProperties containerProps = factory.getContainerProperties();
    containerProps.setAckMode(AckMode.MANUAL_IMMEDIATE);
    containerProps.setCommitOffsetsOnStop(false); // 关闭停止时的自动提交
    return factory;
}

2. 修正监听器参数（可选但规范） #

你的监听器中@Header int offsets的参数绑定不正确，应该使用KafkaHeaders.OFFSET常量来获取偏移量，避免后续潜在问题：

@KafkaListener(topics = "your-topic", groupId = "your-group-id")
public void consume(@Payload String message,
                    @Header(KafkaHeaders.PARTITION_ID) int partition,
                    @Header(KafkaHeaders.OFFSET) long offset,
                    Acknowledgement acknowledgement) {
    try {
        process(message);
        // 只有处理成功才手动提交偏移量
        acknowledgement.acknowledge();
    } catch (Exception e) {
        // 处理失败时不提交，重启后会重新消费该偏移量的消息
        System.err.printf("处理消息失败，partition: %d, offset: %d%n", partition, offset);
    }
}

3. 确认自动偏移重置策略 #

确保auto.offset.reset配置为earliest（如果需要从头消费未提交的消息），不过这个配置只有在消费者组没有已提交的偏移量时才会生效，主要用于兜底：

spring.kafka.consumer.auto-offset-reset=earliest

验证方案 #

按照上述配置修改后，重新测试：

1. 运行应用，消费offset1、2并提交，消费offset3后不提交（比如在process方法中抛出异常，或手动停止应用）
2. 重启应用，此时消费者应该会从offset3开始消费，而不是直接跳到offset4

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