查找分配给 Kafka 流实例的分区

Question

我有一个订阅许多主题的 Kafka 流应用程序，每个主题都有许多分区。 当我创建应用程序拓扑并启动它时，我是否知道哪些主题的哪些分区分配给了我的应用程序的当前实例？我想知道这个独立于任何记录是否已被此实例处理。

我知道当我得到一条记录时，我可以通过processorContext.partition() 和processorContext.topic() 来获取当前正在处理的记录的分区/主题信息。但我不是在寻找那个。

我正在寻找与 kafka 流端的 KafkaConsumer.assigment 等效的东西。

我也尝试了以下代码，但我得到 s 的大小为 0。

<Prepare builder and sconfig>
kafkaStream = new KafkaStreams (builder, sconfig);
kafkaStream.start ();
Collection<StreamsMetadata> s = kafkaStream.allMetadata();
System.out.println("StreamsMetadata: size is " + s.size());
for (StreamsMetadata m : s) {
    Set<TopicPartition> tp = m.topicPartitions();
    System.out.println  ("TopicPartition: " + tp.toString());
}

Answer 1

更新答案（2020 年 11 月）：

当我创建应用拓扑并启动它时，我是否知道哪些主题的哪些分区分配给了我的应用的当前实例？

如果我理解正确，您可以按如下方式执行此操作。在您的应用程序实例中，使用KafkaStreams#localThreadsMetadata() 获取所有本地流线程（该应用程序实例的）的ThreadMetadata。 ThreadMetadata 包含 TaskMetadata 用于该流线程上的所有活动和备用任务。 TaskMetadata 有一个方法 topicPartitions() 来获取输入主题分区。

旧的、过时的答案：据我所知，Kafka Streams 中没有现有的 API 可以公开此信息。可以从 Kafka 消费者（由 Kafka Streams 使用）获取此信息，但不会在 Kafka Streams 中公开。