Console producer/consumer not working in kafka 0.10.2 with kerberos

[#1]

Hi,

We have recently started using kafka 0.10.2 but are unable to produce any messages or consumer them. It has kerberos enabled. Below are my configs. There is no error and kafka data log also doesn't have any entry but the index gets updated whenever we run an producer.

kafka-console-producer --broker-list kafka1.test.com:9092,kafka2.test.com:9092 --producer.config client.properties --topic TEST

kafka-console-consumer --topic TEST --from-beginning --bootstrap-server kafka1.test.com:9092,kafka2.test.com:9092 --consumer.config consumer.properties

jass:

KafkaClient { com.sun.security.auth.module.Krb5LoginModule required useTicketCache=true; };

client.properties/consumer.properties:

security.protocol=SASL_PLAINTEXT

sasl.kerberos.service.name=kafka

[#2]

Below are the errors i see in producer.

17/11/22 12:43:01 ERROR internals.ErrorLoggingCallback: Error when sending message to topic TEST with key: null, value: 4 bytes with error: org.apache.kafka.common.errors.TimeoutException: Failed to update metadata after 60000 ms. 17/11/22 12:44:01 ERROR internals.ErrorLoggingCallback: Error when sending message to topic TEST with key: null, value: 2 bytes with error: org.apache.kafka.common.errors.TimeoutException: Failed to update metadata after 60000 ms. 17/11/22 12:45:01 ERROR internals.ErrorLoggingCallback: Error when sending message to topic TEST with key: null, value: 5 bytes with error: org.apache.kafka.common.errors.TimeoutException: Failed to update metadata after 60000 ms. 17/11/22 12:46:01 ERROR internals.ErrorLoggingCallback: Error when sending message to topic TEST with key: null, value: 4 bytes with error: org.apache.kafka.common.errors.TimeoutException: Failed to update metadata after 60000 ms.

[#3]

@Rajesh Reddy,

Are you using kafka from HDP or a stand alone distribution of Kafka. If you are using Kafka from HDP you have to change the port no from 9092 to 6667 (default)

Check the value of port from (Advanced kafka-broker configs in Ambari)

Thanks,

Aditya

[#4]

@aditya

im using a standalone Kafka.

[#5]

@Rajesh Reddy,

Can you please verify the below things

1) telnet kafka1.test.com 9092 and telnet kafka2.test.com 9092 is working fine.

2) Did you do kinit before running the producer and consumer.

3) Is the topic present : kafka-topics --list --zookeeper <zkhost>:2181

4) Use the jaas config

export KAFKA_OPTS="-Djava.security.auth.login.config=jaas.conf"

5) Check bind address in server.properties . should be 0.0.0.0:9092

