Kafka client can't receive messages

Question

I have kafka and zookeeper set up on a remote machine. On that machine I'm able to see below working using the test method on official website.

> bin/kafka-console-producer.sh --broker-list localhost:9092 --topic listings-incoming
This is a message
This is another message

> bin/kafka-console-consumer.sh --bootstrap-server localhost:9092 --listings-incoming --from-beginning
This is a message
This is another message

but when I use my local consumer script it is not working:

bin/kafka-console-consumer.sh —bootstrap-server X.X.X.X:9092 —listings-incoming —from-beginning —consumer-property group.id=group2

Haven't seen messages showing up but what is showing is:

[2017-08-11 14:39:56,425] WARN Auto-commit of offsets {listings-incoming-4=OffsetAndMetadata{offset=0, metadata=''}, listings-incoming-2=OffsetAndMetadata{offset=0, metadata=''}, listings-incoming-3=OffsetAndMetadata{offset=0, metadata=''}, listings-incoming-0=OffsetAndMetadata{offset=0, metadata=''}, listings-incoming-1=OffsetAndMetadata{offset=0, metadata=''}} failed for group group1: Commit cannot be completed since the group has already rebalanced and assigned the partitions to another member. This means that the time between subsequent calls to poll() was longer than the configured max.poll.interval.ms, which typically implies that the poll loop is spending too much time message processing. You can address this either by increasing the session timeout or by reducing the maximum size of batches returned in poll() with max.poll.records. (org.apache.kafka.clients.consumer.internals.ConsumerCoordinator)

*****************update**********************

My zookeeper and kafka is running on the same machine, right now my configuration on advertised.listeners is this:

advertised.listeners=PLAINTEXT://the.machine.ip.address:9092

I tried to change it to:

advertised.listeners=PLAINTEXT://my.client.ip.address:9092

and then run the client side consumer script, it gives error:

[2017-08-11 15:49:01,591] WARN Error while fetching metadata with correlation id 3 : {listings-incoming=LEADER_NOT_AVAILABLE} (org.apache.kafka.clients.NetworkClient) [2017-08-11 15:49:22,106] WARN Bootstrap broker 10.161.128.238:9092 disconnected (org.apache.kafka.clients.NetworkClient) [2017-08-11 15:49:22,232] WARN Error while fetching metadata with correlation id 7 : {listings-incoming=LEADER_NOT_AVAILABLE} (org.apache.kafka.clients.NetworkClient) [2017-08-11 15:49:22,340] WARN Error while fetching metadata with correlation id 8 : {listings-incoming=LEADER_NOT_AVAILABLE} (org.apache.kafka.clients.NetworkClient) [2017-08-11 15:49:40,453] WARN Bootstrap broker 10.161.128.238:9092 disconnected (org.apache.kafka.clients.NetworkClient) [2017-08-11 15:49:40,531] WARN Error while fetching metadata with correlation id 12 : {listings-incoming=LEADER_NOT_AVAILABLE} (org.apache.kafka.clients.NetworkClient)

Answer 1

You probably have not configured your advertised.listeners properly in the brokers server.properties file.

From https://kafka.apache.org/documentation/

advertised.listeners Listeners to publish to ZooKeeper for clients to use, if different than the listeners above. In IaaS environments, this may need to be different from the interface to which the broker binds. If this is not set, the value for listeners will be used.

and in the same documentation

listeners Listener List - Comma-separated list of URIs we will listen on and the listener names. If the listener name is not a security protocol, listener.security.protocol.map must also be set. Specify hostname as 0.0.0.0 to bind to all interfaces. Leave hostname empty to bind to default interface. Examples of legal listener lists: PLAINTEXT://myhost:9092,SSL://:9091 CLIENT://0.0.0.0:9092,REPLICATION://localhost:9093

So if advertised.listeners is not set and listeners is just listening to localhost:9092 or 127.0.0.1:9092 or 0.0.0.0:9092 then the clients will be told to connect to localhost when they make a meta-data request to the bootstrap server. That works when the client is actually running in the same machine as the broker but it will fail when you connect remotely.

You should set advertised.listeners to be a fully qualified domain name or public IP address for the host that the broker is running on.

For example

advertised.listeners=PLAINTEXT://kafkabrokerhostname.confluent.io:9092

or

advertised.listeners=PLAINTEXT://192.168.1.101:9092