阿帕奇Kafka获得特定主题的消费者列表

Question

顾名思义，有没有一种方法可以在java中获取特定主题的消费者列表？直到现在我才能得到这样的主题列表

    final ListTopicsResult listTopicsResult = adminClient.listTopics();
    KafkaFuture<Set<String>> kafkaFuture = listTopicsResult.names();
    Set<String> map = kafkaFuture.get();

但我还没有找到一种方法来获得每个主题的消费者名单

Answer 1

我知道这个话题有点老了，但我目前正在开发一个仪表板，需要列出相关消费者群体的话题。Katya的回答是可以的，但它仍然不会为没有活跃成员的消费者群体列出TopicPartitions。我偶然发现了这个解决方案，希望有所帮助:

Properties props = ...//here you put your properties
AdminClient kafkaClient = AdminClient.create(props);

Map<TopicPartition, OffsetAndMetadata> offsets = kafkaClient
    .listConsumerGroupOffsets("consumer_group_name")
    .partitionsToOffsetAndMetadata()
    .get();

主题偏移量映射中的分区可以针对特定主题进行解析。此列表可以重复整个组Id列表，从Katya的答案中。

Answer 2

消费者不受主题的约束。他们与消费者群体紧密相连

要获取某个主题的所有使用者，您必须首先列出所有组，然后在每个组中筛选出您的主题

它将以< code > admin client . listconsumergroups 开头，后跟< code > admin client . describeconsumergroups

这将为您提供一个包含“成员”的描述列表，您可以在其中找到主题

https://Kafka . Apache . org/20/javadoc/org/Apache/Kafka/clients/admin/consumergroupdescription . html

注意：有一些外部工具可以使这一过程更加简单，如Hortonworks SMMhttps://docs.hortonworks.com/HDPDocuments/SMM/SMM-1.2.0/monitoring-kafka-clusters/content/smm-monitoring-consumers.html

Answer 3

我最近正在为我的kafka客户端工具解决同样的问题。这并不容易，但我从代码中发现的唯一方法如下：

Properties props = ...//here you put your properties
AdminClient kafkaClient = AdminClient.create(props);

//Here you get all the consumer groups
List<String> groupIds = kafkaClient.listConsumerGroups().all().get().
                       stream().map(s -> s.groupId()).collect(Collectors.toList()); 

//Here you get all the descriptions for the groups
Map<String, ConsumerGroupDescription> groups = kafkaClient.
                                               describeConsumerGroups(groupIds).all().get();
for (final String groupId : groupIds) {
    ConsumerGroupDescription descr = groups.get(groupId);
    //find if any description is connected to the topic with topicName
    Optional<TopicPartition> tp = descr.members().stream().
                                  map(s -> s.assignment().topicPartitions()).
                                  flatMap(coll -> coll.stream()).
                                  filter(s -> s.topic().equals(topicName)).findAny();
            if (tp.isPresent()) {
                //you found the consumer, so collect the group id somewhere
            }
} 

这个API从2.0版开始提供。可能有更好的方法，但我找不到。你也可以在我的bitbucket上找到代码