Kafka TLS Configuration: Encrypting Broker-Client and Inter-Broker Traffic

Apache Kafka’s default configuration runs without any encryption. Every message, every authentication credential, every consumer group offset travels in plaintext between producers, consumers, and brokers. When Kafka is deployed in a multi-tenant environment or carries data that crosses network boundaries, TLS is not optional. This article explains how to generate certificates, configure Kafka brokers for TLS, connect clients securely, and manage certificate rotation on a running cluster.

Kafka TLS architecture

Kafka uses TLS (SSL in Kafka’s own terminology, for historical reasons) in two places:

1. Client-to-broker: Producers and consumers connect to brokers over SSL listeners.
2. Inter-broker: Brokers replicate partitions between themselves — this traffic can also be encrypted.

Kafka uses Java’s standard KeyStore and TrustStore (JKS or PKCS12) for certificate management. Every broker needs a keystore containing its own certificate and key, and a truststore containing the CA that signs all certificates in the cluster.

Step 1: Create a Certificate Authority

For a Kafka cluster, creating a private CA is the standard approach. This CA signs both broker and (optionally) client certificates.

mkdir -p /etc/kafka/ssl/ca
cd /etc/kafka/ssl/ca

# Generate CA key and self-signed certificate
openssl req -new -x509 -keyout ca.key -out ca.crt -days 3650 \
  -passout pass:ca-password \
  -subj "/CN=Kafka-CA/O=example/C=US"

Keep ca.key secure — it signs all certificates in the cluster.

Step 2: Generate a broker certificate and keystore

Repeat for each broker, changing the CN to the broker’s hostname:

BROKER_HOSTNAME=kafka-broker1.example.com
KEYSTORE=/etc/kafka/ssl/kafka.broker1.keystore.jks
TRUSTSTORE=/etc/kafka/ssl/kafka.broker1.truststore.jks
KEYSTORE_PASS=keystore-password
KEY_PASS=key-password
TRUST_PASS=truststore-password

# Generate broker key and certificate in a keystore
keytool -genkey -noprompt \
  -alias broker1 \
  -dname "CN=${BROKER_HOSTNAME}, O=example, C=US" \
  -keystore ${KEYSTORE} \
  -keyalg RSA \
  -keysize 4096 \
  -validity 365 \
  -keypass ${KEY_PASS} \
  -storepass ${KEYSTORE_PASS}

# Export the certificate signing request
keytool -certreq -alias broker1 \
  -keystore ${KEYSTORE} \
  -file /tmp/broker1.csr \
  -storepass ${KEYSTORE_PASS}

# Sign the CSR with the CA
openssl x509 -req -CA /etc/kafka/ssl/ca/ca.crt \
  -CAkey /etc/kafka/ssl/ca/ca.key \
  -in /tmp/broker1.csr \
  -out /tmp/broker1-signed.crt \
  -days 365 \
  -CAcreateserial \
  -passin pass:ca-password \
  -extfile <(printf "subjectAltName=DNS:${BROKER_HOSTNAME},DNS:localhost")

# Import the CA into the keystore
keytool -import -noprompt -alias CARoot \
  -keystore ${KEYSTORE} \
  -file /etc/kafka/ssl/ca/ca.crt \
  -storepass ${KEYSTORE_PASS}

# Import the signed certificate into the keystore
keytool -import -noprompt -alias broker1 \
  -keystore ${KEYSTORE} \
  -file /tmp/broker1-signed.crt \
  -storepass ${KEYSTORE_PASS}

# Create the truststore with the CA certificate
keytool -import -noprompt -alias CARoot \
  -keystore ${TRUSTSTORE} \
  -file /etc/kafka/ssl/ca/ca.crt \
  -storepass ${TRUST_PASS}

Set permissions:

chown kafka:kafka /etc/kafka/ssl/kafka.broker1.keystore.jks /etc/kafka/ssl/kafka.broker1.truststore.jks
chmod 640 /etc/kafka/ssl/kafka.broker1.keystore.jks /etc/kafka/ssl/kafka.broker1.truststore.jks

Step 3: Configure the Kafka broker

Edit server.properties (usually /etc/kafka/server.properties or /opt/kafka/config/server.properties):

# Listeners — add an SSL listener alongside PLAINTEXT (or replace it)
listeners=PLAINTEXT://:9092,SSL://:9093
advertised.listeners=PLAINTEXT://kafka-broker1.example.com:9092,SSL://kafka-broker1.example.com:9093
listener.security.protocol.map=PLAINTEXT:PLAINTEXT,SSL:SSL

# SSL keystore (broker's certificate + key)
ssl.keystore.location=/etc/kafka/ssl/kafka.broker1.keystore.jks
ssl.keystore.******
ssl.key.******

# SSL truststore (CA certificate)
ssl.truststore.location=/etc/kafka/ssl/kafka.broker1.truststore.jks
ssl.truststore.******

# TLS version
ssl.enabled.protocols=TLSv1.2,TLSv1.3
ssl.protocol=TLSv1.3

# For production: remove weak ciphers
# ssl.cipher.suites=TLS_AES_256_GCM_SHA384,TLS_CHACHA20_POLY1305_SHA256,ECDHE-RSA-AES256-GCM-SHA384

# Require client (producer/consumer) certificates (mutual TLS)
# ssl.client.auth=required
# (or "requested" to make it optional)
ssl.client.auth=none

