关于kafka-consumer-offset位移问题

[atguigu@hadoop102 kafka]$ bin/kafka-topics.sh --bootstrap-server hadoop102:9092 --create --topic atguigu --partitions 2 --replication-factor 2

[atguigu@hadoop102 kafka] $ bin/kafka-console-producer.sh --topic atguigu --bootstrap-server hadoop102:9092

[atguigu@hadoop104 kafka]$ bin/kafka-console-consumer.sh bootstrap-server hadoop102:9092--topic atguigu --group test

[atguigu@hadoop102 kafka]$ bin/kafka-console-consumer.sh --topic _consumer_offsets --bootstrap-server hadoop102:9092 --consumer.config config/consumer.properties --formatter "kafka.coordinator.group.GroupMetadataManager\$OffsetsMessageFormatter" --from-beginning

//配置是否是自动提交
properties.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG,true);
//提交时间间隔，单位是ms
properties.put(ConsumerConfig.AUTO_COMNIT_INTERVAL_NS_CONFI6,1000);

//手动提交属性配置
properties.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG ,false);
//消费代码逻辑
XXX
XXX
XXX
//手动提交代码(处理完数据以后，这里为了方便，只展示关键代码)
//手动提交offset
kafkaConsumer.commitsync();

//手动提交属性配置
properties.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG ,false);
//消费代码逻辑
XXX
XXX
XXX
//手动提交代码(处理完数据以后，这里为了方便，只展示关键代码)
//手动提交offset
kafkaConsumer.commitAsync();

//1创建消费者
KafkaConsumer<String，String> kafkaConsumer = new KafkaConsumer<>(properties);
// 2订阅主题
ArrayList<String> topics = new ArrayList<>(;topics.add( "first");
kafkaConsumer.subscribe(topics);
 
//指定位置进行消费
set<TopicPartition> assignment = kafkaConsumer.assignment();//获取所有分区信息
//保证分区分配方案已经制定完毕，因为由于leader消费者制定分配方案会消耗一定时间，有可能此时获取不到分区信息,所以加一层分区空间判断
while (assignment.size() == 0){
    //促使获取的分区数量不为0
    kafkaConsumer.poll(Duration.ofSeconds(1));
    assignment = kafkaConsumer.assignment();
}
 
//遍历所有分区，指定消费的offset
for (TopicPartition topicPartition : assignment) {
    kafkaConsumer.seek(topicPartition, 100);
}
 
// 3消费数据
while (true){

//1创建消费者
KafkaConsumer<String，String> kafkaConsumer = new KafkaConsumer<>(properties);
// 2订阅主题
ArrayList<String> topics = new ArrayList<>(;topics.add( "first");
kafkaConsumer.subscribe(topics);
 
//指定位置进行消费
set<TopicPartition> assignment = kafkaConsumer.assignment();//获取所有分区信息
//保证分区分配方案已经制定完毕，因为由于leader消费者制定分配方案会消耗一定时间，有可能此时获取不到分区信息,所以加一层分区空间判断
while (assignment.size() == 0){
    //促使获取的分区数量不为0
    kafkaConsumer.poll(Duration.ofSeconds(1));
    assignment = kafkaConsumer.assignment();
}
//希望把时间转换为对应的offset
HashMap<TopicPartition，Long> topicPartitionLongHashMap = new HashMap<>();
//封装对应集合
for (TopicPartition topicPartition : assignment) {
    //希望获取当前系统时间一天前的数据。
    topicPartitionLongHashMap.put(topicPartition, System.currentTimeMillis() - 1 * 24 * 3600 * 1000);
}
Nap<TopicPartition，OffsetAnd imestamp> topioPartitionffsetAndrtimestampMep = karfiaConsumer.offsetsForTines(topicPartitionL ongHashiap);
 
 
//遍历所有分区，指定消费的offset
//指定消费的offset
for (TopicPartition topicPartition : assignment) {
    OffsetAndTimestamp offsetAndTimestamp = topicPartition0ffsetAndTimestampHap.get(topicPartition);
    kafkaConsumer.seek(topicPartition,offsetAndTimestamp.offset());
}
 
// 3消费数据
while (true){