Quick Start for Multi-Tenancy in Confluent Private Cloud

Multi-Tenancy in Confluent Private Cloud lets you run multiple isolated Logical Kafka Clusters (LKCs) on a single Physical Kafka Cluster (PKC). Each LKC has its own quotas, limits, and access controls.

In this quick start, you set up Multi-Tenancy on a local Kubernetes cluster by using Confluent for Kubernetes (CFK). You deploy one PKC, create two LKCs on it, and confirm that each LKC administrator can see only their own LKC’s data. For an introduction to PKCs, LKCs, and the roles that manage them, see the Multi-Tenancy Overview page.

The quick start takes 45 to 60 minutes.

When you finish, you have the following running in the confluent namespace:

• CFK, which manages the other components

• OpenLDAP, the identity provider for all users

• One KRaftController

• Three Kafka brokers, which form the PKC (spec.multiTenant.enabled)

• One KafkaRestClass named default

• One ControlCenter, with Prometheus and Alertmanager

• Two LogicalKafkaCluster resources, payments and analytics (the LKCs)

• Four ConfluentRolebinding resources, which grant the OrganizationAdmin and TenantAdmin roles

• One kafka-client pod, which runs the Kafka command-line tools

Important

This quick start is a learning sandbox. It uses demo passwords and an auto-generated certificate authority (CA), and it doesn’t configure production-grade security. To plan a production deployment, see Next steps.

Prerequisites

Before you begin, make sure you have the following:

• Docker with at least 16 GB of memory and six CPUs allocated to it. kind runs each Kubernetes node as a Docker container, so the five nodes that this quick start creates (one control plane and four workers) share these resources. This quick start covers only kind. To use another Kubernetes environment, adapt Step 1 and the load-balancer handling in Step 6.

• Command-line tools: kind, kubectl, helm, git, and openssl. For kind, see the kind installation guide.

• Network access to Docker Hub for the container images, to packages.confluent.io for the CFK Helm chart, and to GitHub for the examples repository.

• A web browser and sudo access on your machine. In Step 7, you add an /etc/hosts entry so that your browser can open Confluent Control Center.

The quick start deploys Confluent Platform 8.4, CFK 3.4, and Confluent Control Center 2.7, the first releases that support Multi-Tenancy. You pull these images in the steps that follow.

This quick start doesn’t require a license. Without a Confluent Private Cloud license, a PKC allows up to three LKCs, and you create two. To create more LKCs, see Apply a license to create more than three LKCs.

Sample users

The quick start uses three users that you add to the Lightweight Directory Access Protocol (LDAP) directory in Step 2:

User	Password	Role
platform-operator	platform-operator-secret	OrganizationAdmin on the PKC. This user represents the platform operator, who creates and manages LKCs.
payments-admin	payments-admin-secret	TenantAdmin on the payments LKC only.
analytics-admin	analytics-admin-secret	TenantAdmin on the analytics LKC only.

You grant these roles in Step 5.

Step 1: Prepare your environment

1. Set the following environment variables. Use the same shell for every step, or set them again in each new shell:

   export EXAMPLES=~/cfk-mt/confluent-kubernetes-examples
   export CP_IMAGE=confluentinc/cp-server:8.4.0
   export INIT_IMAGE=confluentinc/confluent-init-container:3.4.0
   export C3_IMAGE=confluentinc/cp-enterprise-control-center-next-gen:2.7.0
   export PROM_IMAGE=confluentinc/cp-enterprise-prometheus:2.7.0
   export ALERT_IMAGE=confluentinc/cp-enterprise-alertmanager:2.7.0
   export DOMAIN=mt.example.com
   

   To use a later patch release, change the tag.

2. Create a local Kubernetes cluster with one control-plane node and four worker nodes:

   cat <<EOF | kind create cluster --name cfk-mt --config=-
   kind: Cluster
   apiVersion: kind.x-k8s.io/v1alpha4
   nodes:
   - role: control-plane
   - role: worker
   - role: worker
   - role: worker
   - role: worker
   EOF
   kubectl config use-context kind-cfk-mt
   

3. Create the confluent namespace and set it as your default namespace:

   kubectl create namespace confluent
   kubectl config set-context --current --namespace=confluent
   

   Note

   Use the confluent namespace for every step. The manifests hard-code it in service endpoints, such as kafka.confluent.svc.cluster.local.

4. Clone the CFK examples repository. It contains the OpenLDAP chart that you install in the next step:

   git clone --depth 1 https://github.com/confluentinc/confluent-kubernetes-examples.git "$EXAMPLES"
   

Step 2: Install CFK, LDAP, and the security dependencies

In this step, you set up everything the PKC depends on:

• A certificate authority (CA), which CFK uses to issue the certificates for every component.

• CFK itself, which deploys and manages the other components.

• An LDAP directory, which Metadata Service (MDS) uses to authenticate users.

