如何以编程方式获取 Java 中的 Kafka Cluster 和 Broker 信息?

How to programmatically get Kafka Cluster and Broker information in Java?

我希望以编程方式从 worker 本身获取我的 kafka 集群上的所有活动代理。 这个想法是创建一个健康调度程序,它将检查并 return 活跃的经纪人地址。

在每个worker中,我在设置消费者配置的时候设置了集群地址

props.put(onsumerConfig.BOOTSTRAP_SERVERS_CONFIG, myServerAddress);

但是,这不会告诉我目前哪些经纪商处于活跃状态。

我进行了一些搜索,但找不到任何获取此信息的方法。可能吗?

您可以使用 AdminClientdescribeCluster() 检索集群中 Kafka 代理的所有详细信息:

Get information about the nodes in the cluster.

Parameters: options - The options to use when getting information about the cluster.

Returns: The DescribeClusterResult.

// Create AdminClient
Properties props = new Properties();
props.load(new FileInputStream("ac.properties"));
AdminClient adminClient = KafkaAdminClient.create(props);

// Get brokers' details 
DescribeClusterResult describeClusterResult = adminClient.describeCluster();
List<Node> brokers = new ArrayList<>(describeClusterResult.nodes().get());
for (Node broker : brokers) {
    System.out.println("Host=" + broker.host() + ", Port=" + broker.port());
}