PLAIN¶

This example configures the gateway to accept client connections using SASL/SCRAM-SHA-512 and swap credentials to SASL/PLAIN when authenticating to Confluent Cloud. This is a common pattern when migrating from MSK (which supports SASL/SCRAM) to Confluent Cloud.

Overview¶

The switchover follows a three-state pattern:

State	YAML	Description
Initial	`gateway_init.yaml`	Two routes: the main client route passes SCRAM traffic through to MSK; the pre-registration route is backed by Confluent Cloud for registering SCRAM credentials
Fenced	`gateway_fenced.yaml`	Client route is fenced; pre-registration route remains available
Switchover	`gateway_switchover.yaml`	Single client route with auth swap (SCRAM → SASL/PLAIN) routing to Confluent Cloud

The client configuration does not change across any state transition. The gateway transparently swaps SCRAM credentials for SASL/PLAIN when authenticating to Confluent Cloud.

Required before the fenced state: All SCRAM users must be pre-registered via the dedicated pre-registration route while the gateway is in the init state. KCP will apply the fenced state as part of migration — registration cannot happen after that. Failure to pre-register users means clients will be unable to authenticate after switchover.

Note: SCRAM passwords must not contain [ or ] characters — these conflict with the kafka-configs.sh bracket syntax used during registration.

See SCRAM Registration below for instructions.

Environment Variables¶

Variable	Description
`KUBECTL_NAMESPACE`	Kubernetes namespace for the gateway
`JKS_PASSWORD`	Password for JKS truststores
`JAVA_HOME`	Path to JDK installation
`SASL_PLAIN_USERNAME`	Confluent Cloud API key
`SASL_PLAIN_PASSWORD`	Confluent Cloud API secret
`SCRAM_USERNAME`	Client SCRAM username
`SCRAM_PASSWORD`	Client SCRAM password (must not contain `[` or `]`)
`SCRAM_ADMIN_USERNAME`	Admin SCRAM username for credential registration
`SCRAM_ADMIN_PASSWORD`	Admin SCRAM password

Prerequisites¶

CFK operator installed on Kubernetes cluster
kubectl configured to access the cluster
MSK cluster with SASL/SCRAM-SHA-512 and TLS (port 9096)
Confluent Cloud cluster with SASL/PLAIN credentials (API key/secret)
HashiCorp Vault accessible from the cluster
JDK installation (for truststore generation)

Secrets Setup¶

Secret	Purpose
`msk-tls`	JKS truststore for verifying MSK's server certificate (init and fenced states)
`tls`	JKS truststore for verifying Confluent Cloud's server certificate (all states — used by the pre-registration route in init/fenced, and the main route in switchover)
`vault-config`	Vault address, auth token, and path configuration for the credential store (all states)
`scram-admin-credentials`	SCRAM admin username and password used by the gateway to authenticate SCRAM registration requests (all states)
`plain-jaas`	JAAS config template for gateway-to-CCloud SASL/PLAIN authentication (all states)

1. MSK TLS Truststore¶

Create a truststore for verifying MSK's TLS certificate:

cp $JAVA_HOME/lib/security/cacerts msk-truststore.jks
echo "jksPassword=${JKS_PASSWORD}" > msk-jksPassword.txt

kubectl create secret generic msk-tls \
  --from-file=truststore.jks=msk-truststore.jks \
  --from-file=jksPassword.txt=msk-jksPassword.txt \
  -n ${KUBECTL_NAMESPACE}

2. Confluent Cloud TLS Truststore¶

Create a truststore for verifying Confluent Cloud's TLS certificate:

cp $JAVA_HOME/lib/security/cacerts ccloud-truststore.jks
echo "jksPassword=${JKS_PASSWORD}" > ccloud-jksPassword.txt

kubectl create secret generic tls \
  --from-file=truststore.jks=ccloud-truststore.jks \
  --from-file=jksPassword.txt=ccloud-jksPassword.txt \
  -n ${KUBECTL_NAMESPACE}

3. Vault Configuration¶

Create the Vault config secret. Vault must be accessible from the cluster.

kubectl create secret generic vault-config \
  --from-literal=address=<vault-address> \
  --from-literal=authToken=<vault-auth-token> \
  --from-literal=prefixPath=secret/ \
  --from-literal=separator=/ \
  -n ${KUBECTL_NAMESPACE}

4. SCRAM Admin Credentials¶

Create the SCRAM admin credentials secret:

kubectl create secret generic scram-admin-credentials \
  --from-literal=username="${SCRAM_ADMIN_USERNAME}" \
  --from-literal=password="${SCRAM_ADMIN_PASSWORD}" \
  -n ${KUBECTL_NAMESPACE}

5. JAAS Config Template¶

Create the JAAS config template for gateway-to-CCloud SASL/PLAIN authentication:

kubectl create secret generic plain-jaas \
  --from-literal=plain-jaas.conf='org.apache.kafka.common.security.plain.PlainLoginModule required username="%s" password="%s";' \
  -n ${KUBECTL_NAMESPACE}

6. Vault Credential Mappings¶

Pre-populate Vault with SCRAM-to-PLAIN credential mappings for each user:

