Why can't use consumer correctly with Kafka?

Question

There are 3 kafka servers

Set /etc/hosts on all the 3 servers.

192.168.0.1  kafka1
192.168.0.2  kafka2
192.168.0.3  kafka3

Installed zookeeper and kafka on all of them with configuration

/usr/local/kafka_2.12-2.6.0/config/server.properties

kafka1

#
broker.id=1
listeners=PLAINTEXT://kafka1:9092
advertised.listeners=PLAINTEXT://kafka1:9092

kafka2

#
broker.id=2
listeners=PLAINTEXT://kafka2:9092
advertised.listeners=PLAINTEXT://kafka2:9092

kafka3

#
broker.id=3
listeners=PLAINTEXT://kafka3:9092
advertised.listeners=PLAINTEXT://kafka3:9092

After start zookeeper and kafka, create a new topic

[kafka@kafka1 ~]$ bin/kafka-topics.sh --create --zookeeper kafka1:2181,kafka2:2181,kafka3:2181 --replication-factor 1 --partitions 6 --topic topic1 --config cleanup.policy=delete --config delete.retention.ms=60000

Check the cluster status on all the three nodes

kafka1

[kafka@kafka1 ~]$ bin/kafka-topics.sh --describe --zookeeper kafka1:2181 --topic topic1
Topic: topic1   PartitionCount: 6   ReplicationFactor: 1    Configs: cleanup.policy=delete,delete.retention.ms=60000
    Topic: topic1   Partition: 0    Leader: 2   Replicas: 2 Isr: 2
    Topic: topic1   Partition: 1    Leader: 3   Replicas: 3 Isr: 3
    Topic: topic1   Partition: 2    Leader: 1   Replicas: 1 Isr: 1
    Topic: topic1   Partition: 3    Leader: 2   Replicas: 2 Isr: 2
    Topic: topic1   Partition: 4    Leader: 3   Replicas: 3 Isr: 3
    Topic: topic1   Partition: 5    Leader: 1   Replicas: 1 Isr: 1

kafka2

[kafka@kafka2 ~]$ bin/kafka-topics.sh --describe --zookeeper kafka1:2181 --topic topic1
Topic: topic1   PartitionCount: 6   ReplicationFactor: 1    Configs: cleanup.policy=delete,delete.retention.ms=60000
    Topic: topic1   Partition: 0    Leader: 2   Replicas: 2 Isr: 2
    Topic: topic1   Partition: 1    Leader: 3   Replicas: 3 Isr: 3
    Topic: topic1   Partition: 2    Leader: 1   Replicas: 1 Isr: 1
    Topic: topic1   Partition: 3    Leader: 2   Replicas: 2 Isr: 2
    Topic: topic1   Partition: 4    Leader: 3   Replicas: 3 Isr: 3
    Topic: topic1   Partition: 5    Leader: 1   Replicas: 1 Isr: 1

kafka3

[kafka@kafka3 ~]$ bin/kafka-topics.sh --describe --zookeeper kafka1:2181 --topic topic1
Topic: topic1   PartitionCount: 6   ReplicationFactor: 1    Configs: cleanup.policy=delete,delete.retention.ms=60000
    Topic: topic1   Partition: 0    Leader: 2   Replicas: 2 Isr: 2
    Topic: topic1   Partition: 1    Leader: 3   Replicas: 3 Isr: 3
    Topic: topic1   Partition: 2    Leader: 1   Replicas: 1 Isr: 1
    Topic: topic1   Partition: 3    Leader: 2   Replicas: 2 Isr: 2
    Topic: topic1   Partition: 4    Leader: 3   Replicas: 3 Isr: 3
    Topic: topic1   Partition: 5    Leader: 1   Replicas: 1 Isr: 1

But when test the cluster, created a producer on kafka1

[kafka@kafka1 ~]$ bin/kafka-console-producer.sh --broker-list kafka1:9092 --topic topic1
>

On any other nodes(includes) current kafka1 node, run consumer

bin/kafka-console-consumer.sh --bootstrap-server kafka1:9092 --topic topic1

Got the same error:

# kafka1
[2020-08-21 02:09:57,299] WARN [Consumer clientId=consumer-console-consumer-39789-1, groupId=console-consumer-39789] Connection to node 2147483645 (kafka2/192.168.0.2:9092) could not be established. Broker may not be available. (org.apache.kafka.clients.NetworkClient)
# kafka2
[2020-08-21 03:05:00,573] WARN [Consumer clientId=consumer-console-consumer-71891-1, groupId=console-consumer-71891] Connection to node -1 (kafka1/192.168.0.1:9092) could not be established. Broker may not be available. (org.apache.kafka.clients.NetworkClient)
# kafka3
[2020-08-21 03:05:14,331] WARN [Consumer clientId=consumer-console-consumer-55574-1, groupId=console-consumer-55574] Connection to node -1 (kafka1/192.168.0.1:9092) could not be established. Broker may not be available. (org.apache.kafka.clients.NetworkClient)

In /usr/local/kafka_2.12-2.6.0/config/server.properties, I also tried to set to

listeners=PLAINTEXT://localhost:9092
advertised.listeners=PLAINTEXT://localhost:9092

on all the 3 nodes. But the same issue.

Answer 1

The "listeners" should represent the node FQDN and port where your producers and consumers can access. When producer / consumer connects to Kafka, like in your post:

bin/kafka-console-consumer.sh --bootstrap-server kafka1:9092 --topic topic1

It open connection to kafka1:9092 and then it gets from Kafka a map of the cluster with all brokers, theirFQDN,PORT and the topics partitions each broker is a leader of.

You managed to create topic which is done using zookeeper and that worked.

According to your WARN logs, the "broker1,2,3" is resolved to 192.168.0.1/2/3 so the IP resolution is ok.

Try to check network connectivity on port 9092: from broker1 run: "telnet broker2 9092"

or run the producer command to access remote broker i.e.:

[kafka@**kafka1** ~]$ bin/kafka-console-producer.sh --broker-list **kafka2**:9092 --topic topic1

this will tell you if you are able to connect from broker1 to broker2 on port 9092.