Spring Boot服务重启时从Offset 0消费Kafka消息配置无效求助

问题根源 #

当前配置无法实现每次重启从Offset 0消费的核心原因：

• 若消费组ID（GROUP_ID_CONFIG）未变化，Kafka已存储该组的消费偏移量记录，此时AUTO_OFFSET_RESET_CONFIG=earliest和@PartitionOffset(initialOffset="0")仅在消费组首次启动、无偏移量记录或偏移量过期时生效，重启时不会触发重置。
• 即使关闭了自动提交（ENABLE_AUTO_COMMIT_CONFIG=false），若代码中手动调用了Acknowledgment.acknowledge()提交偏移量，Kafka仍会留存该组的偏移量，下次重启会从上次提交位置继续消费。

解决方案 #

方案一：手动强制重置偏移量（推荐） #

通过实现ConsumerSeekAware接口，在消费者分配分区后强制将偏移量定位到0，不受消费组历史偏移量影响。

修改消费者类如下：

@Component
public class KafkaConsumer implements ConsumerSeekAware {

    @Override
    public void onPartitionsAssigned(Map<TopicPartition, Long> assignments, ConsumerSeekCallback callback) {
        // 遍历所有分配到的分区，强制seek到offset 0
        assignments.keySet().forEach(tp -> callback.seek(tp.topic(), tp.partition(), 0L));
    }

    // 你的消息监听方法
    @KafkaListener(id = "config_id", topicPartitions = @TopicPartition(topic = "#{@kafkaConstants.getConfig()}",
            partitionOffsets = {@PartitionOffset(partition = "0-11", initialOffset = "0")}),
            containerFactory = "configKafkaListenerContainerFactory1")
    public void handleMessage(String message, Acknowledgment ack) {
        // 业务逻辑处理
        ack.acknowledge(); // 若无需保留偏移量，可注释此句
    }
}

方案二：每次重启使用唯一消费组ID #

确保每次服务重启时生成全新的消费组ID，使Kafka认为是新消费组，从而触发AUTO_OFFSET_RESET_CONFIG=earliest从最开始消费。

修改consumerFactory1中的消费组配置：

config.put(ConsumerConfig.GROUP_ID_CONFIG, "rasbrandmapper-" + UUID.randomUUID());

注意：此方案会在Kafka中积累大量无用消费组记录，需定期清理。

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