如何获取时间戳指定的kafka偏移数据

Question

当我尝试运行时抛出空指针错误时，我试图根据时间戳获取 Kafka 主题的偏移量，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);任何帮助将不胜感激。

Answer 1

要找到与时间戳对应的偏移量，您需要使用 方法offsetsForTimes()。

例如，这将打印mytopic对应于 1 秒前的分区 0 的偏移量：

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);
}

这将显示如下内容：

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

Answer 2

您可以Admin.listOffsets使用OffsetSpec.forTimestamp：

Map<TopicPartition, OffsetSpec> topicOffsetSpecs = new HashMap<>();

TopicPartition topicPartition = new TopicPartition("topic1", 0);

OffsetSpec offsetSpec = OffsetSpec.forTimestamp(timestamp);

topicOffsetSpecs.put(topicPartition, offsetSpec);

admin.listOffsets(topicOffsetSpecs).all().get(); // Info for given timestamp