java - 获取 Kafka AdminClient.describeLogDirs 调用的代理列表

Question

AdminClient 的 API 包含方法 describeLogDirs两个签名都需要“经纪人列表”。引号中的表达式正是 javadoc 所描述的内容 - 不多了。

我必须提供什么作为该函数的参数？我的第一个猜测是从 0 或 1 开始到复制因子的序列，但后来我在应用程序的日志中看到了以下(不相关的)行，这表明我可能必须期待类似 1001 的内容。

13:47:11.931 [main] DEBUG o.a.k.c.consumer.internals.Fetcher - [Consumer clientId=myInstance-1725351556, groupId=] Sending READ_UNCOMMITTED IncrementalFetchRequest(toSend=(), toForget=(), implied=(myexample-0)) to broker example.com:9092 (id: 1001 rack: null)

我通过硬编码 1001 进行一次运行验证了这一点:

List<Integer> brokers = Collections.singletonList(1001);
DescribeLogDirsResult result = adminClient.describeLogDirs(brokers);

现在:如何使用我的 kafka 客户端的实时值填充变量 brokers ？ (Producer 和 adminClient 实例在我的代码中可用)？我无法找到此调用的任何示例代码。

更新(解决方案):这是我现在使用的最终代码行:

List<Integer> brokers = adminClient.describeCluster().nodes().get().stream().map(Node::id).collect(Collectors.toList());

Answer 1

您可以使用describeCluster()为了获取特定于节点的详细信息，如主机、端口和 ID。

public abstract DescribeClusterResult describeCluster​(DescribeClusterOptions options)

Get information about the nodes in the cluster.

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

Returns: The DescribeClusterResult.

示例:

Properties props = new Properties();
props.put("bootstrap.servers", "localhost:9092,localhost:9093");

AdminClient adminClient = AdminClient.create(props);
DescribeClusterResult describeClusterResult = adminClient.describeCluster();
Collection<Node> clusterDetails = describeClusterResult.nodes().get();