SpringBoot Kafka Consumer配置earliest后重启未从头消费问题排查

为什么auto.offset.reset=earliest没生效？ #

auto.offset.reset=earliest的生效有严格前提：仅当消费者组在Kafka集群中无已提交的偏移量记录，或者**已有的偏移量已失效（如日志被清理）**时，才会触发从头消费逻辑。

你的场景中，消费者组之前已经消费过并提交了偏移量，哪怕重启应用，Kafka会直接读取已提交的偏移量继续消费，不会触发该配置。手动修改偏移量后能从头消费，但重启后回到最新位置，是因为消费过程中偏移量被自动提交，重启后读取的是最新提交的记录。

实现重启即从头消费的配置方案 #

方案1：启动时强制重置偏移量 #

通过自定义重平衡监听器，在消费者加入消费组时主动将偏移量重置到分区起始位置：

@Service
@Getter
@Slf4j
public class KafkaConsumerService {

  private final Map<String, Set<String>> fleetVinsMap = new HashMap<>();

  @KafkaListener(topics = "${app.conf.source.kafka.topic}",
      containerFactory = "heartbeatKafkaListenerContainerFactory",
      id = "heartbeatConsumer")
  public void consume(String message) {
      // 你的消费逻辑
  }

  @Bean
  public ConsumerAwareRebalanceListener rebalanceListener() {
      return new ConsumerAwareRebalanceListener() {
          @Override
          public void onPartitionsAssigned(Consumer<?, ?> consumer, Collection<TopicPartition> partitions) {
              // 重置所有分配到的分区到起始位置
              consumer.seekToBeginning(partitions);
              log.info("已将偏移量重置到分区起始位置，涉及分区：{}", partitions);
          }
      };
  }

  // 确保容器工厂关联自定义监听器和你的consumerFactory
  @Bean
  public ConcurrentKafkaListenerContainerFactory<String, String> heartbeatKafkaListenerContainerFactory(ConsumerFactory<String, String> consumerFactory, ConsumerAwareRebalanceListener rebalanceListener) {
      ConcurrentKafkaListenerContainerFactory<String, String> factory = new ConcurrentKafkaListenerContainerFactory<>();
      factory.setConsumerFactory(consumerFactory);
      factory.getContainerProperties().setConsumerRebalanceListener(rebalanceListener);
      return factory;
  }
}

方案2：禁用自动提交偏移量 #

如果不需要保留消费进度，可关闭自动提交，结合auto.offset.reset=earliest实现每次重启从头消费：

在你的consumerFactory中添加配置：

config.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, false);

注意：这种方式下，若后续需要保留消费进度，需在消费逻辑中手动调用consumer.commitSync()或consumer.commitAsync()提交偏移量。

方案3：临时删除消费者组偏移量（一次性操作） #

若仅需临时从头消费，可通过Kafka命令行工具删除目标消费者组的已提交偏移量：

kafka-consumer-groups.sh --bootstrap-server <你的Broker地址> --delete --group <你的消费者组ID>

删除后重启应用，消费者会因无已提交偏移量触发auto.offset.reset=earliest，但后续消费后偏移量会重新被提交，重启仍会从提交位置继续。

关键配置检查 #

确保你的heartbeatKafkaListenerContainerFactory确实关联了自定义的consumerFactory，否则所有consumer配置（包括auto.offset.reset）都不会生效。

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