Apache Kafka:无法更新Metadata/java.nio.channels.ClosedChannelException

Question

我刚开始使用Apache Kafka/Zookeeper,并且在尝试在AWS上设置集群时遇到了问题.目前我有三台服务器:

一个正在运行的Zookeeper和两个运行Kafka.

我可以毫无问题地启动Kafka服务器,并可以在这两个服务器上创建主题.但是,当我尝试在一台机器上启动一个生产者而在另一台机器上启动一个消费者时遇到了麻烦:

在卡夫卡制片人:

kafka-console-producer.sh --broker-list <kafka server 1 aws public dns>:9092,<kafka server 2 aws public dns>:9092 --topic samsa

关于卡夫卡消费者:

kafka-console-consumer.sh --zookeeper <zookeeper server ip>:2181 --topic samsa

我在制作人("hi")上输入一条消息,暂时没有任何反应.然后我收到这条消息:

ERROR Error when sending message to topic samsa with key: null, value: 2 bytes
with error: Failed to update metadata after 60000 ms.
(org.apache.kafka.clients.producer.internals.ErrorLoggingCallback)

在消费者方面,我收到此消息,该消息会定期重复:

WARN Fetching topic metadata with correlation id # for topics [Set(samsa)] from broker [BrokerEndPoint(<broker.id>,<producer's advertised.host.name>,9092)] failed (kafka.client.ClientUtils$)
java.nio.channels.ClosedChannelException
    at kafka.network.BlockingChannel.send(BlockingChannel.scala:110)
    at kafka.producer.SyncProducer.liftedTree1$1(SyncProducer.scala:75)
    at kafka.producer.SyncProducer.kafka$producer$SyncProducer$$doSend(SyncProducer.scala:74)
    at kafka.producer.SyncProducer.send(SyncProducer.scala:119)
    at kafka.client.ClientUtils$.fetchTopicMetadata(ClientUtils.scala:59)
    at kafka.client.ClientUtils$.fetchTopicMetadata(ClientUtils.scala:94)
    at kafka.consumer.ConsumerFetcherManager$LeaderFinderThread.doWork(ConsumerFetcherManager.scala:66)
    at kafka.utils.ShutdownableThread.run(ShutdownableThread.scala:63)

一段时间后,生产者将开始迅速抛出这个错误消息##递增增加:

WARN Error while fetching metadata with correlation id # : {samsa=LEADER_NOT_AVAILABLE} (org.apache.kafka.clients.NetworkClient)

不知道从哪里开始.如果需要有关我的配置文件的更多详细信息,请告诉我

Answer 1

这是一个配置问题.

为了让它运行,config必须对文件进行一些更改:

在config/server.properties每个Kafka服务器上:

• host.name: <Public IP>
• advertised.host.name: <AWS Public DNS Address>

在config/producer.properties每个Kafka服务器上:

• metadata.broker.list: <Producer Server advertised.host.name>:<Producer Server port>,<Consumer Server advertised.host.name>:<Consumer Server port>

在每个Kafka服务器上的/ etc/hosts中,更改127.0.0.1 localhost localhost.localdomain为:

<Public IP>  localhost localhost.localdomain