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

Question

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

我知道当我得到一条记录时，我可以做processorContext。分区（）和处理器上下文。topic（）获取正在处理的当前记录的分区/主题信息。但我不是在找那个。

我正在寻找一个等效的KafkaConsumer.assigment在kafka流侧。

我也尝试了以下代码，但我得到大小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（）获取（该应用程序实例的）所有本地流线程的ThreadMetadataThreadMetadata包含该流线程上所有活动和备用任务的TaskMetadataTaskMetadata有一个方法topicPartitions（）来获取输入主题分区。

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