如何通过Java配置让Kafka消费者从起始位置（offset 0）消费全量Topic数据？

1. 核心配置属性实现基础需求 #

要实现类似Shell命令bin/kafka-console-consumer.sh --bootstrap-server localhost:9092 --topic myjson --from-beginning的效果，你需要将ConsumerConfig.AUTO_OFFSET_RESET_CONFIG设置为"earliest"：

props.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest");

纠正你对配置值的误解： #

• earliest：当当前消费者组从未提交过该Topic的offset时，会从Topic的最起始位置（通常是offset 0，除非旧数据被清理策略删除）开始消费所有记录。
• latest：当无已提交offset时，仅消费消费者启动后新产生的记录。

注意：如果你的消费者组已经提交过该Topic的offset，这个配置不会生效，Kafka会默认从提交的offset位置继续消费。

2. 强制从offset 0消费（不受已提交offset影响） #

你之前尝试的seekToBeginning代码未生效，是因为consumer.poll(Duration.ZERO)执行时，消费者还未完成分区分配，consumer.assignment()返回的是空集合。以下是两种可行的解决方法：

方法一：等待分区分配完成后再跳转 #

consumer.subscribe(Collections.singletonList("myjson"));
// 循环poll直到获取到分配的分区
while (consumer.assignment().isEmpty()) {
    consumer.poll(Duration.ofMillis(100));
}
// 跳转到所有分配分区的起始位置
consumer.seekToBeginning(consumer.assignment());

方法二：手动指定分区并强制跳转到offset 0 #

如果你明确目标Topic的分区信息，也可以直接指定分区并设置offset：

// 获取目标Topic的所有分区
List<TopicPartition> partitions = consumer.partitionsFor("myjson")
        .stream()
        .map(p -> new TopicPartition(p.topic(), p.partition()))
        .collect(Collectors.toList());
// 手动分配分区
consumer.assign(partitions);
// 逐个将分区offset设置为0
for (TopicPartition partition : partitions) {
    consumer.seek(partition, 0L);
}

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