Securing Kafka with mTLS and SASL/SCRAM

Authentication is the first control plane for a Kafka cluster: until a broker can prove who is connecting, nothing else in your security model holds. This article is a practical, production-oriented treatment of the two authentication mechanisms most teams reach for—mutual TLS (mTLS) and SASL/SCRAM—and how to operate them. It sits within the broader Operations, Security & Observability strategy. Authentication only establishes identity; deciding what an identity may do is authorization, covered in Kafka ACLs and Authorization Best Practices.

Everything below assumes a modern Kafka cluster (3.x or 4.x). Note that Apache Kafka 4.0 removed ZooKeeper entirely—KRaft is the only metadata mode—and switched logging to Log4j2 (KIP-653). Both facts change the commands and configuration files you will actually run, and the draft conventions you may have inherited from older guides are wrong on current releases.

The Authentication Landscape: mTLS vs. SASL/SCRAM

mTLS and SASL/SCRAM solve authentication at different layers with different operational footprints.

mTLS operates at the transport layer (TLS 1.2/1.3), using X.509 certificates to authenticate both peers during the TLS handshake—before any Kafka protocol bytes are exchanged. The broker presents its certificate to the client, and with ssl.client.auth=required the client must present a certificate the broker trusts. This pushes identity management into a Public Key Infrastructure (PKI) such as HashiCorp Vault or cert-manager: strong, but operationally heavier because every principal is a certificate with an expiry.

SASL/SCRAM (Salted Challenge Response Authentication Mechanism, RFC 5802) operates at the application layer with a username and password. The password is never sent on the wire; a challenge-response exchange proves the client knows it. The broker stores only a salted, iterated hash (salt, iterations, StoredKey, ServerKey), so a leak of the stored credential does not directly yield the password. Kafka supports SCRAM-SHA-256 and SCRAM-SHA-512. Credentials are managed dynamically with no broker restart.

The two compose well. A common, strong pattern is SASL_SSL: TLS provides wire encryption and server authentication (the client confirms it reached the right broker), and SASL/SCRAM layers client authentication on top of the encrypted channel. Run SCRAM over TLS, never plaintext—SASL_PLAINTEXT exposes the SCRAM exchange and connection metadata.

The choice usually comes down to operational context. If you already run a robust PKI for service-to-service traffic, mTLS is a natural fit. If you manage many user-like clients with frequently rotating passwords, SCRAM is more practical because credentials change without touching certificates. For a hands-on certificate walkthrough see Step-by-Step Guide to Enabling mTLS on Kafka Brokers; for a platform path see Configuring SASL/SCRAM with Confluent Platform and Kubernetes.

Architecting a Certificate-Based mTLS Deployment

mTLS starts with a certificate hierarchy. At minimum you need a Certificate Authority (CA) trusted by every broker and client. Each broker needs a keystore (its private key and signed certificate) and a truststore (the CA certificate); each client needs the same. The broker certificate must carry the hostname or IP that clients dial in its Subject Alternative Name (SAN)—the modern TLS stack ignores the certificate Common Name for hostname verification, so a missing SAN entry fails the handshake even if the CN looks right.

A production PKI typically uses an offline root CA and an online intermediate CA that issues leaf certificates, so compromise of the issuing CA does not force you to redistribute root trust. Automate issuance and renewal: hand-managing certificates for a 12-broker cluster with hundreds of clients is an outage waiting to happen. cert-manager (Kubernetes) and Vault’s PKI secrets engine are the usual tools. The example below builds a CA and a broker keystore with openssl and keytool—fine for development or bootstrapping an internal CA.

# Generate a self-signed Certificate Authority (CA)
openssl req -new -x509 -keyout ca-key -out ca-cert -days 3650 \
  -subj "/CN=Kafka-Security-CA" -nodes

# Create a broker keystore and generate a key pair (include the SAN!)
keytool -keystore kafka.broker.keystore.jks -alias broker -validity 3650 -genkeypair \
  -keyalg RSA -dname "CN=kafka-broker-1, OU=Engineering, O=Acme" \
  -ext "SAN=DNS:kafka-broker-1,DNS:kafka-broker-1.internal" \
  -storepass changeit -keypass changeit

# Generate a Certificate Signing Request (CSR), preserving the SAN extension
keytool -keystore kafka.broker.keystore.jks -alias broker -certreq -file broker.csr \
  -ext "SAN=DNS:kafka-broker-1,DNS:kafka-broker-1.internal" \
  -storepass changeit

# Sign the CSR with your CA, copying the SAN into the signed cert
openssl x509 -req -CA ca-cert -CAkey ca-key -in broker.csr \
  -out broker-signed.crt -days 3650 -CAcreateserial \
  -extfile <(printf "subjectAltName=DNS:kafka-broker-1,DNS:kafka-broker-1.internal")

