Im running a kafka broker in my home server, and since I want it to be accessbile from outside my network for various reasons I do not wan't my broker to accept which ever connection. In this post I will only show you how to set up SSL encryption over the wire, not authentication, which kafka has support for as well. Also I will not secure Zookeeper since my zookeeper isn't accessible from the outside world.

So here I'll show you how to generate the certificates needed using the java keygen tool and how to configure kafka and your clients to use TLS (SSL).

Firstly you will need to generate certificate authority that we will use to sign our client and server certificates.

This snippet is copied from a confluent blog post that helped me setting up SSL. Not that this post is basically how I configured kafka and kafka-streams without using the confluent flavor of kafka.

keytool -keystore kafka.server.keystore.jks -alias kafka -validity $VALIDITY -genkey
openssl req -new -x509 -keyout ca-key -out ca-cert -days $VALIDITY
keytool -keystore kafka.server.truststore.jks -alias CARoot -import -file ca-cert
keytool -keystore kafka.client.truststore.jks -alias CARoot -import -file ca-cert
keytool -keystore kafka.server.keystore.jks -alias kafka -certreq -file cert-file
openssl x509 -req -CA ca-cert -CAkey ca-key -in cert-file -out cert-signed -days $VALIDITY -CAcreateserial -passin pass:$PASSWORD
keytool -keystore kafka.server.keystore.jks -alias CARoot -import -file ca-cert
keytool -keystore kafka.server.keystore.jks -alias kafka -import -file cert-signed
keytool -keystore kafka.client.keystore.jks -alias kafka -validity $VALIDITY -genkey
keytool -keystore kafka.client.keystore.jks -alias kafka -certreq -file cert-file
openssl x509 -req -CA ca-cert -CAkey ca-key -in cert-file -out cert-signed -days $VALIDITY -CAcreateserial -passin pass:$PASSWORD
keytool -keystore kafka.client.keystore.jks -alias CARoot -import -file ca-cert
keytool -keystore kafka.client.keystore.jks -alias kafka -import -file cert-signed

Note that the alias is only a name, however, if we wan't to run locally we must add localhost as the CN and FQDN when we generate our trust and keystores and the corresponding. Also I've set the validity to 10 years which is high for production systems, but this is only for my own home setup.

keytool -keystore keystore.jks -alias kafka -validity 365 -genkey
Enter keystore password:
Re-enter new password:
What is your first and last name?
  [Unknown]:  localhost
What is the name of your organizational unit?
What is the name of your organization?
What is the name of your City or Locality?
What is the name of your State or Province?
What is the two-letter country code for this unit?
  [Unknown]:  SE
Is CN=localhost, OU=Unknown, O=Unknown, L=Unknown, ST=Unknown, C=SE correct?
  [no]:  yes

Now we have everything that we need to configure kafka and our clients, so lets do that. We'll begin with configuring our broker:

docker run -d \
	   --net=messaging \
	   --name=kafka \
 	   -v /home/username/Projects/docker/kafka.localhost.server.truststore.jks:/var/truststore.jks \
 	   -v /home/username/Projects/docker/kafka.localhost.server.keystore.jks:/var/keystore.jks \
	   -p 29092:29092 \
	   -p 29093:29093 \
	   -e KAFKA_ZOOKEEPER_CONNECT=zookeeper:2181 \
	   -e KAFKA_ADVERTISED_LISTENERS=INSIDE://kafka:9092,PLAINTEXT://:29092,SSL://:29093 \
	   -e KAFKA_LISTENERS=INSIDE://:9092,PLAINTEXT://:29092,SSL://:29093 \
	   -e KAFKA_SSL_KEYSTORE_LOCATION=/var/keystore.jks \
 	   -e KAFKA_SSL_KEY_PASSWORD=***** \
	   -e KAFKA_SSL_TRUSTSTORE_LOCATION=/var/truststore.jks \

That's it for kafka, now for our clients we only need the truststore, since I've choosen to not to authenticate, if the certificate is accepted I'm happy. For the clients we only need the client truststore.

Properties props = new Properties();
    props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, bootstrapServers);
    props.put(ProducerConfig.CLIENT_ID_CONFIG, clientId);
    props.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class);
    props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, KafkaAvroSerializer.class.getName());
    props.put(KafkaAvroSerializerConfig.SCHEMA_REGISTRY_URL_CONFIG, schemaRegistryURL);
    props.put("security.protocol", "SSL");
    props.put("ssl.truststore.location", trustStorePath);
    props.put("ssl.truststore.password", "*****");

And that's it, we now can spin up a docker container locally that is running kafka and have encrypted communication.

Note if you want to change the host, don't forget to change it in the CN and FQDN when you generate the key- and trust-stores.