Switchover: SASL/SCRAM → OAuth¶

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

Overview¶

The gateway transitions through three states by sequentially applying Custom Resource updates:

State	Description
Init	Two routes: the main client route passes SCRAM traffic through to MSK; the pre-registration route is backed by Confluent Cloud and stores SCRAM verifiers in Vault
Fenced	Client route fenced (returns `BROKER_NOT_AVAILABLE`), pre-registration route retained
Switchover	Single client route with SCRAM→OAuth credential swap to Confluent Cloud

The client configuration remains unchanged throughout all transitions. The gateway transparently swaps SCRAM credentials for OAuth 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
`JKS_PASSWORD`	Password for JKS truststores
`JAVA_HOME`	Path to JDK home (for extracting cacerts)
`OAUTH_CLIENT_ID`	OAuth client ID for gateway-to-CCloud authentication
`OAUTH_CLIENT_SECRET`	OAuth client secret for gateway-to-CCloud authentication
`CCLOUD_LOGICAL_CLUSTER`	Confluent Cloud logical cluster ID (e.g. `lkc-xxxxx`)
`CCLOUD_IDENTITY_POOL_ID`	Confluent Cloud identity pool ID (e.g. `pool-xxxxx`)
`SCRAM_USERNAME`	SASL/SCRAM username for client
`SCRAM_PASSWORD`	SASL/SCRAM password for client (must not contain `[` or `]`)
`SCRAM_ADMIN_USERNAME`	SCRAM admin username for registration
`SCRAM_ADMIN_PASSWORD`	SCRAM admin password for registration

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)
`oauth-jaas`	OAuth JAAS config for gateway-to-CCloud authentication (all states)

Create MSK truststore:

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}

Create Confluent Cloud truststore:

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}

Create Vault configuration (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}

Create SCRAM admin credentials:

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

Create OAuth JAAS configuration:

Important: The secret key must be oauth-jass.conf (single 'a'), not oauth-jaas.conf — this is a quirk in the CFK operator's validation code.

kubectl create secret generic oauth-jaas \
  --from-literal=oauth-jass.conf="org.apache.kafka.common.security.oauthbearer.OAuthBearerLoginModule required clientId=\"%s\" clientSecret=\"%s\" extension_logicalCluster=\"${CCLOUD_LOGICAL_CLUSTER}\" extension_identityPoolId=\"${CCLOUD_IDENTITY_POOL_ID}\";" \
  -n ${KUBECTL_NAMESPACE}

Pre-populate Vault with SCRAM-to-OAuth credential mappings:

vault kv put secret/${SCRAM_USERNAME} value="${OAUTH_CLIENT_ID}/${OAUTH_CLIENT_SECRET}"
vault kv put secret/${SCRAM_ADMIN_USERNAME} value="${OAUTH_CLIENT_ID}/${OAUTH_CLIENT_SECRET}"

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: port 9595 passthrough to MSK, port 9599 registration route to Confluent Cloud with SCRAM→OAuth swap. Two streaming domains (MSK and Confluent Cloud).

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
      - name: GATEWAY_OPTS
        # Replace with your token endpoint URI (must match tokenEndpointUri below)
        value: "-Dorg.apache.kafka.sasl.oauthbearer.allowed.urls=<token-endpoint-uri>"
  secretStores:
    - name: vault-store
      provider:
        type: Vault
        configSecretRef: vault-config
  streamingDomains:
    - name: source-kafka-cluster
      type: kafka
      kafkaCluster:
        bootstrapServers:
          - id: SCRAM
            # MSK SCRAM uses SASL_SSL port 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: OAUTH
            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: OAUTH
      security:
        auth: swap
        secretStore: vault-store
        client:
          authentication:
            type: scram
            scram:
              alterScramCredentials: true
              admin:
                secretRef: scram-admin-credentials
        cluster:
          authentication:
            type: oauth
            jaasConfigPassThrough:
              secretRef: oauth-jaas
            oauthSettings:
              # Must match GATEWAY_OPTS value above
              tokenEndpointUri: <token-endpoint-uri>
  externalAccess:
    type: loadBalancer
    loadBalancer:
      domain: <gateway-lb-hostname>

gateway_fenced.yaml¶

Client route fenced (returns BROKER_NOT_AVAILABLE during SASL handshake), registration route retained. Two streaming domains.

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
      - name: GATEWAY_OPTS
        # Replace with your token endpoint URI (must match tokenEndpointUri below)
        value: "-Dorg.apache.kafka.sasl.oauthbearer.allowed.urls=<token-endpoint-uri>"
  secretStores:
    - name: vault-store
      provider:
        type: Vault
        configSecretRef: vault-config
  streamingDomains:
    - name: source-kafka-cluster
      type: kafka
      kafkaCluster:
        bootstrapServers:
          - id: SCRAM
            # MSK SCRAM uses SASL_SSL port 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: OAUTH
            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: OAUTH
      security:
        auth: swap
        secretStore: vault-store
        client:
          authentication:
            type: scram
            scram:
              alterScramCredentials: true
              admin:
                secretRef: scram-admin-credentials
        cluster:
          authentication:
            type: oauth
            jaasConfigPassThrough:
              secretRef: oauth-jaas
            oauthSettings:
              # Must match GATEWAY_OPTS value above
              tokenEndpointUri: <token-endpoint-uri>
  externalAccess:
    type: loadBalancer
    loadBalancer:
      domain: <gateway-lb-hostname>

gateway_switchover.yaml¶

Single route with SCRAM→OAuth credential swap to Confluent Cloud. Single streaming domain (Confluent Cloud only). Note that <token-endpoint-uri> appears in two places and both must be set to the same value.

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
      - name: GATEWAY_OPTS
        # Replace with your token endpoint URI (must match tokenEndpointUri below)
        value: "-Dorg.apache.kafka.sasl.oauthbearer.allowed.urls=<token-endpoint-uri>"
  secretStores:
    - name: vault-store
      provider:
        type: Vault
        configSecretRef: vault-config
  streamingDomains:
    - name: confluent-cloud
      type: kafka
      kafkaCluster:
        bootstrapServers:
          - id: OAUTH
            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: OAUTH
      security:
        auth: swap
        secretStore: vault-store
        client:
          authentication:
            type: scram
            scram:
              alterScramCredentials: true
              admin:
                secretRef: scram-admin-credentials
        cluster:
          authentication:
            type: oauth
            jaasConfigPassThrough:
              secretRef: oauth-jaas
            oauthSettings:
              # Must match GATEWAY_OPTS value above
              tokenEndpointUri: <token-endpoint-uri>
  externalAccess:
    type: loadBalancer
    loadBalancer:
      domain: <gateway-lb-hostname>

SCRAM Registration¶

Before clients can authenticate during the switchover state, their SCRAM credentials must be registered with the gateway via the dedicated pre-registration route.

Create scram-admin.properties:

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}";

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

Register SCRAM user:

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}

All users must be pre-registered before transitioning to the fenced state.

Client Properties¶

Client configuration remains unchanged throughout all state transitions:

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}";

Client Behaviour During State Transitions¶

Fenced state: Clients receive BROKER_NOT_AVAILABLE during the SASL handshake. Consumers exit with a fatal IllegalSaslStateException and must be restarted after switchover. Producers stay alive and automatically reconnect after switchover.
Switchover state: Producers reconnect automatically. Consumers must be restarted (with the same configuration).