• The MDS token key pair, which MDS uses to sign tokens.

• The secrets that the components use to authenticate to each other.

1. Create the CA key pair:

   openssl genrsa -out /tmp/cfk-mt-ca-key.pem 2048
   openssl req -x509 -new -nodes -key /tmp/cfk-mt-ca-key.pem -days 365 -subj "/CN=cfk-mt-ca" -out /tmp/cfk-mt-ca.pem
   kubectl create secret tls ca-pair-sslcerts --cert=/tmp/cfk-mt-ca.pem --key=/tmp/cfk-mt-ca-key.pem
   rm -f /tmp/cfk-mt-ca-key.pem /tmp/cfk-mt-ca.pem
   

2. Install CFK and wait for it to roll out:

   helm repo add confluentinc https://packages.confluent.io/helm
   helm repo update
   helm upgrade --install operator confluentinc/confluent-for-kubernetes --namespace confluent \
     --set managedCerts.enabled=true \
     --set managedCerts.caCertificate.secretRef=ca-pair-sslcerts \
     --set kRaftEnabled=true
   kubectl rollout status deploy/confluent-operator --timeout=180s
   

   The managedCerts settings make CFK read the managed-cert-add-sans annotation that the Kafka manifest in Step 3 sets. managedCerts is disabled by default, and CFK ignores the annotation until you enable it. kRaftEnabled turns on KRaft cluster ID recovery.

   For other installation options, see Deploy Confluent for Kubernetes.

3. Install OpenLDAP. The chart creates the kafka and mds service users:

   helm upgrade --install test-ldap "$EXAMPLES/assets/openldap" \
     --namespace confluent -f "$EXAMPLES/assets/openldap/ldaps-rbac.yaml"
   
   kubectl wait --for=condition=ready pod/ldap-0 --timeout=180s
   

   Wait about 30 seconds before you continue. The chart restarts OpenLDAP once shortly after the pod reports ready, and the restart discards any users you add before it finishes.

4. Add the three sample users:

   kubectl exec -i ldap-0 -- ldapadd -x -c -H ldap://localhost:389 \
     -D "cn=admin,dc=test,dc=com" -w confluentrox <<EOF
   dn: cn=platform-operator,dc=test,dc=com
   objectClass: simpleSecurityObject
   objectClass: organizationalRole
   cn: platform-operator
   userPassword: platform-operator-secret
   
   dn: cn=payments-admin,dc=test,dc=com
   objectClass: simpleSecurityObject
   objectClass: organizationalRole
   cn: payments-admin
   userPassword: payments-admin-secret
   
   dn: cn=analytics-admin,dc=test,dc=com
   objectClass: simpleSecurityObject
   objectClass: organizationalRole
   cn: analytics-admin
   userPassword: analytics-admin-secret
   EOF
   

5. Confirm that the three users exist:

   kubectl exec ldap-0 -- ldapsearch -x -LLL -H ldap://localhost:389 \
     -D "cn=admin,dc=test,dc=com" -w confluentrox \
     -b "dc=test,dc=com" \
     "(|(cn=platform-operator)(cn=payments-admin)(cn=analytics-admin))" dn \
     | grep -c '^dn:'
   

   The command prints 3. If it prints a lower number, run the ldapadd command again.

6. Create the MDS token key pair. MDS uses it to sign the tokens it issues:

   openssl genrsa -out /tmp/cfk-mt-mds-priv.pem 2048
   openssl rsa -in /tmp/cfk-mt-mds-priv.pem -pubout -out /tmp/cfk-mt-mds-pub.pem
   
   kubectl create secret generic mds-token \
     --from-file=mdsTokenKeyPair.pem=/tmp/cfk-mt-mds-priv.pem \
     --from-file=mdsPublicKey.pem=/tmp/cfk-mt-mds-pub.pem
   
   rm -f /tmp/cfk-mt-mds-priv.pem /tmp/cfk-mt-mds-pub.pem
   

