Spring Kafka 1.2.2优雅停机问题及偏移量异常提交求助

我来帮你解决这两个核心问题——SIGTERM触发后的无限重试困境，以及重启后消息被直接提交未处理的问题，同时给出适合1.2.x版本的优雅停机方案（不用升级到2.0+）。

一、先搞清楚问题根源 #

你的配置组合（AlwaysRetryPolicy+无退避的RetryTemplate+MANUAL_IMMEDIATE Ack模式）在SIGTERM时会触发连锁问题：

• 收到SIGTERM后，容器会等待当前消息处理完成，但此时容器已进入关闭流程，新拉取的消息在处理时会触发容器状态异常，而AlwaysRetryPolicy会让这个异常无限循环重试，最终只能用SIGKILL强制终止。
• 重启后offset被直接提交但未调用@KafkaListener，是因为无限重试期间，容器关闭时的异常场景导致offset被错误提交，重启后消费者认为该offset已处理完成，直接跳过。

针对重试与offset异常的修复步骤 #

1. 替换AlwaysRetryPolicy为有限重试策略
   无限重试本身就是不合理的配置，哪怕需要重试，也得设置上限并添加退避策略，避免资源耗尽：

   @Bean
   public RetryTemplate kafkaRetryTemplate() {
       RetryTemplate retryTemplate = new RetryTemplate();
       // 设置最大重试3次
       SimpleRetryPolicy retryPolicy = new SimpleRetryPolicy();
       retryPolicy.setMaxAttempts(3);
       retryTemplate.setRetryPolicy(retryPolicy);
       // 添加退避，避免频繁重试打垮服务
       FixedBackOffPolicy backOffPolicy = new FixedBackOffPolicy();
       backOffPolicy.setBackOffPeriod(1000); // 每次重试间隔1秒
       retryTemplate.setBackOffPolicy(backOffPolicy);
       return retryTemplate;
   }
   

   这样重试达到上限后，消息要么进入死信队列（如果配置的话），要么被正确处理失败逻辑，不会无限占用资源。

2. 严格控制offset提交时机
   在MANUAL_IMMEDIATE模式下，只有当消息100%处理成功后再调用Acknowledgment.acknowledge()。异常场景下绝对不要提交offset，确保重启后消息能被重新处理。

二、1.2.x版本实现SIGTERM后停止接收新消息的优雅停机 #

因为1.2.x没有KafkaListenerEndpointRegistry，我们需要手动管控容器实例，核心思路是：监听SIGTERM信号，先让容器停止拉取新消息，等待当前消息处理完成后再关闭容器。

方案：手动管理KafkaMessageListenerContainer+注册停机钩子 #

步骤1：手动创建容器（替代@KafkaListener自动配置）

这样我们能持有容器的引用，方便后续控制其状态：

@Configuration
public class CustomKafkaConfig {

    @Value("${kafka.topic}")
    private String topic;

    @Bean
    public KafkaMessageListenerContainer<String, String> kafkaListenerContainer(
            ConsumerFactory<String, String> consumerFactory,
            RetryTemplate kafkaRetryTemplate) {

        ContainerProperties containerProps = new ContainerProperties(topic);
        // 配置你的Ack模式和重试模板
        containerProps.setAckMode(ContainerProperties.AckMode.MANUAL_IMMEDIATE);
        containerProps.setRetryTemplate(kafkaRetryTemplate);
        // 设置消息处理逻辑（替代@KafkaListener）
        containerProps.setMessageListener((AcknowledgingMessageListener<String, String>) (message, acknowledgment) -> {
            try {
                // 这里调用你的业务处理方法
                handleMessage(message);
                // 处理成功才提交offset
                acknowledgment.acknowledge();
            } catch (Exception e) {
                // 异常时不提交offset，让消息重试或进入死信
                log.error("处理消息失败，offset: {}", message.offset(), e);
            }
        });

        KafkaMessageListenerContainer<String, String> container =
                new KafkaMessageListenerContainer<>(consumerFactory, containerProps);
        return container;
    }

    // 省略ConsumerFactory等其他必要配置
}

步骤2：注册SIGTERM信号处理器，优雅停止容器

通过Spring的上下文关闭事件和JVM停机钩子，双重保障容器的优雅停止：

@Component
public class KafkaShutdownHandler implements ApplicationListener<ContextClosedEvent> {

    @Autowired
    private KafkaMessageListenerContainer<String, String> kafkaListenerContainer;

    @Override
    public void onApplicationEvent(ContextClosedEvent event) {
        gracefullyShutdownContainer();
    }

    // 注册JVM停机钩子，捕获SIGTERM信号
    @PostConstruct
    public void registerShutdownHook() {
        Runtime.getRuntime().addShutdownHook(new Thread(this::gracefullyShutdownContainer));
    }

    private void gracefullyShutdownContainer() {
        // 先暂停容器拉取新消息
        kafkaListenerContainer.pause();
        log.info("Kafka容器已暂停拉取新消息");

        // 等待当前正在处理的消息完成（可根据业务调整超时时间）
        try {
            Thread.sleep(5000); // 等待5秒，确保当前消息处理完成
        } catch (InterruptedException e) {
            Thread.currentThread().interrupt();
            log.warn("等待消息处理完成时被中断");
        }

        // 正式停止容器
        kafkaListenerContainer.stop();
        try {
            // 等待容器完全停止
            if (kafkaListenerContainer.awaitStop(3000)) {
                log.info("Kafka容器已优雅停止");
            } else {
                log.warn("Kafka容器停止超时，将强制终止");
            }
        } catch (InterruptedException e) {
            Thread.currentThread().interrupt();
            log.error("等待容器停止时发生异常", e);
        }
    }
}

这个方案的核心是：pause()让容器不再向Kafka Broker拉取新消息，等待当前已拉取到本地的消息处理完成后，再调用stop()彻底关闭容器，完美实现SIGTERM后不再接收新消息的需求。

额外优化建议 #

• 配置死信队列（DLQ）：当重试达到上限后，将无法处理的消息转发到DLQ，避免消息丢失，也便于后续排查问题。
• 增加容器状态监控：在日志中记录容器的启停、暂停状态，便于排查停机过程中的异常。
• 测试停机流程：模拟SIGTERM信号，验证容器是否能优雅停止，消息是否能正确处理或重试。

内容的提问来源于stack exchange，提问作者Siva praneeth Alli