码迷,mamicode.com
首页 > 编程语言 > 详细

查询订阅某topic的所有consumer group(Java API)

时间:2017-11-14 14:26:00      阅读:658      评论:0      收藏:0      [点我收藏+]

标签:ext   sum   kafka   adl   final   mave   instance   fse   bin   

在网上碰到的问题,想了下使用现有的API还是可以实现的。

首先,需要引入Kafka服务器端代码,比如加入Kafka 1.0.0依赖:

Maven

<dependency>
<groupId>org.apache.kafka</groupId>
<artifactId>kafka_2.12</artifactId>
<version>1.0.0</version>
</dependency>

Gradle

compile group: ‘org.apache.kafka‘, name: ‘kafka_2.12‘, version: ‘1.0.0‘

然后编写获取订阅某topic的所有group的方法,代码如下:

/**
     * get all subscribing consumer group names for a given topic
     * @param brokerListUrl localhost:9092 for instance
     * @param topic         topic name
     * @return
     */
    public static Set<String> getAllGroupsForTopic(String brokerListUrl, String topic) {
        AdminClient client = AdminClient.createSimplePlaintext(brokerListUrl);

        try {
            List<GroupOverview> allGroups = scala.collection.JavaConversions.seqAsJavaList(client.listAllGroupsFlattened().toSeq());
            Set<String> groups = new HashSet<>();
            for (GroupOverview overview: allGroups) {
                String groupID = overview.groupId();
                Map<TopicPartition, Object> offsets = scala.collection.JavaConversions.mapAsJavaMap(client.listGroupOffsets(groupID));
                Set<TopicPartition> partitions = offsets.keySet();
                for (TopicPartition tp: partitions) {
                    if (tp.topic().equals(topic)) {
                        groups.add(groupID);
                    }
                }
            }
            return groups;
        } finally {
            client.close();
        }
    }  

查询订阅某topic的所有consumer group(Java API)

标签:ext   sum   kafka   adl   final   mave   instance   fse   bin   

原文地址:http://www.cnblogs.com/huxi2b/p/7831787.html

(0)
(0)
   
举报
评论 一句话评论(0
登录后才能评论!
© 2014 mamicode.com 版权所有  联系我们:gaon5@hotmail.com
迷上了代码!