7. Create the secrets that hold the credentials for the PKC, Confluent Control Center, and the sample users. The external-users secret contains only the three sample users. Clients that connect through the external listener authenticate against it:

   kubectl create secret generic credential \
     --from-literal=plain-users.json='{"kafka":"kafka-secret","operator":"operator-secret","kafka_client":"kafka_client-secret"}' \
     --from-literal=plain.txt=$'username=kafka\npassword=kafka-secret' \
     --from-literal=plain-interbroker.txt=$'username=kafka\npassword=kafka-secret' \
     --from-literal=kafka-server-listener-internal-plain-metrics.txt=$'username=kafka\npassword=kafka-secret' \
     --from-literal=controlcenter-client-plain-jaas.conf='org.apache.kafka.common.security.plain.PlainLoginModule required username="kafka" password="kafka-secret";' \
     --from-literal=ldap.txt=$'username=cn=mds,dc=test,dc=com\npassword=Developer!'
   
   kubectl create secret generic external-users \
     --from-literal=plain-users.json='{"platform-operator":"platform-operator-secret","payments-admin":"payments-admin-secret","analytics-admin":"analytics-admin-secret"}'
   
   kubectl create secret generic mds-client-erp --from-literal=bearer.txt=$'username=kafka\npassword=kafka-secret'
   kubectl create secret generic rest-credential --from-literal=bearer.txt=$'username=kafka\npassword=kafka-secret'
   
   kubectl create secret generic prometheus-credentials \
     --from-literal=basic.txt='prometheus-admin: prometheus-admin-password'
   
   kubectl create secret generic prometheus-client-creds \
     --from-literal=basic.txt=$'username=prometheus-admin\npassword=prometheus-admin-password'
   
   kubectl create secret generic alertmanager-credentials \
     --from-literal=basic.txt='alertmanager-admin: alertmanager-admin-password'
   
   kubectl create secret generic alertmanager-client-creds \
     --from-literal=basic.txt=$'username=alertmanager-admin\npassword=alertmanager-admin-password'
   

Step 3: Deploy the PKC

A PKC is a Kafka custom resource (CR) with spec.multiTenant.enabled set to true. When you enable Multi-Tenancy, CFK rejects the CR unless the CR also configures the following:

• TLS on the internal and external listeners

• role-based access control (RBAC) authorization

• An embedded MDS

• A memory request or limit for the broker pods

CFK also adds an in-cluster listener for LKCs, TOKENMT, on port 9076. LKCs use this listener for in-cluster connections.

You don’t enable Multi-Tenancy on the KRaftController or ControlCenter CRs. The ControlCenter CR detects it from the Kafka CR that it references, and the KRaftController CR follows the Kafka CR that references it.

1. Deploy the KRaft controller:

   cat <<EOF | kubectl apply -f -
   apiVersion: platform.confluent.io/v1beta1
   kind: KRaftController
   metadata:
     name: kraftcontroller
     namespace: confluent
   spec:
     replicas: 1
     dataVolumeCapacity: 10G
     configOverrides:
       server:
         - default.replication.factor=3
         - min.insync.replicas=2
     image:
       application: $CP_IMAGE
       init: $INIT_IMAGE
     authorization:
       type: rbac
       superUsers:
         - User:kafka
         - User:mds
         - User:operator
     tls:
       autoGeneratedCerts: true
     listeners:
       controller:
         authentication:
           type: plain
           jaasConfig:
             secretRef: credential
         tls:
           enabled: true
     dependencies:
       mdsKafkaCluster:
         bootstrapEndpoint: kafka.confluent.svc.cluster.local:9071
         authentication:
           type: plain
           jaasConfig:
             secretRef: credential
         tls:
           enabled: true
       metricsClient:
         url: http://controlcenter.confluent.svc.cluster.local:9090
         authentication:
           type: basic
           basic:
             secretRef: prometheus-client-creds
   EOF
   

2. Deploy the PKC:

   cat <<EOF | kubectl apply -f -
   apiVersion: platform.confluent.io/v1beta1
   kind: Kafka
   metadata:
     name: kafka
     namespace: confluent
     annotations:
       platform.confluent.io/managed-cert-add-sans: "*.$DOMAIN,*.b0.$DOMAIN,*.b1.$DOMAIN,*.b2.$DOMAIN"
   spec:
     replicas: 3
     dataVolumeCapacity: 10G
     image:
       application: $CP_IMAGE
       init: $INIT_IMAGE
     podTemplate:
       resources:
         requests:
           memory: 4Gi
         limits:
           memory: 4Gi
     multiTenant:
       enabled: true
       createTenantNameServices: true
     authorization:
       type: rbac
       superUsers:
         - User:kafka
         - User:mds
         - User:operator
     tls:
       autoGeneratedCerts: true
     listeners:
       internal:
         authentication:
           type: plain
           jaasConfig:
             secretRef: credential
         tls:
           enabled: true
       external:
         externalAccess:
           type: loadBalancer
           loadBalancer:
             domain: $DOMAIN
         authentication:
           type: plain
           jaasConfig:
             secretRef: external-users
         tls:
           enabled: true
       replication:
         authentication:
           type: plain
           jaasConfig:
             secretRef: credential
         tls:
           enabled: true
     services:
       mds:
         tls:
           enabled: true
         tokenKeyPair:
           secretRef: mds-token
         provider:
           ldap:
             address: ldap://ldap.confluent.svc.cluster.local:389
             authentication:
               type: simple
               simple:
                 secretRef: credential
             tls:
               enabled: false
             configurations:
               groupNameAttribute: cn
               groupObjectClass: group
               groupMemberAttribute: member
               groupMemberAttributePattern: CN=(.*),DC=test,DC=com
               groupSearchBase: dc=test,dc=com
               userNameAttribute: cn
               userObjectClass: organizationalRole
               userMemberOfAttributePattern: CN=(.*),DC=test,DC=com
               userSearchBase: dc=test,dc=com
     dependencies:
       kRaftController:
         clusterRef:
           name: kraftcontroller
         controllerListener:
           authentication:
             type: plain
             jaasConfig:
               secretRef: credential
           tls:
             enabled: true
       kafkaRest:
         authentication:
           type: bearer
           bearer:
             secretRef: mds-client-erp
       metricsClient:
         url: http://controlcenter.confluent.svc.cluster.local:9090
         authentication:
           type: basic
           basic:
             secretRef: prometheus-client-creds
   EOF
   

   The manifest includes the following Multi-Tenancy settings:

   • multiTenant.createTenantNameServices: true: Makes the in-cluster LKC hostnames resolve without changes to your cluster’s DNS. Use this option only when you can’t edit your cluster’s DNS. It creates a Service for every broker of every LKC, which uses up the cluster’s Service IP range as you add LKCs and brokers. Otherwise, configure DNS rewrite rules for the LKC hostnames.

   • managed-cert-add-sans annotation: Adds the LKC hostnames to the certificate that CFK generates. LKC hostnames take these forms, where <lkc-id> already includes the lkc- prefix:

     • Bootstrap: <lkc-id>-kafka.<domain>

     • Brokers: <lkc-id>-kafka.b<n>.<domain>

     A wildcard matches only one DNS label, so the annotation lists *.<domain> for the bootstrap hostnames and a separate *.b<n>.<domain> for each broker. CFK reads the annotation because the Helm install in Step 2 enables managedCerts. For other certificate approaches, see the Configure TLS and DNS page.

   • podTemplate.resources memory of 4Gi: Each LKC’s quota capacity derives from the broker heap. Lower values of memory can cause the brokers to crash-loop on startup.

   CFK also creates the kafka-bootstrap-lb and kafka-<n>-lb Service objects. On kind, these Service objects stay in <pending> because kind has no external load balancer, which is expected. In Step 6, you reach them by their cluster IP addresses.

