Support Questions
Find answers, ask questions, and share your expertise

Consumer and producer unable to communicate

Consumer and producer unable to communicate

New Contributor

About versions:

HDP: 2.6

Kafka: 0.10.1

Kerberized cluster

I'm able to create the consumer and producer in kafka but in producer I get this error:

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

And when I run the consumer command I get this error:

{metadata.broker.list=s154mshdpdn002.br154.corpintra.net:6667,s154mshdpdn001.br154.corpintra.net:6667,s154mshdpdn003.br154.corpintra.net:6667, request.timeout.ms=30000, client.id=console-consumer-1563, security.protocol=PLAINTEXTSASL}
[2018-12-04 17:15:42,649] WARN Fetching topic metadata with correlation id 0 for topics [Set(primeiro)] from broker [BrokerEndPoint(1002,s154mshdpdn001.br154.corpintra.net,6667)] failed (kafka.client.ClientUtils$)
java.nio.channels.ClosedChannelException
        at kafka.network.BlockingChannel.send(BlockingChannel.scala:122)
        at kafka.producer.SyncProducer.liftedTree1$1(SyncProducer.scala:82)
        at kafka.producer.SyncProducer.kafka$producer$SyncProducer$$doSend(SyncProducer.scala:81)
        at kafka.producer.SyncProducer.send(SyncProducer.scala:126)
        at kafka.client.ClientUtils$.fetchTopicMetadata(ClientUtils.scala:59)
        at kafka.client.ClientUtils$.fetchTopicMetadata(ClientUtils.scala:96)
        at kafka.consumer.ConsumerFetcherManager$LeaderFinderThread.doWork(ConsumerFetcherManager.scala:67)
        at kafka.utils.ShutdownableThread.run(ShutdownableThread.scala:63)
[2018-12-04 17:15:43,023] WARN Fetching topic metadata with correlation id 0 for topics [Set(primeiro)] from broker [BrokerEndPoint(1001,s154mshdpdn002.br154.corpintra.net,6667)] failed (kafka.client.ClientUtils$)
java.nio.channels.ClosedChannelException
        at kafka.network.BlockingChannel.send(BlockingChannel.scala:122)
        at kafka.producer.SyncProducer.liftedTree1$1(SyncProducer.scala:82)
        at kafka.producer.SyncProducer.kafka$producer$SyncProducer$$doSend(SyncProducer.scala:81)
        at kafka.producer.SyncProducer.send(SyncProducer.scala:126)
        at kafka.client.ClientUtils$.fetchTopicMetadata(ClientUtils.scala:59)
        at kafka.client.ClientUtils$.fetchTopicMetadata(ClientUtils.scala:96)
        at kafka.consumer.ConsumerFetcherManager$LeaderFinderThread.doWork(ConsumerFetcherManager.scala:67)
        at kafka.utils.ShutdownableThread.run(ShutdownableThread.scala:63)
[2018-12-04 17:15:43,318] WARN Fetching topic metadata with correlation id 0 for topics [Set(primeiro)] from broker [BrokerEndPoint(1003,s154mshdpdn003.br154.corpintra.net,6667)] failed (kafka.client.ClientUtils$)
java.nio.channels.ClosedChannelException
        at kafka.network.BlockingChannel.send(BlockingChannel.scala:122)
        at kafka.producer.SyncProducer.liftedTree1$1(SyncProducer.scala:82)
        at kafka.producer.SyncProducer.kafka$producer$SyncProducer$$doSend(SyncProducer.scala:81)
        at kafka.producer.SyncProducer.send(SyncProducer.scala:126)
        at kafka.client.ClientUtils$.fetchTopicMetadata(ClientUtils.scala:59)
        at kafka.client.ClientUtils$.fetchTopicMetadata(ClientUtils.scala:96)
        at kafka.consumer.ConsumerFetcherManager$LeaderFinderThread.doWork(ConsumerFetcherManager.scala:67)
        at kafka.utils.ShutdownableThread.run(ShutdownableThread.scala:63)
[2018-12-04 17:15:43,319] WARN [console-consumer-1563_s154mshdpdn001.br154.corpintra.net-1543950941883-ccbd0b00-leader-finder-thread], Failed to find leader for Set(primeiro-0) (kafka.consumer.ConsumerFetcherManager$LeaderFinderThread)
kafka.common.KafkaException: fetching topic metadata for topics [Set(primeiro)] from broker [ArrayBuffer(BrokerEndPoint(1002,s154mshdpdn001.br154.corpintra.net,6667), BrokerEndPoint(1001,s154mshdpdn002.br154.corpintra.net,6667), 

I'm able to create topics and delete correctly.

Commands that I used to run consumer and producer:

/usr/hdp/current/kafka-broker/bin/kafka-console-producer.sh --broker-list $(hostname -f):6667 --topic mercedes --security-protocol PLAINTEXTSASL


/usr/hdp/current/kafka-broker/bin/kafka-console-consumer.sh --zookeeper $(hostname -f):2181 --topic mercedes --security-protocol PLAINTEXTSASL

I also tried to run consumer like this:

/usr/hdp/current/kafka-broker/bin/kafka-console-consumer.sh --bootstrap-server $(hostname -f):6667 --topic mercedes --security-protocol PLAINTEXTSASL

But the consumer was unable to receive the messages from the producer.


I tried to change the hostname to localhost and the IP address from the server, removed kafka broker by zookeeper-shell and restart kafka service and even reinstall kafka. But it is still not working. Any suggestions?

Thanks all.

Carolina.

Don't have an account?