# Import the CA cert, then the signed broker cert, into the keystore
keytool -keystore kafka.broker.keystore.jks -alias CARoot -importcert -file ca-cert \
  -storepass changeit -noprompt
keytool -keystore kafka.broker.keystore.jks -alias broker -importcert -file broker-signed.crt \
  -storepass changeit -noprompt

# Create a truststore that trusts the CA (used by brokers and clients)
keytool -keystore kafka.truststore.jks -alias CARoot -importcert -file ca-cert \
  -storepass changeit -noprompt

A common, silent failure: omitting the SAN on the CSR or dropping it during signing. keytool’s old -genkey/-import aliases still work but -genkeypair/-importcert are the current spellings, and only an -ext SAN=... on both the request and the signing step gets the SAN into the leaf certificate.

Configure the broker’s server.properties to enable mTLS. ssl.client.auth=required mandates mutual authentication. Avoid requested: clients that present no certificate (or an invalid one) still connect, authenticated as the ANONYMOUS principal—a false sense of security that fails open.

# server.properties - mTLS configuration
listeners=SSL://:9093
advertised.listeners=SSL://kafka-broker-1:9093

# Broker keystore and truststore
ssl.keystore.location=/var/private/ssl/kafka.broker.keystore.jks
ssl.keystore.password=changeit
ssl.key.password=changeit
ssl.truststore.location=/var/private/ssl/kafka.truststore.jks
ssl.truststore.password=changeit

# Enforce mutual TLS
ssl.client.auth=required

# Optional: restrict to strong protocols and ciphers
ssl.enabled.protocols=TLSv1.2,TLSv1.3
ssl.cipher.suites=TLS_ECDHE_RSA_WITH_AES_256_GCM_SHA384,TLS_ECDHE_ECDSA_WITH_AES_256_GCM_SHA384

After a rolling restart, brokers accept only clients presenting a certificate chaining to the trusted CA. By default the principal is the full certificate distinguished name (e.g. CN=kafka-client,OU=Engineering,O=Acme); you can remap it with ssl.principal.mapping.rules if you want a shorter ACL subject. That principal is what you reference when defining ACLs, as detailed in Kafka ACLs and Authorization Best Practices.

Implementing SASL/SCRAM Authentication

SCRAM is the more dynamic option. On a KRaft cluster, SCRAM credentials live in the cluster metadata log, so you create, update, and delete users with no broker restart. (On Kafka 4.0+ there is no other option—ZooKeeper is gone.)

Create a SCRAM user with kafka-configs.sh. Use --bootstrap-server, not the removed --zookeeper flag. The iterations value is optional and defaults to 4096; raising it (e.g. 8192) increases the work factor against offline cracking. Because this command must itself authenticate to the cluster, point it at an admin --command-config.

# Create/update a SCRAM-SHA-512 credential for user "alice"
kafka-configs.sh --bootstrap-server kafka-broker-1:9094 --alter \
  --add-config 'SCRAM-SHA-512=[iterations=8192,password=my-secret-password]' \
  --entity-type users --entity-name alice \
  --command-config admin.properties

# Describe the user's credentials (mechanism + iterations only; no secrets are returned)
kafka-configs.sh --bootstrap-server kafka-broker-1:9094 --describe \
  --entity-type users --entity-name alice \
  --command-config admin.properties

--describe deliberately returns only the configured mechanism and iteration count—never the salt, StoredKey, or ServerKey—per the KIP-554 SCRAM API.

On the broker, configure a SASL-enabled listener and the SCRAM mechanism. The broker’s own credentials (used for inter-broker traffic and for the SCRAM login module) are supplied via the listener-scoped sasl.jaas.config. Running SCRAM over SASL_SSL keeps the exchange on an encrypted channel.

# server.properties - SASL/SCRAM over TLS
listeners=SASL_SSL://:9094
advertised.listeners=SASL_SSL://kafka-broker-1:9094
security.inter.broker.protocol=SASL_SSL
sasl.mechanism.inter.broker.protocol=SCRAM-SHA-512

# Enabled SASL mechanism for this listener
sasl.enabled.mechanisms=SCRAM-SHA-512

# Broker credentials (for inter-broker auth). Listener-scoped JAAS, no external jaas.conf needed.
listener.name.sasl_ssl.scram-sha-512.sasl.jaas.config=org.apache.kafka.common.security.scram.ScramLoginModule required \
  username="admin" \
  password="admin-secret";

# TLS for encryption + server authentication; no client cert required on this listener
ssl.keystore.location=/var/private/ssl/kafka.broker.keystore.jks
ssl.keystore.password=changeit
ssl.key.password=changeit
ssl.truststore.location=/var/private/ssl/kafka.truststore.jks
ssl.truststore.password=changeit
ssl.client.auth=none

# Principal "admin" must exist as a SCRAM user AND be a super user
super.users=User:admin