3. Create the KafkaRestClass that CFK uses to manage LKCs and role bindings through MDS:

   cat <<EOF | kubectl apply -f -
   apiVersion: platform.confluent.io/v1beta1
   kind: KafkaRestClass
   metadata:
     name: default
     namespace: confluent
   spec:
     kafkaClusterRef:
       name: kafka
     kafkaRest:
       authentication:
         type: bearer
         bearer:
           secretRef: rest-credential
   EOF
   

4. Wait 5 to 10 minutes for the brokers to start, and then confirm that Multi-Tenancy is active:

   kubectl wait --for=jsonpath='{.status.phase}'=RUNNING kafka/kafka --timeout=15m
   kubectl get kafka kafka -o jsonpath='phase={.status.phase} multiTenant={.status.multiTenant.enabled} clusterID={.status.clusterID}{"\n"}'
   kubectl wait --for=jsonpath='{.status.phase}'=RUNNING kraftcontroller/kraftcontroller --timeout=10m
   

   The second command prints multiTenant=true, which confirms that Multi-Tenancy is active.

   CFK restarts the KRaft controller once after the PKC becomes multi-tenant. Allow a few extra minutes for the restart to finish.

For the full list of settings that a multi-tenant PKC supports, see the Deploy a Multi-Tenant Physical Kafka Cluster page.

Step 4: Deploy Confluent Control Center

Use the environment variables that you set in Step 1. Confluent Control Center detects Multi-Tenancy from the Kafka CR that it references, so you don’t set a Multi-Tenancy field. The ControlCenter CR must set spec.authorization.type to rbac.

cat <<EOF | kubectl apply -f -
apiVersion: platform.confluent.io/v1beta1
kind: ControlCenter
metadata:
  name: controlcenter
  namespace: confluent
spec:
  replicas: 1
  image:
    application: $C3_IMAGE
    init: $INIT_IMAGE
  dataVolumeCapacity: 10Gi
  authorization:
    type: rbac
  tls:
    autoGeneratedCerts: true
  dependencies:
    kafka:
      discovery:
        name: kafka
        namespace: confluent
        secretRef: credential
      tls:
        enabled: true
    mds:
      endpoint: https://kafka.confluent.svc.cluster.local:8090
      tokenKeyPair:
        secretRef: mds-token
      authentication:
        type: bearer
        bearer:
          secretRef: rest-credential
      tls:
        enabled: true
    prometheusClient:
      url: http://controlcenter.confluent.svc.cluster.local:9090
      authentication:
        type: basic
        basic:
          secretRef: prometheus-client-creds
    alertManagerClient:
      url: http://controlcenter.confluent.svc.cluster.local:9093
      authentication:
        type: basic
        basic:
          secretRef: alertmanager-client-creds
  services:
    prometheus:
      image: $PROM_IMAGE
      authentication:
        type: basic
        basic:
          secretRef: prometheus-credentials
      pvc:
        dataVolumeCapacity: 10Gi
    alertmanager:
      image: $ALERT_IMAGE
      authentication:
        type: basic
        basic:
          secretRef: alertmanager-credentials
