0

我在三个单独的节点上托管 Kafka，每个节点都有本地 Zookeeper 实例。在制作主题live时，我得到以下异常。metadata.brokers.list未设置为localhost, 但kafka1.domain.com, kafka2.domain.com, kafka3.domain.com. advertised.host.name设置为 Amazon 实例的本地 IP 地址。Kafka 设置中没有引用localhost。

这是生产者配置：

props.put("metadata.broker.list", "kafka1.domain.com:9092,kafka2.domain.com:9092,kafka3.domain.com:9092")
props.put("serializer.class", "kafka.serializer.StringEncoder")
props.put("producer.type", "async")
props.put("batch.num.messages", "1000")
props.put("queue.buffering.max.ms", "20000")
props.put("request.required.acks", "0")

每个都会producer.send产生异常。

有谁知道可能出了什么问题？

WARN ClientUtils$: Fetching topic metadata with correlation id 5 for topics [Set(live)] from broker [id:0,host:localhost,port:9092] failed
java.nio.channels.ClosedChannelException
        at kafka.network.BlockingChannel.send(BlockingChannel.scala:100)
        at kafka.producer.SyncProducer.liftedTree1$1(SyncProducer.scala:73)
        at kafka.producer.SyncProducer.kafka$producer$SyncProducer$$doSend(SyncProducer.scala:72)
        at kafka.producer.SyncProducer.send(SyncProducer.scala:113)
        at kafka.client.ClientUtils$.fetchTopicMetadata(ClientUtils.scala:58)
        at kafka.producer.BrokerPartitionInfo.updateInfo(BrokerPartitionInfo.scala:82)
        at kafka.producer.BrokerPartitionInfo.getBrokerPartitionInfo(BrokerPartitionInfo.scala:49)
        at kafka.producer.async.DefaultEventHandler.kafka$producer$async$DefaultEventHandler$$getPartitionListForTopic(DefaultEventHandler.scala:186)
        at kafka.producer.async.DefaultEventHandler$$anonfun$partitionAndCollate$1.apply(DefaultEventHandler.scala:150)
        at kafka.producer.async.DefaultEventHandler$$anonfun$partitionAndCollate$1.apply(DefaultEventHandler.scala:149)
        at scala.collection.mutable.ResizableArray$class.foreach(ResizableArray.scala:59)
        at scala.collection.mutable.ArrayBuffer.foreach(ArrayBuffer.scala:47)
        at kafka.producer.async.DefaultEventHandler.partitionAndCollate(DefaultEventHandler.scala:149)
        at kafka.producer.async.DefaultEventHandler.dispatchSerializedData(DefaultEventHandler.scala:95)
        at kafka.producer.async.DefaultEventHandler.handle(DefaultEventHandler.scala:72)
        at kafka.producer.async.ProducerSendThread.tryToHandle(ProducerSendThread.scala:105)
        at kafka.producer.async.ProducerSendThread.processEvents(ProducerSendThread.scala:94)
        at kafka.producer.async.ProducerSendThread.run(ProducerSendThread.scala:45)
ERROR DefaultEventHandler: Failed to collate messages by topic, partition due to: fetching topic metadata for topics [Set(live)] from broker [ArrayBuffer(id:0,host:localhost,port:9092)] failed
WARN ClientUtils$: Fetching topic metadata with correlation id 6 for topics [Set(live)] from broker [id:0,host:localhost,port:9092] failed
java.nio.channels.ClosedChannelException
        at kafka.network.BlockingChannel.send(BlockingChannel.scala:100)
        at kafka.producer.SyncProducer.liftedTree1$1(SyncProducer.scala:73)
        at kafka.producer.SyncProducer.kafka$producer$SyncProducer$$doSend(SyncProducer.scala:72)
        at kafka.producer.SyncProducer.send(SyncProducer.scala:113)
        at kafka.client.ClientUtils$.fetchTopicMetadata(ClientUtils.scala:58)
        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:172)
        at kafka.utils.Logging$class.swallowError(Logging.scala:106)
        at kafka.utils.Utils$.swallowError(Utils.scala:45)
        at kafka.producer.async.DefaultEventHandler.handle(DefaultEventHandler.scala:78)
        at kafka.producer.async.ProducerSendThread.tryToHandle(ProducerSendThread.scala:105)
        at kafka.producer.async.ProducerSendThread.processEvents(ProducerSendThread.scala:94)
        at kafka.producer.async.ProducerSendThread.run(ProducerSendThread.scala:45)
ERROR Utils$: fetching topic metadata for topics [Set(live)] from broker [ArrayBuffer(id:0,host:localhost,port:9092)] failed
kafka.common.KafkaException: fetching topic metadata for topics [Set(live)] from broker [ArrayBuffer(id:0,host:localhost,port:9092)] failed
        at kafka.client.ClientUtils$.fetchTopicMetadata(ClientUtils.scala:72)
        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:172)
        at kafka.utils.Logging$class.swallowError(Logging.scala:106)
        at kafka.utils.Utils$.swallowError(Utils.scala:45)
        at kafka.producer.async.DefaultEventHandler.handle(DefaultEventHandler.scala:78)
        at kafka.producer.async.ProducerSendThread.tryToHandle(ProducerSendThread.scala:105)
        at kafka.producer.async.ProducerSendThread.processEvents(ProducerSendThread.scala:94)
        at kafka.producer.async.ProducerSendThread.run(ProducerSendThread.scala:45)
Caused by: java.nio.channels.ClosedChannelException
        at kafka.network.BlockingChannel.send(BlockingChannel.scala:100)
        at kafka.producer.SyncProducer.liftedTree1$1(SyncProducer.scala:73)
        at kafka.producer.SyncProducer.kafka$producer$SyncProducer$$doSend(SyncProducer.scala:72)
        at kafka.producer.SyncProducer.send(SyncProducer.scala:113)
        at kafka.client.ClientUtils$.fetchTopicMetadata(ClientUtils.scala:58)
        ... 9 more
4

1 回答 1

1

我正在使用kafka.producer.Producerin kafka:0.8.2.1。显然它被窃听了，并且通过改为org.apache.kafka.clients.producer.KafkaProducerin kafka-clients:0.8.2.1，它在相同的配置上立即运行而没有任何问题。

于 2015-11-13T11:54:38.460 回答