How to get kafka offset data, specified on timesta

问题:

I've tried to get the offset from Kafka topic based on timestamp when I tried to run it was throwing null pointer error,

Map<TopicPartition, Long> timestampsToSearch = new HashMap<>();
              for (TopicPartition partition : partitions) {
                timestampsToSearch.put(partition,  startTimestamp);
              }
Map<TopicPartition, OffsetAndTimestamp> outOffsets = consumer.offsetsForTimes(timestampsToSearch);
              for (TopicPartition partition : partitions) {
                Long seekOffset = outOffsets.get(partition).offset();
consumer.seek(partition, seekOffset);

Any help will be appreciated.

回答1:

To find the offsets that correspond to a timestamp, you need to use the offsetsForTimes() method.

For example, this will print the offsets for partition 0 of mytopic that correspond to 1 second ago:

try (KafkaConsumer<String, String> consumer = new KafkaConsumer<>(configs);) {
    Map<TopicPartition, Long> timestamps = new HashMap<>();
    timestamps.put(new TopicPartition("mytopic", 0), System.currentTimeMillis()-1*1000);
    Map<TopicPartition, OffsetAndTimestamp> offsets = consumer.offsetsForTimes(timestamps);
    System.err.println(offsets);
}

That will display something like:

{offset-test-0=(timestamp=1561469319192, leaderEpoch=0, offset=100131)}