Kafka使用admin API提交和最后一次偏移

Question

我正在使用admin client API查询kafka broker，以使用以下代码获取CONSUMER_GROUP的提交偏移量：

Map<TopicPartition, OffsetAndMetadata> offsets =
                admin.listConsumerGroupOffsets(CONSUMER_GROUP)
                        .partitionsToOffsetAndMetadata().get();

上述代码将触发对特殊创建的_consumer_offsets主题的查询，以获取主题的每个分区（consumer_GROUP负责的分区）的提交偏移量。

另一方面，我使用下面的代码检索CONSUMER_GROUP的每个主题分区的最新（结束）偏移量

for(TopicPartition tp: offsets.keySet()) {
requestLatestOffsets.put(tp, OffsetSpec.latest());
}

Map<TopicPartition, ListOffsetsResult.ListOffsetsResultInfo> latestOffsets =
admin.listOffsets(requestLatestOffsets).all().get();

for (Map.Entry<TopicPartition, OffsetAndMetadata> e: offsets.entrySet()) {
long latestOffset = latestOffsets.get(e.getKey()).offset();

我的问题是，已提交和最新的偏移量因此是从两个不同的主题查询/请求的。已提交的偏移量是从__consumer_offsets主题请求的，最新的（结束）偏移量是从CONSUMER_GROUP的实际主题请求的。

（1） 上述关于请求提交和最新补偿的描述是否准确？

②是否可以直接查询__consumer_offsets题目？

谢谢你。

Answer 1

>

是的，你的理解是正确的。提交的偏移量存储在\uuuu consumer\u offset主题中，而您需要查询特定分区以获取其末端偏移量。

是的_消费者_补偿是一个常规话题，如果你想直接消费它。通过提供的API检索数据通常更容易，但如果您对其内容感兴趣，可以使用它。如果想了解如何反序列化数据，请查看控制台格式化程序。