[#6]

@Aditya Sirna

Thanks for the prompt reply. Yes, we did all the mentioned above.

[#7]

@Rajesh Reddy

What is your kafka log telling? Maybe you can try something like this...

Consumer:
/bin/kafka-console-consumer.sh --zookeeper zk1:2181,zk2:2181,zk2:2181 --topic your_topic_name --security-protocol SASL_PLAINTEXT --from-beginning

Producer:
/bin/kafka-console-producer.sh --broker-list broker1:6667,broker2:6667 --topic your_topic_name--security-protocol SASL_PLAINTEXT

If issue due to security (make sure backup your original file in case not working):
KafkaClient {
com.sun.security.auth.module.Krb5LoginModule required
useKeyTab=true
keyTab="/etc/security/keytabs/kafka.service.keytab"
storeKey=true
useTicketCache=false
serviceName="kafka"
principal="kafka/<hostname>@<domain.com>";

};

Client {

com.sun.security.auth.module.Krb5LoginModule required
useKeyTab=true
keyTab="/etc/security/keytabs/kafka.service.keytab"
storeKey=true
useTicketCache=false
serviceName="zookeeper"
principal="kafka/<hostname>@<domain.com>";;

};

[#8]

@SIOKEK YIM

Do i need to be using kafka keytab??

[#9]

Yes, if you keberized your cluster

[#10]

@SIOKEK YIM

Do you mean all my topic creation, producing and consuming has to be done with kafka user? I'm not sure if kafka has any impersonation..!! Tried doing using kafka keytab. Below is the topic created with kafka service keytab. But its the same issue

Topic:Hello1 PartitionCount:2 ReplicationFactor:2 Configs:min.insync.replicas=2

Topic: Hello1 Partition: 0 Leader: 35 Replicas: 35,38 Isr: 35,38

Topic: Hello1 Partition: 1 Leader: 38 Replicas: 38,33 Isr: 38,33

Producer:

=======kafka-verifiable-producer.sh --topic Hello1 --broker-list server1.kafka2.pre.corp:9092,server2.kafka2.pre.corp:9092 --producer.config client.properties

17/11/23 09:58:07 INFO utils.AppInfoParser: Kafka version : 0.10.2-kafka-2.2.0 17/11/23 09:58:07 INFO utils.AppInfoParser: Kafka commitId : unknown

1 2 3 4 45

17/11/23 09:58:25 INFO producer.KafkaProducer: Closing the Kafka producer with timeoutMillis = 9223372036854775807 ms. {"name":"shutdown_complete"} {"sent":1,"name":"tool_data","avg_throughput":0.0,"target_throughput":-1,"acked":0}

Consumer:

kafka-console-consumer --topic Hello1 --from-beginning --zookeeper server.kafka2.pre.corp:2181,server.kafka2.pre.corp:2181/kafka2dc1pre --consumer.config consumer.properties

17/11/23 09:59:04 INFO utils.ZKCheckedEphemeral: Creating /consumers/console-consumer-59653/ids/console-consumer-59653_server1.kafka2.pre.corp-1511431144657-52e4b1e7 (is it secure? false)

17/11/23 09:59:04 INFO utils.ZKCheckedEphemeral: Result of znode creation is: OK

.

.

17/11/23 09:59:05 INFO consumer.ZookeeperConsumerConnector: [console-consumer-59653_server1.kafka2.pre.corp-1511431144657-52e4b1e7], end rebalancing consumer console-consumer-59653_server1.kafka2.pre.corp-1511431144657-52e4b1e7 try #0

17/11/23 09:59:05 INFO consumer.ZookeeperConsumerConnector: [console-consumer-59653_server1.kafka2.pre.corp-1511431144657-52e4b1e7], Creating topic event watcher for topics Hello1

17/11/23 09:59:05 INFO consumer.ZookeeperConsumerConnector: [console-consumer-59653_server1.kafka2.pre.corp-1511431144657-52e4b1e7], Topics to consume = ArrayBuffer(Hello1)

17/11/23 09:59:05 WARN consumer.ConsumerFetcherManager$LeaderFinderThread: [console-consumer-59653_server1.kafka2.pre.corp-1511431144657-52e4b1e7-leader-finder-thread], Failed to find leader for Set(Hello1-0, Hello1-1) kafka.common.BrokerEndPointNotAvailableException: End point with security protocol PLAINTEXT not found for broker 33 at kafka.client.ClientUtils$anonfun$getPlaintextBrokerEndPoints$1$anonfun$apply$5.apply(ClientUtils.scala:146) at kafka.client.ClientUtils$anonfun$getPlaintextBrokerEndPoints$1$anonfun$apply$5.apply(ClientUtils.scala:146) at scala.Option.getOrElse(Option.scala:121) at kafka.client.ClientUtils$anonfun$getPlaintextBrokerEndPoints$1.apply(ClientUtils.scala:146) at kafka.client.ClientUtils$anonfun$getPlaintextBrokerEndPoints$1.apply(ClientUtils.scala:142) at scala.collection.TraversableLike$anonfun$map$1.apply(TraversableLike.scala:234) at scala.collection.TraversableLike$anonfun$map$1.apply(TraversableLike.scala:234) at scala.collection.mutable.ResizableArray$class.foreach(ResizableArray.scala:59) at scala.collection.mutable.ArrayBuffer.foreach(ArrayBuffer.scala:48) at scala.collection.TraversableLike$class.map(TraversableLike.scala:234) at scala.collection.AbstractTraversable.map(Traversable.scala:104) at kafka.client.ClientUtils$.getPlaintextBrokerEndPoints(ClientUtils.scala:142) at kafka.consumer.ConsumerFetcherManager$LeaderFinderThread.doWork(ConsumerFetcherManager.scala:67) at kafka.utils.ShutdownableThread.run(ShutdownableThread.scala:63)

17/11/23 09:59:05 INFO consumer.ConsumerFetcherManager: [ConsumerFetcherManager-1511431144680] Added fetcher for partitions ArrayBuffer()

17/11/23 09:59:05 WARN consumer.ConsumerFetcherManager$LeaderFinderThread: [console-consumer-59653_server1.kafka2.pre.corp-1511431144657-52e4b1e7-leader-finder-thread], Failed to find leader for Set(Hello1-0, Hello1-1) kafka.common.BrokerEndPointNotAvailableException: End point with security protocol PLAINTEXT not found for broker 33 at kafka.client.ClientUtils$anonfun$getPlaintextBrokerEndPoints$1$anonfun$apply$5.apply(ClientUtils.scala:146)

[#11]

The issue actually seems to be with the kafka root in zookeeper. Initially the kafka root was / and then we changed it to /kafkapre in zookeeper. But kafka seems to be searching for / itself still.

2017-11-23 09:16:12,539 WARN nl.techop.kafka.dao.zookeeper.KafkaZkClient: KeeperErrorCode = NoNode for /brokers/topics/TEST/partitions/1/state

2017-11-23 09:17:42,548 WARN nl.techop.kafka.dao.zookeeper.KafkaZkClient: KeeperErrorCode = NoNode for /brokers/topics/TEST/partitions/1/state

2017-11-23 09:19:12,533 WARN nl.techop.kafka.dao.zookeeper.KafkaZkClient: KeeperErrorCode = NoNode for /brokers/topics/TEST/partitions/1/state

[#12]

@Rajesh Reddy

1. Is your Kafka broker security protocol using PLAINTEXT? But when you consume/produce you used SASL_PLAINTEXT (client.properties/consumer.properties) ?
2. Maybe you can try disable kerberos security and try whether Kafka working or not in none-security mode then only enable to security for further testing.
3. Since you do not have ranger policy, I think the account that you're using should be okay. But prefer use the same account for troubleshooting