The broker’s admin identity must also exist as a SCRAM credential (created with the same kafka-configs.sh --add-config command) so brokers can authenticate to one another; otherwise inter-broker replication will fail to start.

The client configuration mirrors it:

# client.properties - SASL/SCRAM client
security.protocol=SASL_SSL
sasl.mechanism=SCRAM-SHA-512
ssl.truststore.location=/var/private/ssl/kafka.truststore.jks
ssl.truststore.password=changeit
sasl.jaas.config=org.apache.kafka.common.security.scram.ScramLoginModule required \
  username="alice" \
  password="my-secret-password";

Credential rotation is where SCRAM earns its keep. Re-running --add-config with a new password updates the stored credential without downtime. Because authentication is per-connection, existing connections keep working and new connections use the updated secret—roll your clients over to the new password, then the change is complete. (There is no built-in dual-secret window; Kafka stores one credential per mechanism per user, so coordinate the client cutover.) The canonical reference is the Apache Kafka Security documentation.

Operationalizing Credential Lifecycle Management

Whichever mechanism you choose, credential lifecycle is where most teams stumble.

For mTLS, the dominant failure mode is expiry. An expired broker certificate can partition the cluster; an expired client certificate is a hard authentication failure. Monitor and alert on certificate expiry well ahead of time, and prefer short-lived, automatically renewed certificates so a missed renewal is contained. For exporting and scraping these signals, see Monitoring Kafka with JMX, Prometheus, and Grafana.

For SASL/SCRAM, the work is password rotation and auditing. The script below rotates a SCRAM user with the confluent-kafka Python client, whose alter_user_scram_credentials admin API (KIP-554) is the correct, ZooKeeper-free way to manage credentials programmatically—it does not go through generic config alteration.

import secrets
import string

from confluent_kafka.admin import (
    AdminClient,
    ScramCredentialInfo,
    ScramMechanism,
    UserScramCredentialUpsertion,
)
from confluent_kafka import KafkaException


def rotate_scram_user(conf, username, iterations=8192, password_length=32):
    """Rotate a SCRAM-SHA-512 user's password via the broker SCRAM API."""
    admin = AdminClient(conf)

    alphabet = string.ascii_letters + string.digits + "!@#$%^&*"
    new_password = "".join(secrets.choice(alphabet) for _ in range(password_length))

    info = ScramCredentialInfo(ScramMechanism.SCRAM_SHA_512, iterations)
    upsertion = UserScramCredentialUpsertion(
        username, info, new_password.encode("utf-8")
    )

    futmap = admin.alter_user_scram_credentials([upsertion])
    futmap[username].result()  # raises KafkaException on failure
    print(f"Successfully rotated password for user {username}")
    return new_password


# Usage: the admin connection must authenticate as a super user.
conf = {
    "bootstrap.servers": "kafka-broker-1:9094",
    "security.protocol": "SASL_SSL",
    "sasl.mechanism": "SCRAM-SHA-512",
    "sasl.username": "admin",
    "sasl.password": "admin-secret",
    "ssl.ca.location": "/var/private/ssl/ca-cert",
}

try:
    new_pass = rotate_scram_user(conf, "alice")
    # Store new_pass in a secrets manager (e.g. Vault) and roll clients over.
except KafkaException as e:
    print(f"Failed to rotate password: {e}")
    raise

alter_user_scram_credentials accepts a list of UserScramCredentialUpsertion/UserScramCredentialDeletion objects and returns a dict of per-user futures; call .result() on each to surface errors.

For mTLS automation, integrate the PKI with a secrets manager and issue short-lived certificates. The example below issues a 24-hour client certificate from Vault’s PKI engine and packages it into a Java keystore for the client:

# Request a short-lived client certificate from Vault (TTL: 24h)
vault write -format=json pki/issue/kafka-client \
    common_name="my-kafka-client" \
    ttl="24h" > client-cert.json

# Extract the certificate and key, then build a PKCS#12 and import into a JKS
jq -r '.data.certificate' client-cert.json > client.crt
jq -r '.data.private_key' client-cert.json > client.key
openssl pkcs12 -export -in client.crt -inkey client.key \
    -out client.p12 -name kafka-client -password pass:changeit
keytool -importkeystore -srckeystore client.p12 -srcstoretype pkcs12 \
    -srcstorepass changeit \
    -destkeystore kafka.client.keystore.jks -deststorepass changeit

Note that -format=json is a global Vault flag and must precede the path/arguments, as shown.

Inter-Broker Security and Multi-Tenancy Considerations

Securing client-to-broker traffic is only half the job; inter-broker traffic needs equal protection. With mTLS, inter-broker replication naturally rides the same TLS listener and mutual authentication. With SCRAM, set security.inter.broker.protocol=SASL_SSL and sasl.mechanism.inter.broker.protocol explicitly, and ensure the broker principal is both a SCRAM user and listed in super.users. The super.users principals bypass ACL checks for cluster operations—keep that list tight.

