Kafka新手求助：是否有API可实现kafka-consumer-groups.sh/bat的列表功能？

嘿，作为Kafka新手能想到用API替代命令行工具，这点特别好——毕竟在自动化脚本或者应用程序里，直接调用API比执行shell命令灵活多了。

其实kafka-consumer-groups.sh这个命令行工具底层就是用Kafka官方的AdminClient API实现的，所以你完全可以用它来获取消费组列表，效果和命令行输出一模一样。

具体实现方式（以Java为例，官方推荐） #

首先你需要引入Kafka客户端依赖（比如Maven）：

<dependency>
    <groupId>org.apache.kafka</groupId>
    <artifactId>kafka-clients</artifactId>
    <version>你的Kafka集群版本</version>
</dependency>

然后写代码调用AdminClient的listConsumerGroups()方法：

import org.apache.kafka.clients.admin.AdminClient;
import org.apache.kafka.clients.admin.AdminClientConfig;
import org.apache.kafka.clients.admin.ListConsumerGroupsResult;
import org.apache.kafka.clients.admin.ConsumerGroupListing;

import java.util.Properties;
import java.util.concurrent.ExecutionException;

public class ConsumerGroupLister {
    public static void main(String[] args) {
        // 配置AdminClient，和命令行的--bootstrap-server对应
        Properties adminProps = new Properties();
        adminProps.put(AdminClientConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9091");

        // 用try-with-resources自动关闭AdminClient
        try (AdminClient adminClient = AdminClient.create(adminProps)) {
            // 调用API获取消费组列表
            ListConsumerGroupsResult groupsResult = adminClient.listConsumerGroups();
            // 因为是异步操作，用get()阻塞获取结果
            for (ConsumerGroupListing groupListing : groupsResult.listings().get()) {
                // 输出消费组ID，和命令行的输出一致
                System.out.println(groupListing.groupId());
            }
        } catch (InterruptedException | ExecutionException e) {
            // 处理异常，比如连接失败、权限不足等
            e.printStackTrace();
        }
    }
}

其他语言的实现（比如Python） #

如果你用Python开发，可以用confluent-kafka库的AdminClient，效果一样：

from confluent_kafka.admin import AdminClient

# 初始化AdminClient
admin_client = AdminClient({"bootstrap.servers": "localhost:9091"})

# 获取消费组列表
group_list = admin_client.list_consumer_groups().result()

# 遍历输出消费组ID
for group in group_list:
    print(group[0])

注意事项 #

• 确保你的客户端版本和Kafka集群版本尽量匹配，避免兼容性问题
• 配置的bootstrap.servers要和命令行里的一致，保证能连接到集群
• 运行代码的账号需要有足够的权限（和执行kafka-consumer-groups.sh的账号权限相同）

内容的提问来源于stack exchange，提问作者deepak PATRA