在Kafka中，如何找到给定开始日期和结束日期(或时间戳)之间的所有分区的偏移量，并重放消息

import org.apache.kafka.clients.consumer.*;
import org.apache.kafka.common.*;
import java.time.*;
import java.util.*;

public class KafkaReplayMessages {
    public static void main(String[] args) {
        String bootstrapServers = "kafka-broker1:9092,kafka-broker2:9092";
        String topic = "your-topic";
        String groupId = "your-consumer-group";
        LocalDateTime startDateTime = LocalDateTime.of(2022, 1, 1, 0, 0, 0);
        LocalDateTime endDateTime = LocalDateTime.of(2022, 1, 2, 0, 0, 0);

        Properties props = new Properties();
        props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, bootstrapServers);
        props.put(ConsumerConfig.GROUP_ID_CONFIG, groupId);
        props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringDeserializer");
        props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringDeserializer");

        KafkaConsumer<String, String> consumer = new KafkaConsumer<>(props);

        // 获取指定主题的所有分区列表
        List<PartitionInfo> partitions = consumer.partitionsFor(topic);

        // 遍历分区列表
        for (PartitionInfo partition : partitions) {
            TopicPartition topicPartition = new TopicPartition(partition.topic(), partition.partition());

            // 设置消费者的偏移量为给定开始日期对应的偏移量
            long startOffset = getOffsetForDateTime(consumer, topicPartition, startDateTime);
            consumer.seek(topicPartition, startOffset);

            // 开始消费消息
            while (true) {
                ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(100));

                for (ConsumerRecord<String, String> record : records) {
                    // 处理消息记录
                    System.out.println(record.value());
                }

                // 判断是否达到给定结束日期对应的偏移量
                long currentOffset = consumer.position(topicPartition);
                long endOffset = getOffsetForDateTime(consumer, topicPartition, endDateTime);
                if (currentOffset >= endOffset) {
                    break;
                }
            }
        }

        consumer.close();
    }

    private static long getOffsetForDateTime(KafkaConsumer<String, String> consumer, TopicPartition topicPartition, LocalDateTime dateTime) {
        Map<TopicPartition, Long> timestampsToSearch = new HashMap<>();
        timestampsToSearch.put(topicPartition, dateTime.atZone(ZoneId.systemDefault()).toInstant().toEpochMilli());

        Map<TopicPartition, OffsetAndTimestamp> offsetAndTimestampMap = consumer.offsetsForTimes(timestampsToSearch);
        OffsetAndTimestamp offsetAndTimestamp = offsetAndTimestampMap.get(topicPartition);

        if (offsetAndTimestamp != null) {
            return offsetAndTimestamp.offset();
        } else {
            // 如果找不到给定日期对应的偏移量，则返回最早的偏移量
            return consumer.beginningOffsets(Collections.singleton(topicPartition)).get(topicPartition);
        }
    }
}