Spring-Kafka长任务重复执行问题求助（Offset提交失败）

从你提供的日志和代码来看，任务反复重启主要有两个核心原因：

1. 你的任务耗时2小时，远超Kafka消费者默认的max.poll.interval.ms（默认5分钟），导致消费者长时间未调用poll()，Kafka集群判定该消费者已失效，触发组重平衡并重新分配分区，进而导致任务重复执行。
2. 你尝试用Runnable但调用task.run()是同步执行，并没有释放监听线程，线程仍会被阻塞2小时，依然会触发超时；另外你的消费者组是自动生成的匿名ID（日志里的anonymous.xxx），这会导致每个实例（或重启后的实例）都作为独立消费组接收消息，进一步加剧重复执行的问题。

下面是具体的修复方案：

1. 配置固定消费者组ID（最关键！） #

匿名消费组会让每个消费者实例都独立接收topic的所有消息，必须设置固定的消费组ID，确保同一条消息只会被组内一个实例处理：

spring:
  cloud:
    stream:
      bindings:
        consumer-message:
          destination: consumer-message
          contentType: application/json
          group: steak-processing-group  # 设置固定消费组ID
          # 其他配置...

2. 异步执行耗时任务，释放监听线程 #

将同步执行改为异步，让@StreamListener的线程立刻返回，确保消费者能持续调用poll()避免超时。推荐使用Spring自带的TaskExecutor（比手动创建线程更可控）：

@Component 
@Slf4j 
public class KafkaConsumer { 
    private final CommandRunnerService commandRunnerService;
    private final TaskExecutor taskExecutor;

    // 注入Spring默认的TaskExecutor
    public KafkaConsumer(CommandRunnerService commandRunnerService, TaskExecutor taskExecutor) { 
        this.commandRunnerService = commandRunnerService; 
        this.taskExecutor = taskExecutor;
    } 

    @StreamListener(KafkaStreams.INPUT) 
    public void handleWorkUnit(@Payload Steak steak, @Headers MessageHeaders headers) { 
        // 异步提交任务，监听线程立刻返回
        taskExecutor.execute(() -> {
            try {
                log.info("开始执行长任务，牛排ID：{}", steak.getId());
                commandRunnerService.executeCreateSteak(steak);
                log.info("长任务执行完成，牛排ID：{}", steak.getId());
                
                // 任务完成后手动提交offset（需配合关闭自动提交）
                Acknowledgment acknowledgment = headers.get(KafkaHeaders.ACKNOWLEDGMENT, Acknowledgment.class);
                if (acknowledgment != null) {
                    acknowledgment.acknowledge();
                    log.info("已提交offset，牛排ID：{}", steak.getId());
                }
            } catch (Exception e) {
                log.error("长任务执行失败，牛排ID：{}", steak.getId(), e);
                // 可添加失败处理逻辑：比如重试、发送到死信队列等
            }
        });
    } 
}

3. 调整Kafka消费者参数适配长任务 #

修改配置关闭自动提交offset（确保任务完成后再提交），并调大max.poll.interval.ms：

spring:
  cloud:
    stream:
      kafka:
        binder:
          brokers: 192.168.0.100
        bindings:
          consumer-message:
            destination: consumer-message
            contentType: application/json
            group: steak-processing-group
            consumer:
              enable-auto-commit: false  # 关闭自动提交，手动控制offset
              max-poll-interval-ms: 7200000  # 设置为2小时（7200000毫秒），比任务最长耗时更长
              max-poll-records: 1  # 每次只拉取1条消息，避免同时处理多个长任务
              session-timeout-ms: 300000  # 会话超时保持默认5分钟即可，或按需调大
          consumer-response:
            destination: consumer-response
            contentType: application/json

方案说明 #

• 固定消费组ID：确保同一条消息只会被组内一个实例处理，避免多实例重复消费。
• 异步执行：监听线程立刻返回，消费者能持续调用poll()，不会触发max.poll.interval.ms超时，也就不会导致组重平衡和任务重新分配。
• 手动提交offset：只有任务成功完成后才提交offset，即使消费者重启，也会从未提交的offset开始处理，不会重复执行已完成的任务。
• 调大max.poll.interval.ms：给足够的时间处理任务，即使异步执行出现异常导致线程卡住，也不会被Kafka集群误判为消费者失效。

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