zoukankan      html  css  js  c++  java
  • 查询订阅某topic的所有consumer group(Java API)

    在网上碰到的问题,想了下使用现有的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();
            }
        }  
  • 相关阅读:
    周记(2015-11-30 -- 2015-12-05)
    周记(2015-11-22 -- 2015-11-27)
    周记(2015-11-15 -- 2015-11-20)
    周记(2015-11-01 -- 2015-11-06)
    设备与主机的攻击日志类型分析总结
    OWASP十大攻击类型详解
    乌云TOP 10 简单介绍
    《启示录》读书笔记三
    百度地图和定位
    获取Android studio的SHA1值
  • 原文地址:https://www.cnblogs.com/huxi2b/p/7831787.html
Copyright © 2011-2022 走看看