Spring Boot+Java项目中如何查询Kafka连接的Broker地址

一、通过应用日志查看 #

Spring Kafka默认会输出连接相关日志，只需将日志级别调整为DEBUG或TRACE，就能直接看到生产者/消费者与Broker建立连接的详细信息。

• 配置日志级别：在application.yml中添加：

logging:
  level:
    org.apache.kafka.clients.producer: DEBUG
    org.apache.kafka.clients.consumer: DEBUG
    org.springframework.kafka: DEBUG

或application.properties格式：

logging.level.org.apache.kafka.clients.producer=DEBUG
logging.level.org.apache.kafka.clients.consumer=DEBUG
logging.level.org.springframework.kafka=DEBUG

• 启动应用后，日志中会出现类似Connected to node X的条目，其中X是Broker的ID，同时会附带Broker的主机地址和端口信息。

二、通过代码获取连接信息 #

1. 生产者端获取连接的Broker #

通过KafkaTemplate获取底层生产者，进而查询主题分区的Leader Broker（生产者实际连接的是对应分区的Leader节点）：

@Autowired
private KafkaTemplate<String, MyRecord> kafkaTemplate;

public void getProducerConnectedBroker() {
    Producer<String, MyRecord> producer = kafkaTemplate.getProducerFactory().createProducer();
    // 查询指定主题的分区信息
    List<PartitionInfo> partitionInfos = producer.partitionsFor("mytest");
    for (PartitionInfo info : partitionInfos) {
        Node leaderNode = info.leader();
        System.out.printf("生产者连接的Broker：ID=%d, Host=%s, Port=%d%n", 
                          leaderNode.id(), leaderNode.host(), leaderNode.port());
    }
    producer.close();
}

2. 消费者端获取连接的Broker #

修改@KafkaListener方法，注入Consumer对象来获取当前消费分区的Leader Broker信息：

@KafkaListener(topics = "mytest")
private void consume(ConsumerRecord<String, MyRecord> cr, Consumer<String, MyRecord> consumer) {
    // 根据当前消费的分区，找到对应的Leader Broker
    PartitionInfo partitionInfo = consumer.partitionsFor(cr.topic()).stream()
            .filter(info -> info.partition() == cr.partition())
            .findFirst()
            .orElse(null);
    if (partitionInfo != null) {
        Node leaderNode = partitionInfo.leader();
        System.out.printf("当前消费分区的Leader Broker：ID=%d, Host=%s, Port=%d%n", 
                          leaderNode.id(), leaderNode.host(), leaderNode.port());
    }
    // 原有消费逻辑
}

三、通过JMX监控查看 #

Kafka客户端默认暴露JMX指标，可通过JConsole或VisualVM连接应用进程，查看对应的MBean：

• 生产者：kafka.producer:type=producer-metrics,client-id=*，其中node.id、node.host、node.port属性会显示当前连接的Broker信息。
• 消费者：kafka.consumer:type=consumer-metrics,client-id=*，同样可通过node.id等属性获取连接的Broker详情。

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