# Inter-broker communication security
inter.broker.listener.name=SSL
# (or PLAINTEXT if inter-broker traffic stays on a private network)

Restart the broker:

sudo systemctl restart kafka
sudo systemctl status kafka

Step 4: Verify the TLS port

openssl s_client -connect kafka-broker1.example.com:9093 \
  -servername kafka-broker1.example.com \
  -CAfile /etc/kafka/ssl/ca/ca.crt </dev/null 2>&1 \
  | openssl x509 -noout -subject -dates

Step 5: Configure producers and consumers

Java client

Create a client.properties file:

# Connect via SSL
security.protocol=SSL

# Truststore (to verify the broker certificate)
ssl.truststore.location=/etc/app/kafka.client.truststore.jks
ssl.truststore.******

# Keystore (required only for mutual TLS)
# ssl.keystore.location=/etc/app/kafka.client.keystore.jks
# ssl.keystore.******
# ssl.key.******

ssl.endpoint.identification.algorithm=https

Use it with the console consumer:

kafka-console-consumer.sh \
  --bootstrap-server kafka-broker1.example.com:9093 \
  --topic test \
  --consumer.config /etc/app/client.properties

Build the client truststore:

keytool -import -noprompt -alias CARoot \
  -keystore /etc/app/kafka.client.truststore.jks \
  -file /etc/kafka/ssl/ca/ca.crt \
  -storepass truststore-password

Python client (confluent-kafka)

from confluent_kafka import Producer, Consumer

conf = {
    'bootstrap.servers': 'kafka-broker1.example.com:9093',
    'security.protocol': 'SSL',
    'ssl.ca.location': '/etc/app/ca.crt',
    # For mutual TLS:
    # 'ssl.certificate.location': '/etc/app/client.crt',
    # 'ssl.key.location': '/etc/app/client.key',
    'ssl.endpoint.identification.algorithm': 'https',
}

producer = Producer(conf)
producer.produce('test-topic', key='key', value='hello over TLS')
producer.flush()

Spring Boot / Spring Kafka

In application.properties:

spring.kafka.bootstrap-servers=kafka-broker1.example.com:9093
spring.kafka.security.protocol=SSL
spring.kafka.ssl.trust-store-location=classpath:kafka.truststore.jks
spring.kafka.ssl.trust-store-******
spring.kafka.properties.ssl.endpoint.identification.algorithm=https

Step 6: Securing inter-broker replication

When inter.broker.listener.name=SSL, brokers authenticate each other using their own keystore/truststore. Since all broker certificates are signed by the same CA, and all truststores contain that CA, this works transparently. No additional configuration is needed beyond what is already set.

To verify inter-broker SSL is working:

kafka-broker-api-versions.sh \
  --bootstrap-server kafka-broker2.example.com:9093 \
  --command-config /etc/app/client.properties 2>&1 | head -20

Step 7: Certificate rotation on a running cluster

Kafka does not require downtime for certificate rotation — you rotate one broker at a time.

For each broker:

1. Generate a new signed certificate using the same CA (Steps 1–2, new keystore).
2. Copy the new keystore to the broker.
3. Update ssl.keystore.location if the filename changed (or overwrite the same file).
4. Perform a rolling restart: restart one broker while others continue serving clients.
5. Clients will fail over to other brokers during the brief restart.

For the entire cluster in sequence:

for broker in kafka-broker1 kafka-broker2 kafka-broker3; do
  echo "Rotating $broker..."
  ssh $broker "systemctl restart kafka"
  sleep 60  # wait for broker to rejoin before proceeding
  kafka-broker-api-versions.sh --bootstrap-server ${broker}.example.com:9093 \
    --command-config /etc/app/client.properties > /dev/null && echo "$broker OK"
done

SASL+TLS — combining authentication and encryption

TLS encrypts the channel; SASL authenticates the client identity. The most common combination for Kafka in production is SASL_SSL:

listeners=SASL_SSL://:9094
listener.security.protocol.map=SASL_SSL:SASL_SSL
sasl.enabled.mechanisms=SCRAM-SHA-512
sasl.mechanism.inter.broker.protocol=SCRAM-SHA-512

Client configuration:

security.protocol=SASL_SSL
sasl.mechanism=SCRAM-SHA-512
sasl.jaas.config=org.apache.kafka.common.security.scram.ScramLoginModule required \
  username="myuser" \
  ******;
ssl.truststore.location=/etc/app/kafka.client.truststore.jks
ssl.truststore.******

Troubleshooting

Error	Cause	Fix
SSL handshake failed	Truststore does not contain the broker’s CA	Import CA into client truststore
CERTIFICATE_UNKNOWN	Broker certificate expired	Rotate the broker certificate
Hostname verification failed	Certificate CN/SAN does not match broker hostname	Issue new cert with correct SAN
LEADER_NOT_AVAILABLE after cert rotation	Partition leadership election in progress	Wait 30-60 seconds and retry
java.io.IOException: Invalid keystore format	Wrong keystore type (PKCS12 vs JKS)	Specify -storetype PKCS12 in keytool or convert

Summary

Kafka TLS requires generating a CA, signing broker certificates into Java keystores, configuring ssl.keystore.* and ssl.truststore.* in server.properties, and updating clients to use the SSL port with a matching truststore. For inter-broker encryption, set inter.broker.listener.name=SSL. Certificate rotation is rolling and non-disruptive — restart brokers one at a time while clients fail over automatically. Combine TLS with SASL for full transport encryption plus identity authentication.