vault kv put secret/${SCRAM_USERNAME} value="${SASL_PLAIN_USERNAME}/${SASL_PLAIN_PASSWORD}"
vault kv put secret/${SCRAM_ADMIN_USERNAME} value="${SASL_PLAIN_USERNAME}/${SASL_PLAIN_PASSWORD}"

These mappings are used: - Init state (pre-registration route): When the admin connects to the pre-registration route with SCRAM credentials to register a user, the gateway swaps to SASL/PLAIN to authenticate to Confluent Cloud - Switchover state: When clients connect, the gateway swaps their SCRAM credentials for SASL/PLAIN to authenticate to Confluent Cloud

These Vault entries must exist before deploying the init state — the pre-registration route needs the admin's mapping to authenticate to Confluent Cloud on first connection.

Gateway YAMLs¶

gateway_init.yaml¶

Two routes and two streaming domains:

Main client route: Normal SCRAM clients connect here. Traffic is routed to MSK with auth: passthrough (no credential transformation)
Pre-registration route: A dedicated route backed by Confluent Cloud where SCRAM credentials are registered via kafka-configs.sh. Uses auth: swap (SCRAM → SASL/PLAIN) so the gateway can authenticate to CCloud. The gateway intercepts AlterUserScramCredentials requests and stores SCRAM verifiers in Vault

Every user that needs to authenticate during switchover must be pre-registered via the pre-registration route before applying the fenced state.

apiVersion: platform.confluent.io/v1beta1
kind: Gateway
metadata:
  labels:
    app.kubernetes.io/name: confluent-for-kubernetes
    app.kubernetes.io/managed-by: kustomize
  name: migration-gateway
spec:
  replicas: 3
  image:
    application: <gateway-image>
    init: confluentinc/confluent-init-container:3.1.0
  podTemplate:
    envVars:
      - name: GATEWAY_ROOT_LOG_LEVEL
        value: DEBUG  # Can be changed to INFO for production
  secretStores:
    - name: vault-store
      provider:
        type: Vault
        configSecretRef: vault-config
  streamingDomains:
    - name: source-kafka-cluster
      type: kafka
      kafkaCluster:
        bootstrapServers:
          - id: SCRAM
            # Replace with your MSK bootstrap server (SASL_SSL port, typically 9096)
            endpoint: SASL_SSL://<msk-bootstrap-server>:9096
            tls:
              secretRef: msk-tls
        nodeIdRanges:
          - name: pool-1
            start: 1
            end: 3
    - name: confluent-cloud
      type: kafka
      kafkaCluster:
        bootstrapServers:
          - id: SASL_PLAIN
            # Replace with your Confluent Cloud bootstrap server
            endpoint: SASL_SSL://<ccloud-bootstrap-server>:9092
            tls:
              secretRef: tls
        nodeIdRanges:
          - name: pool-1
            start: 0
            end: 17
  routes:
    - name: migration-route
      endpoint: <gateway-lb-hostname>:9595
      brokerIdentificationStrategy:
        type: port
      streamingDomain:
        name: source-kafka-cluster
        bootstrapServerId: SCRAM
      security:
        auth: passthrough
    - name: scram-registration-route
      endpoint: <gateway-lb-hostname>:9599
      brokerIdentificationStrategy:
        type: port
      streamingDomain:
        name: confluent-cloud
        bootstrapServerId: SASL_PLAIN
      security:
        auth: swap
        secretStore: vault-store
        client:
          authentication:
            type: scram
            scram:
              alterScramCredentials: true
              admin:
                secretRef: scram-admin-credentials
        cluster:
          authentication:
            type: plain
            jaasConfigPassThrough:
              secretRef: plain-jaas
  externalAccess:
    type: loadBalancer
    loadBalancer:
      domain: <gateway-lb-hostname>

gateway_fenced.yaml¶

The client route on port 9595 is fenced with errorCode: BROKER_NOT_AVAILABLE. The registration route on port 9599 is retained.

Note: With kcp migration you do not author or supply this file. kcp derives the fenced state at cutover from the live initial CR, injecting the fence block below onto the route(s) named in spec.gateway.fence.routes. The YAML here is shown only to illustrate what the fenced state looks like — it is exactly what kcp applies.

apiVersion: platform.confluent.io/v1beta1
kind: Gateway
metadata:
  labels:
    app.kubernetes.io/name: confluent-for-kubernetes
    app.kubernetes.io/managed-by: kustomize
  name: migration-gateway