EOF
kubectl wait --for=jsonpath='{.status.phase}'=RUNNING \
  controlcenter/controlcenter --timeout=10m

Step 5: Create the LKCs and grant access

1. Grant the OrganizationAdmin role on the PKC to the kafka user, the identity that CFK uses to call MDS. CFK can’t create LKCs until this identity holds a role that manages LKCs, such as OrganizationAdmin. Superuser status doesn’t grant this permission.

   PKC=$(kubectl get kafka kafka -o jsonpath='{.status.clusterID}')
   cat <<EOF | kubectl apply -f -
   apiVersion: platform.confluent.io/v1beta1
   kind: ConfluentRolebinding
   metadata:
     name: cfk-orgadmin
     namespace: confluent
   spec:
     principal:
       type: user
       name: kafka
     role: OrganizationAdmin
     clustersScopeByIds:
       kafkaClusterId: $PKC
     kafkaRestClassRef:
       name: default
   EOF
   kubectl wait --for=jsonpath='{.status.state}'=CREATED \
     confluentrolebinding/cfk-orgadmin --timeout=5m
   

2. Create the payments and analytics LKCs:

   cat <<EOF | kubectl apply -f -
   apiVersion: platform.confluent.io/v1beta1
   kind: LogicalKafkaCluster
   metadata:
     name: payments
     namespace: confluent
   spec:
     kafkaClusterRef:
       name: kafka
     kafkaRestClassRef:
       name: default
     quotas:
       ingressByteRate: 10485760
       egressByteRate: 31457280
     limits:
       maxPartitions: 1000
   ---
   apiVersion: platform.confluent.io/v1beta1
   kind: LogicalKafkaCluster
   metadata:
     name: analytics
     namespace: confluent
   spec:
     kafkaClusterRef:
       name: kafka
     kafkaRestClassRef:
       name: default
     quotas:
       ingressByteRate: 10485760
       egressByteRate: 31457280
     limits:
       maxPartitions: 1000
   EOF
   kubectl wait --for=jsonpath='{.status.state}'=CREATED \
     lkc/payments lkc/analytics --timeout=5m
   

   Clients can produce up to 10 MiB per second to each LKC and consume up to 30 MiB per second from it, and each LKC can hold at most 1,000 partitions. The quotas and limits fields are optional.

   Caution

   When you update an LKC, include every spec.quotas and spec.limits field that you want to keep. CFK overwrites both objects in MDS on every reconciliation and clears any field that’s missing from the CR. For more information, see the Monitor and Manage Per-LKC Resource Quotas page.

3. Confirm that each LKC is ready and view its bootstrap URLs:

   kubectl get lkc -o custom-columns='NAME:.metadata.name,ID:.status.logicalClusterID,STATE:.status.state,PHASE:.status.phase,EXTERNAL:.status.bootstrapEndpoint,IN-CLUSTER:.status.internalBootstrapEndpoint'
   

   An LKC is ready when STATE is CREATED, PHASE is Running, and it has an lkc-<id> identifier. Each LKC has two bootstrap URLs:

   Path	LKC status field and example endpoint	Client authentication
   In-cluster	status.internalBootstrapEndpoint, for example <lkc-id>-kafka.confluent.svc.cluster.local:9076	OAUTHBEARER with an MDS token, using the user’s LDAP username and password
   External	status.bootstrapEndpoint, for example <lkc-id>-kafka.mt.example.com:9092	SASL/PLAIN, using the credentials in the external-users secret

4. Grant the roles to the sample users:

   cat <<EOF | kubectl apply -f -
   apiVersion: platform.confluent.io/v1beta1
   kind: ConfluentRolebinding
   metadata:
     name: platform-operator-orgadmin
     namespace: confluent
   spec:
     principal:
       type: user
       name: platform-operator
     role: OrganizationAdmin
     clustersScopeByIds:
       kafkaClusterId: $PKC
     kafkaRestClassRef:
       name: default
   ---
   apiVersion: platform.confluent.io/v1beta1
   kind: ConfluentRolebinding
   metadata:
     name: payments-admin-payments
     namespace: confluent
   spec:
     principal:
       type: user
       name: payments-admin
     role: TenantAdmin
     kafkaClusterRef:
       name: payments
       kind: LogicalKafkaCluster
     kafkaRestClassRef:
       name: default
   ---
   apiVersion: platform.confluent.io/v1beta1
   kind: ConfluentRolebinding
   metadata:
     name: analytics-admin-analytics
     namespace: confluent
   spec:
     principal:
       type: user
       name: analytics-admin
     role: TenantAdmin
     kafkaClusterRef:
       name: analytics
       kind: LogicalKafkaCluster
     kafkaRestClassRef:
       name: default
   EOF
   kubectl wait --for=jsonpath='{.status.state}'=CREATED \
     confluentrolebinding/platform-operator-orgadmin \
     confluentrolebinding/payments-admin-payments \
     confluentrolebinding/analytics-admin-analytics --timeout=5m
   

   The TenantAdmin role covers topics, records, and consumer groups, so these users need no access control lists (ACLs). CFK looks up each LKC’s ID from kafkaClusterRef.

   Caution

   Set kind: LogicalKafkaCluster and the LKC’s name on an LKC role binding. A binding that names the PKC itself, with kind: Kafka and the PKC’s name, is scoped to the PKC and grants access to matching resources in every LKC on the PKC. For more information, see the Manage LKC Access with RBAC page.