In multi-tenant clusters you often need several mechanisms at once. Kafka supports multiple named listeners, each with its own security protocol, mapped via listener.security.protocol.map. A typical split: an internal listener using mTLS for service-to-service traffic, and an external listener using SASL/SCRAM for user-facing applications.

# Multi-listener configuration
listeners=INTERNAL://:9093,EXTERNAL://:9094
advertised.listeners=INTERNAL://kafka-broker-1.internal:9093,EXTERNAL://kafka-broker-1.external:9094
listener.security.protocol.map=INTERNAL:SSL,EXTERNAL:SASL_SSL
inter.broker.listener.name=INTERNAL

# Internal listener (mTLS)
listener.name.internal.ssl.client.auth=required
listener.name.internal.ssl.keystore.location=/var/private/ssl/internal.keystore.jks
listener.name.internal.ssl.keystore.password=changeit
listener.name.internal.ssl.truststore.location=/var/private/ssl/internal.truststore.jks
listener.name.internal.ssl.truststore.password=changeit

# External listener (SASL/SCRAM over TLS)
listener.name.external.sasl.enabled.mechanisms=SCRAM-SHA-512
listener.name.external.ssl.client.auth=none

Because mapped principals are normalized to User:<name>, identities from either listener are referenced the same way in ACLs—an mTLS certificate DN and a SCRAM username both become a User: subject.

Monitoring, Auditing, and Troubleshooting Authentication Failures

A secure cluster you cannot observe is a liability. Authentication failures are usually the first symptom of misconfiguration, expired credentials, or an attack, so expose and alert on them.

The relevant JMX metrics are the network/authentication counters, not a BrokerTopicMetrics counter (there is no FailedAuthenticationCountPerSec—it does not exist). Brokers and clients expose, per listener:

• failed-authentication-total / failed-authentication-rate
• successful-authentication-total / successful-authentication-rate
• failed-reauthentication-total / successful-reauthentication-total

On the broker these live under the socket-server/network metrics group, tagged by listener name; on clients they live under the producer-metrics/consumer-metrics group keyed by client-id. Browse the exact object names for your build with JConsole (point it at the broker’s JMX port), since the listener tag is part of the object name. A sustained rise in failed-authentication-rate on a listener warrants immediate investigation.

To debug handshakes, temporarily raise logging on org.apache.kafka.common.security to DEBUG. Kafka 3.x and 4.x use Log4j2 (KIP-653; 4.0 dropped Log4j 1.x entirely), so configure a logger and a rolling appender in log4j2.yaml, not the legacy log4j.properties:

# log4j2.yaml - scoped security debugging
Configuration:
  Appenders:
    RollingFile:
      - name: SecurityAppender
        fileName: "/var/log/kafka/security-debug.log"
        filePattern: "/var/log/kafka/security-debug.log.%d{yyyy-MM-dd-HH}"
        PatternLayout:
          pattern: "%d{ISO8601} [%t] %-5p %c - %m%n"
        Policies:
          TimeBasedTriggeringPolicy:
            modulate: true
            interval: 1
  Loggers:
    Logger:
      - name: org.apache.kafka.common.security
        level: DEBUG
        additivity: false
        AppenderRef:
          ref: SecurityAppender

For mTLS, the usual culprits are an expired certificate, a hostname missing from the SAN, or a CA the broker does not trust. openssl s_client inspects the handshake from the client side:

# Test mTLS connectivity to a broker
openssl s_client -connect kafka-broker-1:9093 \
  -cert client.crt -key client.key \
  -CAfile ca-cert -state

Check the verify result (Verify return code: 0 (ok)) and that the certificate chain and SAN match the host you dialed; -state traces the handshake states.

For SCRAM, the most common failures are a wrong sasl.mechanism, a JAAS typo, or a user that does not exist in the metadata log. Broker logs surface a SaslAuthenticationException with a specific reason. Principal names are case-sensitive, so confirm the username matches exactly. The Kafka Protocol Guide documents the SASL handshake frames if you need to read a packet capture.

Conclusion

Authentication is only the first gate: once a principal is authenticated, authorization decides what it can touch, and audit logging records what it did. The identities you establish here—a certificate DN for mTLS or a username for SCRAM—are exactly the subjects you reference in ACLs. On current Kafka (4.0+), the operational details that matter are concrete: manage SCRAM with --bootstrap-server against the KRaft metadata log, put hostnames in the certificate SAN, run SCRAM over SASL_SSL, watch the failed-authentication-rate metric per listener, and debug with log4j2.yaml. Get those right, pair them with automated certificate or password rotation, and you have a cluster that is both secure and maintainable at scale.