如何确定主题已被Kafka Stream应用程序从第一个偏移量到最后一个偏移量从Java应用程序完全阅读

Question

我在Kafka Streams中需要一些帮助。我已经启动了一个Kafka流应用程序，它从第一个偏移量流式传输一个主题。主题的数据非常庞大，所以我想在我的应用程序中实现一种机制，使用Kafka流，这样我就可以在主题被完全读取到最后一个偏移量时得到通知。

我已经阅读了Kafka Streams 2.8.0 api，我找到了一个api方法i-e allLocalStore分区延迟，它将存储名称映射返回到另一个分区映射，其中包含每个分区的所有滞后信息。此方法返回此Streams本地的所有存储分区（活动或备用）的滞后信息。在上面的情况下，当我有一个节点运行该流应用程序时，此方法对我非常有用。

但是在我的例子中，系统是分布式的，应用程序节点是3个，主题分区是10个，这意味着每个节点至少有3个分区供主题读取。

我在这里需要帮助。我如何实现此功能，当主题从分区 0 完全读取到分区 9 时，我可以收到通知。请注意，到目前为止，我无法在此处使用数据库。

实现目标的其他方法也是受欢迎的。谢谢你。

Answer 1

我能够从管理员客户端 API 获得滞后信息。下面的代码结果针对给定流应用程序 i-e applicationId 读取的主题结束每个分区的偏移量和当前偏移量。

AdminClient adminClient = AdminClient.create(kafkaProperties);
ListConsumerGroupOffsetsResult listConsumerGroupOffsetsResult = adminClient.listConsumerGroupOffsets(applicationId);

// Current offsets.
Map<TopicPartition, OffsetAndMetadata> topicPartitionOffsetAndMetadataMap = listConsumerGroupOffsetsResult.partitionsToOffsetAndMetadata().get();

// all topic partitions.
Set<TopicPartition> topicPartitions = topicPartitionOffsetAndMetadataMap.keySet();
// list of end offsets for each partitions.
ListOffsetsResult listOffsetsResult = adminClient.listOffsets(topicPartitions.stream()
    .collect(Collectors.toMap(Function.identity(), tp -> OffsetSpec.latest())));