Step 6: Create a client pod and map LKC hostnames

Use the environment variables that you set in Step 1. In this step, you create a pod to run the Kafka command-line tools from inside the cluster. The pod includes the cluster’s CA certificate and a client configuration for each TenantAdmin and connection path.

Create the pod only after the LKCs exist, because the pod’s host aliases include each LKC’s ID. If you create another LKC later and want to reach it through the external listener, recreate the pod.

1. Create a ConfigMap that holds the client configurations. Every configuration makes the client verify the broker’s hostname:

   cat <<EOF | kubectl apply -f -
   apiVersion: v1
   kind: ConfigMap
   metadata:
     name: tenant-clients
     namespace: confluent
   data:
     payments-admin-tokenmt.properties: |
       security.protocol=SASL_SSL
       sasl.mechanism=OAUTHBEARER
       sasl.login.callback.handler.class=io.confluent.kafka.clients.plugins.auth.token.TokenUserLoginCallbackHandler
       sasl.jaas.config=org.apache.kafka.common.security.oauthbearer.OAuthBearerLoginModule required username="payments-admin" password="payments-admin-secret" metadataServerUrls="https://kafka.confluent.svc.cluster.local:8090";
       ssl.truststore.type=PEM
       ssl.truststore.location=/mnt/ca/ca.pem
     analytics-admin-tokenmt.properties: |
       security.protocol=SASL_SSL
       sasl.mechanism=OAUTHBEARER
       sasl.login.callback.handler.class=io.confluent.kafka.clients.plugins.auth.token.TokenUserLoginCallbackHandler
       sasl.jaas.config=org.apache.kafka.common.security.oauthbearer.OAuthBearerLoginModule required username="analytics-admin" password="analytics-admin-secret" metadataServerUrls="https://kafka.confluent.svc.cluster.local:8090";
       ssl.truststore.type=PEM
       ssl.truststore.location=/mnt/ca/ca.pem
     payments-admin-external.properties: |
       security.protocol=SASL_SSL
       sasl.mechanism=PLAIN
       sasl.jaas.config=org.apache.kafka.common.security.plain.PlainLoginModule required username="payments-admin" password="payments-admin-secret";
       ssl.truststore.type=PEM
       ssl.truststore.location=/mnt/ca/ca.pem
     analytics-admin-external.properties: |
       security.protocol=SASL_SSL
       sasl.mechanism=PLAIN
       sasl.jaas.config=org.apache.kafka.common.security.plain.PlainLoginModule required username="analytics-admin" password="analytics-admin-secret";
       ssl.truststore.type=PEM
       ssl.truststore.location=/mnt/ca/ca.pem
   EOF
   

2. Create the pod. The external LKC hostnames aren’t in DNS, so the pod’s hostAliases map them to the cluster IP addresses of the load-balancer Service objects:

   PAY=$(kubectl get lkc payments -o jsonpath='{.status.logicalClusterID}')
   ANA=$(kubectl get lkc analytics -o jsonpath='{.status.logicalClusterID}')
   BS=$(kubectl get svc kafka-bootstrap-lb -o jsonpath='{.spec.clusterIP}')
   B0=$(kubectl get svc kafka-0-lb -o jsonpath='{.spec.clusterIP}')
   B1=$(kubectl get svc kafka-1-lb -o jsonpath='{.spec.clusterIP}')
   B2=$(kubectl get svc kafka-2-lb -o jsonpath='{.spec.clusterIP}')
   
   cat <<EOF | kubectl apply -f -
   apiVersion: v1
   kind: Pod
   metadata:
     name: kafka-client
     namespace: confluent
   spec:
     hostAliases:
       - ip: "$BS"
         hostnames: ["$PAY-kafka.$DOMAIN", "$ANA-kafka.$DOMAIN"]
       - ip: "$B0"
         hostnames: ["$PAY-kafka.b0.$DOMAIN", "$ANA-kafka.b0.$DOMAIN"]
       - ip: "$B1"
         hostnames: ["$PAY-kafka.b1.$DOMAIN", "$ANA-kafka.b1.$DOMAIN"]
       - ip: "$B2"
         hostnames: ["$PAY-kafka.b2.$DOMAIN", "$ANA-kafka.b2.$DOMAIN"]
     containers:
       - name: cli
         image: $CP_IMAGE
         imagePullPolicy: IfNotPresent
         command: ["sleep", "infinity"]
         volumeMounts:
           - name: ca
             mountPath: /mnt/ca
           - name: clients
             mountPath: /clients
     volumes:
       - name: ca
         secret:
           secretName: ca-pair-sslcerts
           items:
             - key: tls.crt
               path: ca.pem
       - name: clients
         configMap:
           name: tenant-clients
   EOF
   kubectl wait --for=condition=ready pod/kafka-client --timeout=180s
   

