java - Apache Kafka UnresolvedAddressException 发送消息时

Question

在我的 java web 应用程序中，我正在尝试发送消息 kafka (7.2 版)。我正在使用动物园管理员连接到我的经纪人，如下所示:

props = new Properties();
props.put("zk.connect", <zk-ip1>:<2181>,<zk-ip3>:<2181>,<zk-ip3>:<2181>);
props.put("zk.connectiontimeout.ms", "1000000");
props.put("producer.type", "async");

Producer<String, Message> producer = new kafka.javaapi.producer.Producer<String, Message>(new ProducerConfig(props));

ProducerData<String, Message> producerData = new ProducerData<String, Message>("someTopic", messageTosend);

producer.send(producerData);

当我创建生产者时，日志看起来不错:

    27 Jan 2014 07:38:25  INFO ZooKeeper - Initiating client connection, connectString=<zk-ip1>:2181,<zk-ip2>:2181,<zk-ip3>:2181 sessionTimeout=6000 watcher=org.I0Itec.zkclient.ZkClient@2a2fd6f7
    27 Jan 2014 07:38:25  INFO ClientCnxn - Opening socket connection to server /<zk-ip2>:2181
    27 Jan 2014 07:38:25  INFO ClientCnxn - Socket connection established to <zp-ipk>/<zk-ip2>:2181, initiating session
    27 Jan 2014 07:38:25  INFO ClientCnxn - Session establishment complete on server <zk-ip2>/<zp-ip2>:2181, sessionid = 0x143b94e6434000c, negotiated timeout = 6000
    27 Jan 2014 07:38:25  INFO ZkClient - zookeeper state changed (SyncConnected)
    27 Jan 2014 07:38:25  INFO ProducerPool - Creating async producer for broker id = 326278963 at <broker-ip2>:9092
    27 Jan 2014 07:38:25  INFO ProducerPool - Creating async producer for broker id = 936343853 at <broker-ip1>:9092

但是在尝试发送消息时我得到:

27 Jan 2014 07:39:07  INFO SyncProducer - Disconnecting from <broker-ip2>:9092
27 Jan 2014 07:39:07 ERROR SyncProducer - Connection attempt to <broker-ip2>:9092 failed, next attempt in 100 ms
java.nio.channels.UnresolvedAddressException
        at sun.nio.ch.Net.checkAddress(Net.java:127)
        at sun.nio.ch.SocketChannelImpl.connect(SocketChannelImpl.java:640)
        at kafka.producer.SyncProducer.connect(SyncProducer.scala:173)
        at kafka.producer.SyncProducer.getOrMakeConnection(SyncProducer.scala:196)
        at kafka.producer.SyncProducer.send(SyncProducer.scala:92)
        at kafka.producer.SyncProducer.multiSend(SyncProducer.scala:135)
        at kafka.producer.async.DefaultEventHandler.send(DefaultEventHandler.scala:58)
        at kafka.producer.async.DefaultEventHandler.handle(DefaultEventHandler.scala:44)
        at kafka.producer.async.ProducerSendThread.tryToHandle(ProducerSendThread.scala:116)
        at kafka.producer.async.ProducerSendThread$$anonfun$processEvents$3.apply(ProducerSendThread.scala:95)
        at kafka.producer.async.ProducerSendThread$$anonfun$processEvents$3.apply(ProducerSendThread.scala:71)
        at scala.collection.immutable.Stream.foreach(Stream.scala:526)
        at kafka.producer.async.ProducerSendThread.processEvents(ProducerSendThread.scala:70)
        at kafka.producer.async.ProducerSendThread.run(ProducerSendThread.scala:41)

有什么想法吗？

Answer 1

问题解决了。看来我们在 DNS 配置方面遇到了一些内部 IT 问题(与我们的 zk 和代理相关)。