Spring Boot Kafka手动提交偏移后仍触发重试问题咨询

我有一个Spring Boot应用，Kafka消费逻辑如下：

@KafkaListener(topics = "journal-topic")
public void onMessage(ConsumerRecord<String,String> consumerRecord, Acknowledgment acknowledgment) {
    acknowledgment.acknowledge();
    var message = extractMessage(consumerRecord);
    messageService.saveOrUpdateMessage(message);
}

其中extractMessage(consumerRecord)会抛出IllegalArgumentException，但单元测试中该方法被重试了6次——明明已经在方法开头调用acknowledge()提交了偏移，理论上不该触发重试。我的Kafka配置如下：

kafka:
  topic: "journal-topic"
  properties:
    auto-create-topics-enable: true

  consumer:
    bootstrap-servers: localhost:9092
    group-id: "journal-group"
    key-deserializer: org.apache.kafka.common.serialization.StringDeserializer
    value-deserializer: org.apache.kafka.common.serialization.StringDeserializer
    auto-offset-reset: latest
    max-poll-records: 10
    enable-auto-commit: false
  listener:
      ack-mode: manual_immediate
      concurrency: 3

问题原因 #

1. 偏移提交的异步特性：manual_immediate模式下，acknowledge()调用触发的是异步偏移提交操作，并非立刻完成。当方法后续抛出异常时，Spring Kafka的错误处理器可能在偏移提交完成前就触发了重试逻辑。
2. 默认错误处理器的重试逻辑：Spring Kafka默认使用DefaultErrorHandler，它仅依据方法是否抛出异常来触发重试，不会检查偏移是否已提交。只要方法抛出未捕获的异常，就会按照默认重试策略执行重试（你看到的6次可能是初始调用加5次重试，取决于环境默认配置）。
3. 批次消费的偏移提交逻辑：max-poll-records=10表示一次拉取10条记录，acknowledge()默认提交的是当前批次最后一条记录的偏移。如果当前失败的是批次中的某条记录，后续重试可能重复消费同批次内的其他记录，但你的场景是单条记录处理，这个影响相对较小。

解决方案 #

1. 自定义错误处理器，跳过已提交偏移的重试

通过自定义ErrorHandler，在重试前判断偏移状态，若已提交则终止重试：

@Bean
public ErrorHandler kafkaErrorHandler() {
    // 设置重试次数为0，或根据需求调整
    DefaultErrorHandler errorHandler = new DefaultErrorHandler(
        (record, ex) -> log.error("处理消息失败，偏移已提交，不再重试: {}", record.offset(), ex),
        new FixedBackOff(1000L, 0L)
    );

    // 添加重试监听器，直接终止重试
    errorHandler.setRetryListeners((record, ex, deliveryAttempt) -> {
        throw new StopRetryException("偏移已提交，终止重试", ex);
    });

    return errorHandler;
}

2. 调整偏移提交时机（可选）

如果业务允许，将acknowledge()移到消息处理成功之后，确保只有处理成功才提交偏移，异常时重试符合预期：

@KafkaListener(topics = "journal-topic")
public void onMessage(ConsumerRecord<String,String> consumerRecord, Acknowledgment acknowledgment) {
    try {
        var message = extractMessage(consumerRecord);
        messageService.saveOrUpdateMessage(message);
        acknowledgment.acknowledge();
    } catch (IllegalArgumentException e) {
        log.error("消息处理失败", e);
        throw e;
    }
}

3. 禁用默认重试

通过配置直接关闭重试机制，自行处理异常：

kafka:
  listener:
    retry:
      enabled: false

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