如何获取每个分区的当前最新偏移量，然后只消耗该偏移量？答案

【问题标题】：How to get current latest offset for each partition, then only consume up to that offset?如何获取每个分区的当前最新偏移量，然后只消耗该偏移量？
【发布时间】：2019-12-04 11:53:42
【问题描述】：

我正在尝试检查接收大量数据的主题中是否缺少键。由于这项工作是一种按需运行的事情，因此它需要一些标准来知道它何时搜索了它关心的所有记录。我们确定这将是作业启动时每个分区的最新偏移量。

我的问题首先是如何获取主题的所有分区信息而不实际使用它（我需要使用它为每个分区创建单独的消费者，以跟踪他们的偏移量与最大偏移量）。

其次，当消费者看到达到最大偏移量后，如何停止它。

编辑：我找到了一种获取分区的方法，即为单个消费者订阅主题，做一个虚拟的poll，然后使用partitionsFor(...)。不确定这是否是“推荐”的方式。

【问题讨论】：

当你消费时，你会收到新消息……在轮询循环中，你可以检查迭代器是否为空，然后你就到了终点

标签： java apache-kafka kafka-consumer-api

【解决方案1】：

您可以使用 consumer.partitionsFor 和 consumer.endOffsets 获取分区和最后一个偏移量

partitionsFor

 /*Get metadata about the partitions for a given topic. This method will issue a remote call to the server if it does  not already have any metadata about the given topic.*/ 
    public java.util.List<PartitionInfo> partitionsFor(java.lang.String topic)

endOffsets

/*Get the last offset for the given partitions. The last offset of a partition is the offset of the upcoming message, i.e. the offset of the last available message + 1*/
public java.util.Map<TopicPartition,java.lang.Long> endOffsets(java.util.Collection<TopicPartition> partitions)

下面是示例代码

Properties consumerProperties = new Properties();
consumerProperties.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092");
consumerProperties.put(ConsumerConfig.GROUP_ID_CONFIG, "consumerid");
Consumer<String, byte[]> consumer = new KafkaConsumer<>(consumerProperties);    
List<PartitionInfo> parts = consumer.partitionsFor(topic);
consumer.assign(partitions);
Map<TopicPartition, Long> offsets = consumer.endOffsets(partitions);
for (TopicPartition tp : offsets.keySet()) {
    OffsetAndMetadata commitOffset = consumer.committed(new 
    TopicPartition(tp.topic(), tp.partition()));
    //Consumer offset for partition tp
    long offset=offsets.get(tp);
    //Consumed committed offset
    long consumedOffset=commitOffset.offset();
}

【讨论】：