TLS and mTLS

The Kafka connector opens two listeners: 9092 (plaintext) and 9093 (TLS). This guide covers the TLS listener — server-authentication TLS, mutual TLS, and how the client certificate's common name becomes the authenticated principal when SASL isn't used.

The 9093 TLS listener

The TLS listener has no certificate configuration of its own — it reuses the server's global Security block (Cert/Key/Ca), the same block every other TCP connector (gRPC, REST, STOMP, AMQP, RabbitMQ) shares. There is no Connectors.Kafka.Tls*Cert field to set.

The mode is auto-derived, not a flag: Cert + Key alone is server-authentication TLS; adding Ca promotes it to mTLS. See Security → TLS / mTLS for the full field reference and Docker/Helm examples.

Server-authentication TLS

With Cert + Key set (no Ca), the server presents its certificate and the minimum negotiated protocol is TLS 1.2; no client certificate is requested. The client authenticates separately — with SASL/PLAIN, SCRAM, or OAUTHBEARER over the now-encrypted channel — see Authentication.

kcat -b localhost:9093 -L \
  -X security.protocol=SSL \
  -X ssl.ca.location=/path/to/ca.pem

import (
    "crypto/tls"
    "crypto/x509"
    "os"

    "github.com/twmb/franz-go/pkg/kgo"
)

caPEM, _ := os.ReadFile("/path/to/ca.pem")
pool := x509.NewCertPool()
pool.AppendCertsFromPEM(caPEM)

cl, err := kgo.NewClient(
    kgo.SeedBrokers("localhost:9093"),
    kgo.DialTLSConfig(&tls.Config{RootCAs: pool}),
)

security.protocol=SSL
ssl.truststore.location=/path/to/truststore.jks
ssl.truststore.password=changeit

Mutual TLS — certificate identity

Adding Ca to the Security block promotes the listener to mTLS: the server now requires and verifies a client certificate (RequireAndVerifyClientCert) using that CA pool, on every connection to 9093. A client that fails the chain check never completes the handshake.

When the connection carries no SASL layer, the connector reads the verified certificate's leaf Subject.CommonName as the authenticated principal — the same identity the Casbin ACL then authorizes against (see Authentication). An empty CN, or a certificate whose chain didn't verify, never becomes a principal.

kcat -b localhost:9093 -L \
  -X security.protocol=SSL \
  -X ssl.ca.location=/path/to/ca.pem \
  -X ssl.key.location=/path/to/client.key \
  -X ssl.certificate.location=/path/to/client.crt

clientCert, _ := tls.LoadX509KeyPair("/path/to/client.crt", "/path/to/client.key")

cl, err := kgo.NewClient(
    kgo.SeedBrokers("localhost:9093"),
    kgo.DialTLSConfig(&tls.Config{
        RootCAs:      pool, // the CA that signed the SERVER certificate
        Certificates: []tls.Certificate{clientCert},
    }),
)

security.protocol=SSL
ssl.truststore.location=/path/to/truststore.jks
ssl.truststore.password=changeit
ssl.keystore.location=/path/to/client-keystore.jks
ssl.keystore.password=changeit
ssl.key.password=changeit

Related