Step 7: Verify access and LKC isolation

1. Wait for the KRaft controller restart from Step 3 to finish. If you create topics before it finishes, the request can fail:

   kubectl rollout status statefulset/kraftcontroller --timeout=10m
   

2. Check the status of the components and the LKCs:

   kubectl get kafka,kraftcontroller,controlcenter -o wide
   kubectl get lkc -o custom-columns='NAME:.metadata.name,ID:.status.logicalClusterID,STATE:.status.state,PHASE:.status.phase'
   

   The output shows the following:

   • The Kafka, KRaftController, and ControlCenter CRs each report RUNNING. The Kafka and ControlCenter CRs also report MULTITENANT set to true.

   • Both LKCs report a state of CREATED and a phase of Running.

3. Connect to the payments LKC as payments-admin over the in-cluster path. Create a topic, produce 10 messages to it, and consume them:

   PAY_INT=$(kubectl get lkc payments -o jsonpath='{.status.internalBootstrapEndpoint}')
   PAY_EXT=$(kubectl get lkc payments -o jsonpath='{.status.bootstrapEndpoint}')
   
   kubectl exec kafka-client -- kafka-topics \
     --bootstrap-server "$PAY_INT" \
     --command-config /clients/payments-admin-tokenmt.properties \
     --create --topic orders --partitions 3 --replication-factor 3
   
   seq 1 10 | kubectl exec -i kafka-client -- kafka-console-producer \
     --bootstrap-server "$PAY_INT" \
     --producer.config /clients/payments-admin-tokenmt.properties \
     --topic orders
   
   kubectl exec kafka-client -- kafka-console-consumer \
     --bootstrap-server "$PAY_INT" \
     --consumer.config /clients/payments-admin-tokenmt.properties \
     --topic orders --group payments-check --from-beginning \
     --max-messages 10 --timeout-ms 30000
   

   The first command prints Created topic orders. The consumer prints the records 1 through 10.

4. Connect to the payments LKC as payments-admin over the external path. List the topics and consume the records:

   kubectl exec kafka-client -- kafka-topics \
     --bootstrap-server "$PAY_EXT" \
     --command-config /clients/payments-admin-external.properties --list
   
   kubectl exec kafka-client -- kafka-console-consumer \
     --bootstrap-server "$PAY_EXT" \
     --consumer.config /clients/payments-admin-external.properties \
     --topic orders --group payments-check-ext --from-beginning \
     --max-messages 10 --timeout-ms 30000
   

   The topic list includes orders, and the consumer prints the same 10 records.

5. As analytics-admin, list the topics at the payments LKC’s bootstrap URL:

   kubectl exec kafka-client -- kafka-topics \
     --bootstrap-server "$PAY_INT" \
     --command-config /clients/analytics-admin-tokenmt.properties --list
   

   The command returns an empty list. The analytics-admin user has no role on the payments LKC, so the user can’t see any of that LKC’s topics.

6. Open Confluent Control Center to confirm that each TenantAdmin sees only their own LKC. Confluent Control Center accepts only its own certificate’s hostname, so map that hostname to your machine and forward the port:

   echo "127.0.0.1 controlcenter.confluent.svc.cluster.local" | sudo tee -a /etc/hosts
   kubectl port-forward svc/controlcenter 9021:9021
   

   In a browser, go to https://controlcenter.confluent.svc.cluster.local:9021 and log in:

   • As platform-operator, Confluent Control Center lists the PKC and both LKCs.

   • As payments-admin, Confluent Control Center lists only the payments LKC.

Clean up

If you want to keep the cluster and delete individual resources instead, note the following:

• Deleting an LKC is permanent. CFK deletes the LKC’s topics within minutes, and its LKC ID is never reused.

• Don’t delete all of the broker pods at once. If no other broker is running, a broker that uses RBAC and LDAP can wait up to 10 minutes for its authorizer.

To remove everything this quick start created, delete the cluster and the /etc/hosts entry that you added in Step 7:

1. Delete the kind cluster. This removes all the resources in it:

   kind delete cluster --name cfk-mt
   

2. Remove the /etc/hosts entry for Confluent Control Center:

   • On macOS:

     sudo sed -i '' '/controlcenter.confluent.svc.cluster.local/d' /etc/hosts
     

   • On Linux:

     sudo sed -i '/controlcenter.confluent.svc.cluster.local/d' /etc/hosts
     

