# TLS and mTLS (/connectors/kafka/how-to/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.

<Callout type="warn">
  **`Port` stays required, even if you only want TLS.** A TLS-only configuration — `TlsPort` set,
  `Port` left empty — is **rejected**: the TLS accept path isn't wired independently of the
  plaintext one yet, so `Port` must stay set (and differ from `TlsPort`). Restrict access to the
  plaintext listener at your network layer (firewall rules, a `ClusterIP`-only Service) instead of
  by unsetting `Port`.
</Callout>

## The 9093 TLS listener [#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.

| Setting        | Env var                                         | Default | Notes                                                                       |
| -------------- | ----------------------------------------------- | ------- | --------------------------------------------------------------------------- |
| Plaintext port | `CONNECTORS_KAFKA_PORT`                         | `9092`  | Must stay set — see the callout above.                                      |
| TLS port       | `CONNECTORS_KAFKA_TLS_PORT`                     | `9093`  | Set to `""` (empty) to disable the TLS listener; otherwise must be 1–65535. |
| Server cert    | `SECURITY_CERT_DATA` / `SECURITY_CERT_FILENAME` | `""`    | Required for TLS and mTLS.                                                  |
| Server key     | `SECURITY_KEY_DATA` / `SECURITY_KEY_FILENAME`   | `""`    | Required for TLS and mTLS.                                                  |
| Client CA      | `SECURITY_CA_DATA` / `SECURITY_CA_FILENAME`     | `""`    | Presence **promotes the mode to mTLS** — see below.                         |

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](/configure/reference/security#tls--mtls) for the full field
reference and Docker/Helm examples.

<Callout type="info">
  **If you expose 9093 externally, the certificate must cover `AdvertisedHost`.** The connector
  serves one certificate with no per-SNI selection — a certificate whose SAN covers only the
  in-cluster Service DNS fails TLS hostname verification for an external client connecting
  through the advertised address.
</Callout>

## Server-authentication TLS [#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](/connectors/kafka/how-to/authentication).

<Tabs groupId="language" items="['kcat', 'Go', 'Java']">
  <Tab value="kcat">
    ```bash
    kcat -b localhost:9093 -L \
      -X security.protocol=SSL \
      -X ssl.ca.location=/path/to/ca.pem
    ```
  </Tab>

  <Tab value="Go">
    ```go
    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}),
    )
    ```
  </Tab>

  <Tab value="Java">
    ```properties
    security.protocol=SSL
    ssl.truststore.location=/path/to/truststore.jks
    ssl.truststore.password=changeit
    ```
  </Tab>
</Tabs>

## Mutual TLS — certificate identity [#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
&#x2A;*leaf `Subject.CommonName`** as the authenticated principal — the same identity the Casbin ACL
then authorizes against (see [Authentication](/connectors/kafka/how-to/authentication)). An
empty CN, or a certificate whose chain didn't verify, never becomes a principal.

<Tabs groupId="language" items="['kcat', 'Go', 'Java']">
  <Tab value="kcat">
    ```bash
    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
    ```
  </Tab>

  <Tab value="Go">
    ```go
    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},
        }),
    )
    ```
  </Tab>

  <Tab value="Java">
    ```properties
    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
    ```
  </Tab>
</Tabs>

<Callout type="warn">
  **mTLS and SASL don't stack.** Only one identity source wins per connection: the certificate CN
  is used **only when SASL isn't required**. If the listener also has `Credentials` configured,
  the SASL identity takes precedence and the CN is never consulted — see
  [Authentication](/connectors/kafka/how-to/authentication).
</Callout>

## Related [#related]

<Cards>
  <Card title="Authentication" href="/connectors/kafka/how-to/authentication" description="Authenticate Kafka clients to KubeMQ — SASL/PLAIN and SCRAM, OAUTHBEARER/OIDC federated tokens, mTLS client certificates, and the ACL model that authorizes each request." />

  <Card title="Configuration" href="/connectors/kafka/concepts/configuration" description="How the Kafka connector is enabled, ported, and secured — the opt-in CONNECTORS_KAFKA_ENABLE flag and the 9092/9093 listeners." />

  <Card title="Kafka settings reference" href="/configure/reference/connectors#kafka" description="The full field-by-field Connectors.Kafka settings, including the Port/TlsPort pairing rule." />
</Cards>