我刚刚开始使用Apache Kafka / Zookeeper,并且遇到了尝试在AWS上建立集群的问题。目前,我有三台服务器:
一名正在运行的Zookeeper和两名正在运行的Kafka。
我可以毫无问题地启动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消费者上:
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)
不知道从这里去哪里。让我知道是否需要有关我的配置文件的更多详细信息
最佳答案
这是一个配置问题。
为了使其运行,必须对config
文件进行几处更改:
在每台Kafka服务器上的config/server.properties
中:
host.name: <Public IP>
advertised.host.name: <AWS Public DNS Address>
在每台Kafka服务器上的
config/producer.properties
中: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