spec:
  replicas: 3
  image:
    application: <gateway-image>
    init: confluentinc/confluent-init-container:3.1.0
  podTemplate:
    envVars:
      - name: GATEWAY_ROOT_LOG_LEVEL
        value: DEBUG  # Can be changed to INFO for production
  secretStores:
    - name: vault-store
      provider:
        type: Vault
        configSecretRef: vault-config
  streamingDomains:
    - name: source-kafka-cluster
      type: kafka
      kafkaCluster:
        bootstrapServers:
          - id: SCRAM
            # Replace with your MSK bootstrap server (SASL_SSL port, typically 9096)
            endpoint: SASL_SSL://<msk-bootstrap-server>:9096
            tls:
              secretRef: msk-tls
        nodeIdRanges:
          - name: pool-1
            start: 1
            end: 3
    - name: confluent-cloud
      type: kafka
      kafkaCluster:
        bootstrapServers:
          - id: SASL_PLAIN
            # Replace with your Confluent Cloud bootstrap server
            endpoint: SASL_SSL://<ccloud-bootstrap-server>:9092
            tls:
              secretRef: tls
        nodeIdRanges:
          - name: pool-1
            start: 0
            end: 17
  routes:
    - name: migration-route
      endpoint: <gateway-lb-hostname>:9595
      fence:
        scope: ALL
        errorCode: BROKER_NOT_AVAILABLE
      brokerIdentificationStrategy:
        type: port
      streamingDomain:
        name: source-kafka-cluster
        bootstrapServerId: SCRAM
      security:
        auth: passthrough
    - name: scram-registration-route
      endpoint: <gateway-lb-hostname>:9599
      brokerIdentificationStrategy:
        type: port
      streamingDomain:
        name: confluent-cloud
        bootstrapServerId: SASL_PLAIN
      security:
        auth: swap
        secretStore: vault-store
        client:
          authentication:
            type: scram
            scram:
              alterScramCredentials: true
              admin:
                secretRef: scram-admin-credentials
        cluster:
          authentication:
            type: plain
            jaasConfigPassThrough:
              secretRef: plain-jaas
  externalAccess:
    type: loadBalancer
    loadBalancer:
      domain: <gateway-lb-hostname>

gateway_switchover.yaml¶

Single route on port 9595 with auth: swap (SCRAM → SASL/PLAIN) routing to Confluent Cloud. Only the CCloud streaming domain remains.

apiVersion: platform.confluent.io/v1beta1
kind: Gateway
metadata:
  labels:
    app.kubernetes.io/name: confluent-for-kubernetes
    app.kubernetes.io/managed-by: kustomize
  name: migration-gateway
spec:
  replicas: 3
  image:
    application: <gateway-image>
    init: confluentinc/confluent-init-container:3.1.0
  podTemplate:
    envVars:
      - name: GATEWAY_ROOT_LOG_LEVEL
        value: DEBUG  # Can be changed to INFO for production
  secretStores:
    - name: vault-store
      provider:
        type: Vault
        configSecretRef: vault-config
  streamingDomains:
    - name: confluent-cloud
      type: kafka
      kafkaCluster:
        bootstrapServers:
          - id: SASL_PLAIN
            # Replace with your Confluent Cloud bootstrap server
            endpoint: SASL_SSL://<ccloud-bootstrap-server>:9092
            tls:
              secretRef: tls
        nodeIdRanges:
          - name: pool-1
            start: 0
            end: 17
  routes:
    - name: migration-route
      endpoint: <gateway-lb-hostname>:9595
      brokerIdentificationStrategy:
        type: port
      streamingDomain:
        name: confluent-cloud
        bootstrapServerId: SASL_PLAIN
      security:
        auth: swap
        secretStore: vault-store
        client:
          authentication:
            type: scram
            scram:
              alterScramCredentials: true
              admin:
                secretRef: scram-admin-credentials
        cluster:
          authentication:
            type: plain
            jaasConfigPassThrough:
              secretRef: plain-jaas
  externalAccess:
    type: loadBalancer
    loadBalancer:
      domain: <gateway-lb-hostname>

SCRAM Registration¶

Before clients can authenticate, their SCRAM credentials must be registered with the gateway.

Create a scram-admin.properties file:

security.protocol=SASL_PLAINTEXT
sasl.mechanism=SCRAM-SHA-512
sasl.jaas.config=org.apache.kafka.common.security.scram.ScramLoginModule required username="${SCRAM_ADMIN_USERNAME}" password="${SCRAM_ADMIN_PASSWORD}";

Register each user via the registration route on port 9599:

kafka-configs.sh \
  --bootstrap-server <gateway-lb-hostname>:9599 \
  --command-config scram-admin.properties \
  --alter \
  --add-config "SCRAM-SHA-512=[iterations=8192,password=${SCRAM_PASSWORD}]" \
  --entity-type users \
  --entity-name ${SCRAM_USERNAME}

Important: SCRAM passwords must not contain [ or ] characters (these conflict with the bracket syntax used by kafka-configs.sh).

Client Properties¶

Clients connect using SCRAM credentials:

security.protocol=SASL_PLAINTEXT
sasl.mechanism=SCRAM-SHA-512
sasl.jaas.config=org.apache.kafka.common.security.scram.ScramLoginModule required username="${SCRAM_USERNAME}" password="${SCRAM_PASSWORD}";

These properties remain unchanged across all three states.

Client Behaviour During State Transitions¶

When the gateway is fenced, BROKER_NOT_AVAILABLE is returned during the SASL handshake. kafka-console-consumer.sh treats this as a fatal IllegalSaslStateException and exits — it must be restarted after applying the switchover state. kafka-console-producer.sh stays alive and reconnects automatically. Production clients with proper retry logic behave like the producer.