Apache Kafka控制台正常但API生产者无法发送消息，求排查方案

看起来你遇到的核心问题是Kafka客户端版本不兼容加上发送逻辑的异步特性没处理到位，咱们一步步拆解排查：

一、先解决最致命的依赖版本冲突 #

你的Scala项目里同时引入了kafka-clients 0.8.2.1和kafka 0.10.2.0，这两个版本的客户端协议差异极大——0.8.x到0.10.x的Kafka客户端和服务端通信协议有重大更新，混搭版本必然导致兼容性问题。Java项目里用的kafka-clients 0.8.2.1如果和你的Kafka集群版本不匹配（比如集群是0.10.x+），也会连不上。

修复方案： #

• 统一依赖版本：确保客户端版本和你的Kafka集群版本完全一致（至少大版本匹配）。比如你的集群是0.10.2.0，那所有依赖都用这个版本：
  Scala项目的build.sbt只保留：

  libraryDependencies += "org.apache.kafka" %% "kafka" % "0.10.2.0"
  

  Java项目的build.sbt改成：

  libraryDependencies += "org.apache.kafka" % "kafka-clients" % "0.10.2.0"
  

• 不要重复引入依赖：kafka artifact已经包含了kafka-clients，重复引入会导致Jar包冲突，删掉多余的依赖项。

二、修复生产者发送逻辑的问题 #

你原来的代码里producer.send(data)是异步操作，调用后立刻producer.close()，消息根本没来得及提交到Kafka就被关闭了。后来用get(30, TimeUnit.SECONDS)等待发现超时，说明消息连发送请求都没成功送达Broker。

正确的发送逻辑： #

要么同步等待发送完成，要么flush后再关闭：

// 方式1：同步等待发送结果
producer.send(new ProducerRecord<>(topic, "1", "2")).get();
producer.close();

// 方式2：flush确保消息被提交到Broker
producer.send(new ProducerRecord<>(topic, "1", "2"));
producer.flush(); // 强制把缓冲区的消息发送出去
producer.close();

三、进一步调试排查步骤 #

如果版本统一后还是超时，按以下顺序排查：

1. 检查Broker的监听配置
   控制台命令能正常运行，可能是因为控制台用的是localhost/内部地址，而API用的是外部地址，但Broker没配置对外监听。打开Kafka的server.properties：

   • 确保listeners包含外部可访问的地址，比如listeners=PLAINTEXT://0.0.0.0:9092（允许所有网卡连接）
   • 确认advertised.listeners配置的是客户端实际能访问的地址，必须和你代码里的<broker>:9092完全一致

2. 验证网络连通性
   虽然修改Broker地址会报错，但还是要确认端口是否真的能正常通信：

   # 用telnet测试
   telnet <broker> 9092
   # 或者用nc（更直观）
   nc -zv <broker> 9092
   

   如果连接失败，排查防火墙/安全组是否开放了9092端口，或者Broker所在机器的网络路由是否正常。

3. 开启客户端DEBUG日志
   开启Kafka客户端的DEBUG日志，能看到客户端和Broker的交互细节（比如是否能获取Topic元数据、是否发送了请求等）。在Java代码里添加：

   import org.apache.log4j.Logger;
   import org.apache.log4j.Level;
   
   public class ProducerExample {
       static {
           // 开启Kafka客户端的DEBUG日志
           Logger.getLogger("org.apache.kafka").setLevel(Level.DEBUG);
       }
   
       public static void main(String[] args) {
           // ... 你的代码
       }
   }
   

   日志里如果出现Failed to update metadata之类的错误，说明客户端连不上Broker获取元数据，大概率是版本或监听配置的问题。

4. 检查Topic状态
   确保testTopic存在且状态正常：

   kafka-topics.sh --describe --topic test --bootstrap-server <broker>:9092
   

   重点看Leader列是否有可用的Broker ID，Isr列的副本是否在线。如果Leader不可用，生产者会无法发送消息。

5. 调整acks配置测试
   你Java代码里用了acks=all，这需要所有副本都确认消息，如果你的Topic副本数大于可用Broker数（比如副本数是3但只有1个Broker在运行），会导致超时。可以先临时改成acks=1测试：

   props.put("acks", "1");
   

   如果能发送成功，说明是副本配置的问题，需要调整Topic的副本数或者确保足够的Broker在线。

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