1 场景描述
在Java中，使用Kafka时，如果你想从头开始消费（始终从最早的消息开始消费），你需要设置Kafka消费者的auto.offset.reset属性为earliest。这样，当消费者没有为分区找到存在的偏移量时，它会从分区的开始处开始消费消息。
但我使用的是旧的分区，已经存在分区的offset，则不会从头消费。
本博客将从重置分区offset下手， 在不更换消费组的情况设置topic永久从头消费。
2 实现思路
常规办法，随机消费组，每次测试时候生成一个新的消费组。
重置offset，遍历该主题改消费组的分区信息，一一重置为0，再去消费数据，就实现了topic永久从头消费。
3 实现核心代码
// 获取分区信息
List<PartitionInfo> partitions = consumer.partitionsFor("test_topic");
// 因为 seek（） 是懒依赖 需要虚拟调用一下poll（）方法
ConsumerRecords<String, String> qq = consumer.poll(100);
for (PartitionInfo partition : partitions) {
TopicPartition topicPartition = new TopicPartition(partition.topic(), partition.partition());
// 指定重置分区的offset为0
consumer.seek(topicPartition, 0);
}
//提交offset信息
consumer.commitSync()
温馨提示：因为 seek（） 是懒依赖 ，所以需要虚拟调用一下poll（）方法。不然会报错,如下。
Exception in thread "main" java.lang.IllegalStateException: No current assignment for partition test_topic-0
at org.apache.kafka.clients.consumer.internals.SubscriptionState.assignedState(SubscriptionState.java:356)
at org.apache.kafka.clients.consumer.internals.SubscriptionState.seekUnvalidated(SubscriptionState.java:373)
at org.apache.kafka.clients.consumer.KafkaConsumer.seek(KafkaConsumer.java:1569)
at qiyun_test_kafkaConsumer.main(qiyun_test_kafkaConsumer.java:40)
4 kafka永久重头消费demo
import org.apache.kafka.clients.consumer.KafkaConsumer;
import org.apache.kafka.clients.consumer.ConsumerRecords;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.common.PartitionInfo;
import org.apache.kafka.common.TopicPartition;
import java.util.Arrays;
import java.util.List;
import java.util.Properties;
public class qiyun_test_kafkaConsumer {
public static void main(String[] args) {
Properties props = new Properties();
props.put("bootstrap.servers", "kafak01:9092");
props.put("group.id", "test1");
props.put("enable.auto.commit", "true");
props.put("auto.commit.interval.ms", "1000");
props.put("key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
props.put("value.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
props.put("auto.offset.reset", "earliest"); // 设置从最早的位置开始消费
KafkaConsumer<String, String> consumer = new KafkaConsumer<>(props);
consumer.subscribe(Arrays.asList("test_topic"));
// 获取分区信息
List<PartitionInfo> partitions = consumer.partitionsFor("test_topic");
// 因为 seek（） 是懒依赖 需要虚拟调用一下poll（）方法
ConsumerRecords<String, String> qq = consumer.poll(100);
for (PartitionInfo partition : partitions) {
TopicPartition topicPartition = new TopicPartition(partition.topic(), partition.partition());
// 指定重置分区的offset为0
consumer.seek(topicPartition, 0);
}
//提交offset信息
consumer.commitSync();
while (true) {
ConsumerRecords<String, String> records = consumer.poll(100);
for (ConsumerRecord<String, String> record : records) {
System.out.printf("offset = %d, key = %s, value = %s\n", record.offset(), record.key(), record.value());
}
}
}
}
4 kafka shell 消费命令
./kafka-console-consumer.sh --bootstrap-server kafka01:19092 --topic test_topic --from-beginning