Apply a license to create more than three LKCs

Without a Confluent Private Cloud license, a PKC allows up to three LKCs. MDS rejects a request to create a fourth. Deleting an LKC frees its slot.

To create more than three LKCs, apply your Confluent Private Cloud license to the PKC:

1. Store the license in a secret. CFK expects the secret content in the form license=<license>:

   { printf 'license='; cat <path-to-license-file>; } > /tmp/cfk-mt-license.txt
   kubectl create secret generic confluent-license --from-file=license.txt=/tmp/cfk-mt-license.txt
   rm -f /tmp/cfk-mt-license.txt
   

   If your license file already starts with license=, skip the printf and pass the file to --from-file as it is.

2. Reference the secret from the Kafka, KRaft, and Confluent Control Center CRs. Each patch restarts that component:

   kubectl patch kafka kafka --type=merge -p '{"spec":{"license":{"secretRef":"confluent-license"}}}'
   kubectl patch kraftcontroller kraftcontroller --type=merge -p '{"spec":{"license":{"secretRef":"confluent-license"}}}'
   kubectl patch controlcenter controlcenter --type=merge -p '{"spec":{"license":{"secretRef":"confluent-license"}}}'
   

For more information, see Manage Confluent Platform Licenses.

Troubleshooting

The following table lists common problems, organized by the step where you’re likely to see them. Find the symptom, and then apply the fix.

Symptom	Cause and fix
Step 2: Fewer than three users in LDAP	Run the ldapadd command again.
Step 3: The Kafka CR is rejected with a message that a field is required when multiTenant.enabled is true	A Multi-Tenancy prerequisite is missing: TLS, RBAC, MDS, or broker memory. Restore it in the CR.
Step 3: Brokers crash-loop with an error that references confluent.quota.tenant.broker.max.producer.rate	Broker memory is too low. Keep the memory at 4Gi.
Step 3: Pods stay Pending with Insufficient memory	Give Docker more memory.
Step 3: Brokers log metrics-push failures	This is expected until Confluent Control Center is running.
Step 5: An LKC is in the ERROR state with error code 40301	The cfk-orgadmin role binding isn’t CREATED yet. The LKC recovers on its own.
Step 5: An LKC is in the ERROR state with a message that authentication is enabled for the Kafka REST API	The KafkaRestClass named default is missing. Create it as described in Step 3.
Step 5: An LKC stays in PROVISIONING for a few minutes after the brokers start	MDS is still starting and returns 503. Wait for MDS to finish starting.
Step 5: A fourth LKC is in the ERROR state with a tenant creation limit message	A PKC without a license allows up to three LKCs. Delete one or apply a license. See Apply a license to create more than three LKCs.
Step 5: The operator log reports No Enterprise license detected	This is expected when you haven’t applied a license.
Step 7: payments-admin can’t create a topic right after setup	The KRaftController is still restarting. Wait for kraftcontroller-0 to be RUNNING.
Step 7: An external client reports an unknown host	The LKC is missing from the client pod’s hostAliases. Recreate the client pod.
Step 7: An external client reports a TLS hostname error	The certificate lacks the LKC’s Subject Alternative Names (SANs). Check the managed-cert-add-sans annotation on the Kafka CR, and confirm that the Helm install in Step 2 enabled managedCerts.
Step 7: Confluent Control Center returns 400 Invalid SNI	Browse to controlcenter.confluent.svc.cluster.local, not localhost.
Step 7: platform-operator sees only the PKC in Confluent Control Center	A PKC-scoped OrganizationAdmin binding makes every running LKC visible. Confirm that the platform-operator-orgadmin role binding is CREATED and that each LKC is Running. As a workaround, add an OrganizationAdmin role binding for platform-operator on each LKC, with kind: LogicalKafkaCluster.

To investigate other problems, run the following commands:

kubectl logs deploy/confluent-operator | grep -iE 'error|logicalkafkacluster' | tail -50
kubectl describe lkc payments
kubectl get events --sort-by=.lastTimestamp | tail -30
kubectl logs kafka-0 -c kafka | grep -i tenant | tail -30

Next steps

To learn more about Multi-Tenancy or to move toward production, see the following pages:

• Multi-Tenancy Overview: Learn how LKCs, quotas, and isolation work.

• Configure TLS and DNS: Choose a certificate and DNS approach for LKC endpoints.

• Deploy a Multi-Tenant Physical Kafka Cluster: Set up a PKC for production.

• Onboard an LKC: Provision an LKC and assign its TenantAdmin.

• Onboard a Developer: Grant a developer access to an LKC.

• Manage LKC Access with RBAC: Scope role bindings to an LKC and grant LKC access.

• Monitor and Manage Per-LKC Resource Quotas: Track each LKC’s usage and adjust its limits.