Spring Boot Kafka批量消费者的幂等消息重试机制咨询

重试时的默认行为 #

当你的批量消费者处理批次X（包含A、B、C、D）失败并抛出异常时，只要未提交该批次的偏移量，下一次重试会拉取到完全相同的消息批次。

原因在于：Spring Kafka默认会在批次处理成功后才提交偏移量，失败时不会提交。Kafka消费者会从该分区最后一次提交的偏移量位置开始拉取消息，所以重试时会从批次X的第一条消息的偏移量重新拉取，只要你的max.poll.records配置值不小于原批次的消息数量，就能一次性拉取到A、B、C、D这四条消息。

即便发生消费者再均衡（比如当前消费者崩溃、被踢出消费组），新接管分区的消费者同样会从未提交的偏移量开始拉取，只要max.poll.records足够大，依然能拿到相同的批次。

如何确保重试时批次绝对一致 #

如果业务对批次一致性要求极高，可以通过以下配置和编码方式强化保障：

• 确保偏移量仅在批次完全处理成功后提交
  保持Spring Kafka默认的自动提交配置（enable.auto.commit=false），容器会自动在批次处理成功后提交偏移量；如果是手动提交偏移量，必须在myBusinessLogic()执行无异常后，再调用Acknowledgment.acknowledge()，失败时绝不提交。

• 避免重试期间发生消费者再均衡
  调整消费组的会话超时和心跳参数，给重试留出足够时间：

  spring.kafka.consumer.session.timeout.ms=300000
  spring.kafka.consumer.heartbeat.interval.ms=100000
  

  这样即使重试耗时较长，消费者也不会因为心跳超时被踢出消费组，避免分区再均衡导致的批次拉取变化。

• 固定max.poll.records配置
  不要动态修改该参数，确保其值不小于业务中最大的批次消息数量，保证从起始偏移量拉取时能一次性获取完整的原批次消息：

  spring.kafka.consumer.max.poll.records=100
  

• 使用SeekToCurrentErrorHandler处理重试
  配置该错误处理器并结合FixedBackOff.UNLIMITED_ATTEMPTS，它会在每次失败后让消费者回到原偏移量位置，确保下一次拉取的是同一批次：

  @Bean
  public SeekToCurrentErrorHandler errorHandler() {
      return new SeekToCurrentErrorHandler(new FixedBackOff(1000L, FixedBackOff.UNLIMITED_ATTEMPTS));
  }
  

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