Connecting Kafka producer/consumer to broker via TLS

Question

I am trying to setup TLS for kafka broker. I have followed the steps here and able to setup the Kafka with TLS. (In log, I see SSL entry for the configured port).

Now I am facing the issue with connecting the producer/consumer.

1. I created a client keystore using the below command,

   keytool -keystore client.keystore.jks -alias localhost -validity 365 -keyalg RSA -genkey
   

2. Added the CA cert to the keystore,

   keytool -keystore client.keystore.jks -alias CARoot -import -file ca-cert
   

3. Ran the below command in the client, where the ca-cert is the certificate used on the server.

   keytool -keystore client.truststore.jks -alias CARoot -import -file ca-cert
   keytool -keystore client.keystore.jks -alias localhost -validity 365 -keyalg RSA -genkey
   keytool -keystore client.keystore.jks -alias CARoot -import -file ca-cert
   

4. Added the below config in the producer.properties,

   security.protocol=SSL
   ssl.truststore.location=path to client.truststore.jks
   ssl.truststore.password=<password>
   ssl.keystore.location=path to client.keystore.jks
   ssl.keystore.password=<password>
   ssl.key.password=<password>
   

5. Ran kafka-console-producer

   kafka-console-producer.sh --broker-list 0.0.0.0:9092 --topic test --producer.config ../config/producer.properties

But I am getting the below error when running the util,

WARN Connection to node -1 terminated during authentication. This may indicate that authentication failed due to invalid credentials. (org.apache.kafka.clients.NetworkClient)

Suspecting that I am missing something in the client config. Any help would be greatly appreciated.

Answer 1

Are you trying with client side certificate ? Rather I would recommend, try without client certificate. In that case you only need below entries,

producer.properties file:-

security.protocol=SSL
ssl.truststore.location=/<path-to>/truststore.jks
ssl.truststore.type=JKS

Read more about it here - http://kafka.apache.org/documentation/#security_configclients

For client authentication kafka uses SASL, This part of the document covers it clearly - http://kafka.apache.org/documentation/#security_sasl