I've created a 3 broker kafka cluster, changing only the config values for
broker id, log.dirs, and zookeeper connect. I left the remaining fields as
default.
The broker ids are 1, 2, 3. I opened the port 9092 on AWS.
I then created a topic 'test' with replication factor of 2, and 3
partitions.
When I describe the topic using kafka-topics.sh --describe , it shows:
Topic:test PartitionCount:3 ReplicationFactor:2 Configs:
Topic: test Partition: 0 Leader: 1 Replicas: 1,3 Isr: 1,3
Topic: test Partition: 1 Leader: 2 Replicas: 2,1 Isr: 2,1
Topic: test Partition: 2 Leader: 3 Replicas: 3,2 Isr: 3,2
So it looks like the 3 brokers have successfully connected to each other.
I then tried running
bin/kafka-console-consumer.sh --zookeeper $KAFKA_ZOOKEEPER_CONNECT --topic
test --from-beginning
But it began to give me a lot of the following exceptions:
WARN Fetctestng topic metadata with correlation id 1 for topics
[Set(test)] from broker [BrokerEndPoint(1,kafka1-1876849043-91zso,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:80)
at
kafka.producer.SyncProducer.kafka$producer$SyncProducer$$doSend(SyncProducer.scala:79)
at kafka.producer.SyncProducer.send(SyncProducer.scala:124)
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)
[2016-09-16 17:22:11,323] WARN Fetctestng topic metadata with correlation
id 2 for topics [Set(test)] from broker
[BrokerEndPoint(3,kafka3-2571399577-96he4,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:80)
at
kafka.producer.SyncProducer.kafka$producer$SyncProducer$$doSend(SyncProducer.scala:79)
at kafka.producer.SyncProducer.send(SyncProducer.scala:124)
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)
Likewise when I run kafka-console-producer.sh , I see errors like:
[2016-09-16 17:24:42,901] WARN Error while fetching metadata with
correlation id 1343 : {test=LEADER_NOT_AVAILABLE}
(org.apache.kafka.clients.NetworkClient)
[2016-09-16 17:24:43,012] WARN Error while fetching metadata with
correlation id 1344 : {test=LEADER_NOT_AVAILABLE}
(org.apache.kafka.clients.NetworkClient)
[2016-09-16 17:24:43,127] WARN Error while fetching metadata with
correlation id 1345 : {test=LEADER_NOT_AVAILABLE}
(org.apache.kafka.clients.NetworkClient)
Any ideas what the problem here is?
I'm using kafka_2.11-0.10.0.1
<http://www-us.apache.org/dist/kafka/0.10.0.1/kafka_2.11-0.10.0.1.tgz>
