KafkaListener消费Topic X时无限重复执行问题排查

问题描述 #

调用接口发送消息至Topic X，消费者通过@KafkaListener监听该主题，预期处理失败时将消息转发至X.error主题，但目前出现listenX方法无限重复执行的情况，且消息已成功发送至X.error主题。

环境信息 #

• Kafka版本：3.1.2
• Spring Boot版本：2.7.7

代码片段 #

@KafkaListener(topics = "X", groupId = "X-consumer", properties = "max.poll.interval.ms:1000")
public void listenX(KafkaMessageDTO msg) {
    // 此处为CRUD处理代码
    kafkaTokenMessageService.sendMessage("X.error", msg);
}

补充说明 #

• 设置max.poll.interval.ms:1000原本期望实现“每秒监听一次主题”，但未达到预期效果
• listenX方法仅包含CRUD操作，无其他额外逻辑
• @KafkaListener注解来自org.springframework.kafka.annotation.KafkaListener，查阅文档未找到properties参数的冒号分隔写法，但日志无错误或警告输出

核心原因及解决方案 #

1. properties参数配置错误 #

@KafkaListener的properties属性要求传入键值对数组，而非单个冒号分隔的字符串。你当前的写法properties = "max.poll.interval.ms:1000"不会生效，正确配置方式为：

@KafkaListener(topics = "X", groupId = "X-consumer", properties = {"max.poll.interval.ms=1000"})

同时需要纠正对该参数的误解：它并非控制“每秒监听一次”，而是限制消费者两次poll请求的最大间隔时间——若超过该时间未发起新的poll，Kafka集群会判定消费者失效并触发重平衡，与无限重复执行无直接关联。

2. 未正确处理重试与偏移量提交 #

listenX无限重复执行的核心原因是消息重试机制未被正确控制或偏移量未正常提交：

• 若CRUD操作抛出异常，Spring Kafka默认会无限重试该消息（默认重试次数为Integer.MAX_VALUE），即使你在方法中发送了消息到X.error，只要异常未被捕获处理，重试就会持续。
• 若方法正常执行但偏移量未提交（比如开启手动提交但未调用提交逻辑），消费者重启或重平衡后会重新消费同一批消息。

3. 正确的处理方案 #

方案一：手动捕获异常并控制偏移量

如果需要在处理失败时转发消息到X.error且停止重试，可捕获CRUD操作的异常，发送错误消息后手动确认偏移量：

@KafkaListener(topics = "X", groupId = "X-consumer")
public void listenX(KafkaMessageDTO msg, Acknowledgment ack) {
    try {
        // 执行CRUD处理代码
        ack.acknowledge(); // 处理成功，手动提交偏移量
    } catch (Exception e) {
        kafkaTokenMessageService.sendMessage("X.error", msg);
        ack.acknowledge(); // 处理失败，提交偏移量避免重复消费
    }
}

同时需在配置文件中关闭自动提交：

spring:
  kafka:
    consumer:
      enable-auto-commit: false

方案二：使用Spring Kafka死信处理器

配置DeadLetterPublishingRecoverer和SeekToCurrentErrorHandler，自动将处理失败的消息转发至死信主题（X.error），并停止重试：

@Bean
public ErrorHandler errorHandler(KafkaTemplate<String, Object> kafkaTemplate) {
    DeadLetterPublishingRecoverer recoverer = new DeadLetterPublishingRecoverer(kafkaTemplate,
            (record, ex) -> new TopicPartition("X.error", record.partition()));
    // 0次重试，直接转发死信
    return new SeekToCurrentErrorHandler(recoverer, new FixedBackOff(0L, 0L));
}

此时监听方法只需正常执行CRUD逻辑，异常会被自动捕获并处理：

@KafkaListener(topics = "X", groupId = "X-consumer")
public void listenX(KafkaMessageDTO msg) {
    // CRUD处理代码，抛出异常时自动触发死信转发
}

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