运行 kafka-console-producer.sh (0.8.1.1) 的异常答案

【问题标题】：Exception running kafka-console-producer.sh (0.8.1.1)运行 kafka-console-producer.sh (0.8.1.1) 的异常
【发布时间】：2014-11-24 22:53:55
【问题描述】：

我正在试用 kafka-console-producer.sh。我在控制台输入一些东西并按回车键，我得到大量的堆栈跟踪到控制台，就像这样。感谢任何指针。

启动生产者的命令是：

./bin/kafka-console-producer.sh --broker-list localhost:2181 --topic test

然后我输入一些字符串并按回车键，然后在控制台中看到以下内容。

[2014-11-24 15:27:36,022] ERROR fetching topic metadata for topics [Set(test)] from broker [ArrayBuffer(id:0,host:localhost,port:2181)] failed (kafka.utils.Utils$)
kafka.common.KafkaException: fetching topic metadata for topics [Set(test)] from broker [ArrayBuffer(id:0,host:localhost,port:2181)] failed
    at kafka.client.ClientUtils$.fetchTopicMetadata(ClientUtils.scala:67)
    at kafka.producer.BrokerPartitionInfo.updateInfo(BrokerPartitionInfo.scala:82)
    at kafka.producer.async.DefaultEventHandler$$anonfun$handle$2.apply$mcV$sp(DefaultEventHandler.scala:78)
    at kafka.utils.Utils$.swallow(Utils.scala:167)
    at kafka.utils.Logging$class.swallowError(Logging.scala:106)
    at kafka.utils.Utils$.swallowError(Utils.scala:46)
    at kafka.producer.async.DefaultEventHandler.handle(DefaultEventHandler.scala:78)
    at kafka.producer.async.ProducerSendThread.tryToHandle(ProducerSendThread.scala:104)
    at kafka.producer.async.ProducerSendThread$$anonfun$processEvents$3.apply(ProducerSendThread.scala:87)
    at kafka.producer.async.ProducerSendThread$$anonfun$processEvents$3.apply(ProducerSendThread.scala:67)
    at scala.collection.immutable.Stream.foreach(Stream.scala:547)
    at kafka.producer.async.ProducerSendThread.processEvents(ProducerSendThread.scala:66)
    at kafka.producer.async.ProducerSendThread.run(ProducerSendThread.scala:44)
Caused by: java.io.EOFException: Received -1 when reading from channel, socket has likely been closed.
    at kafka.utils.Utils$.read(Utils.scala:376)
    at kafka.network.BoundedByteBufferReceive.readFrom(BoundedByteBufferReceive.scala:54)
    at kafka.network.Receive$class.readCompletely(Transmission.scala:56)
    at kafka.network.BoundedByteBufferReceive.readCompletely(BoundedByteBufferReceive.scala:29)
    at kafka.network.BlockingChannel.receive(BlockingChannel.scala:100)
    at kafka.producer.SyncProducer.liftedTree1$1(SyncProducer.scala:74)
    at kafka.producer.SyncProducer.kafka$producer$SyncProducer$$doSend(SyncProducer.scala:71)
    at kafka.producer.SyncProducer.send(SyncProducer.scala:112)
    at kafka.client.ClientUtils$.fetchTopicMetadata(ClientUtils.scala:53)
    ... 12 more

我在 zookeeper 控制台上也看到了警告：

WARN Exception causing close of session 0x0 due to java.io.EOFException (org.apache.zookeeper.server.NIOServerCnxn)

我使用以下内容来设置我的主题：

./bin/kafka-topics.sh --create --zookeeper localhost:2181 --partitions 1 --replication-factor 1 --topic test

我可以列出我的主题：

./bin/kafka-topics.sh --list --zookeeper localhost:2181

【问题讨论】：

标签： apache-kafka

【解决方案1】：

您的 broker-list 参数指向 Zookeeper 而不是实际的代理。正确的用法是：

./bin/kafka-console-producer.sh --broker-list localhost:9092 --topic test

假设您的代理在端口 9092（默认）上运行。

您可以参考here 了解更多信息（您的问题在Step 4: Send some messages 中描述）

【讨论】：

我们可以配置这个broker运行的端口9092吗？我找不到相应的文档。
@Vinit89，是的，它在config/server.properties 称为port

【解决方案2】：

这是因为您以错误的方式启动消费者。如果你检查official documentation，你可以看到你必须：

启动 Zookeper

bin/zookeeper-server-start.sh config/zookeeper.properties

启动卡夫卡

bin/kafka-server-start.sh config/server.properties

创建一个主题

bin/kafka-topics.sh --create --zookeeper localhost:2181 --replication-factor 1 --partitions 1 --topic test

开始为某个主题生成消息

bin/kafka-console-producer.sh --broker-list localhost:9092 --topic test

注意区别：您正在连接到 zookeper（端口 2181），您需要连接到代理（端口 9092）

【讨论】：