Deploy Apache Kafka to GKE using Confluent

Prepare the environment

In this tutorial, you use Cloud Shell to manage resources hosted on Google Cloud. Cloud Shell is preinstalled with the software you need for this tutorial, including kubectl, the gcloud CLI, Helm, and Terraform.

To set up your environment with Cloud Shell, follow these steps:

1. Launch a Cloud Shell session from the Google Cloud console, by clicking Activate Cloud Shell in the Google Cloud console. This launches a session in the bottom pane of the Google Cloud console.

2. Set environment variables:

   export PROJECT_ID=PROJECT_ID
   export KUBERNETES_CLUSTER_PREFIX=kafka
   export REGION=us-central1
   

   Replace PROJECT_ID: your Google Cloud with your project ID.

3. Clone the GitHub repository:

   git clone https://github.com/GoogleCloudPlatform/kubernetes-engine-samples
   

4. Change to the working directory:

   cd kubernetes-engine-samples/streaming
   

Create your cluster infrastructure

In this section, you run a Terraform script to create a private, highly-available, regional GKE cluster. The following steps allow public access to the control plane. To restrict access, create a private cluster.

You can install the operator using a Standard or Autopilot cluster.

export GOOGLE_OAUTH_ACCESS_TOKEN=$(gcloud auth print-access-token)
terraform -chdir=kafka/terraform/gke-standard init
terraform -chdir=kafka/terraform/gke-standard apply -var project_id=${PROJECT_ID} \
  -var region=${REGION} \
  -var cluster_prefix=${KUBERNETES_CLUSTER_PREFIX}

...
Apply complete! Resources: 14 added, 0 changed, 0 destroyed.

Outputs:

kubectl_connection_command = "gcloud container clusters get-credentials kafka-cluster --region us-central1"

export GOOGLE_OAUTH_ACCESS_TOKEN=$(gcloud auth print-access-token)
terraform -chdir=kafka/terraform/gke-autopilot init
terraform -chdir=kafka/terraform/gke-autopilot apply -var project_id=${PROJECT_ID} \
  -var region=${REGION} \
  -var cluster_prefix=${KUBERNETES_CLUSTER_PREFIX}

...
Apply complete! Resources: 12 added, 0 changed, 0 destroyed.

Outputs:

kubectl_connection_command = "gcloud container clusters get-credentials kafka-cluster --region us-central1"

Connect to the cluster

Configure kubectl to communicate with the cluster:

gcloud container clusters get-credentials ${KUBERNETES_CLUSTER_PREFIX}-cluster --region ${REGION}

Deploy the CFK operator to your cluster

In this section, you deploy the Confluent for Kubernetes (CFK) operator using a Helm chart and then deploy a Kafka cluster.

1. Add the Confluent Helm Chart repository:

   helm repo add confluentinc https://packages.confluent.io/helm
   

2. Add a namespace for the CFK operator and the Kafka cluster:

   kubectl create ns kafka
   

3. Deploy the CFK cluster operator using Helm:

   helm install confluent-operator confluentinc/confluent-for-kubernetes -n kafka
   

   To enable CFK to manage resources across all namespaces, add the parameter --set-namespaced=false to the Helm command.

4. Verify that the Confluent operator has been deployed successfully using Helm:

   helm ls -n kafka
   

   The output is similar to the following:

   NAME                  NAMESPACE  REVISION UPDATED                                  STATUS      CHART                                APP VERSION
   confluent-operator    kafka      1        2023-07-07 10:57:45.409158 +0200 CEST    deployed    confluent-for-kubernetes-0.771.13    2.6.0
   

Deploy Kafka

In this section, you deploy Kafka in a basic configuration and then try various advanced configuration scenarios to address availability, security, and observability requirements.

Basic configuration

The basic configuration for the Kafka instance includes the following components:

• Three replicas of Kafka brokers, with a minimum of two available replicas required for cluster consistency.
• Three replicas of ZooKeeper nodes, forming a cluster.
• Two Kafka listeners: one without authentication, and one utilizing TLS authentication with a certificate generated by CFK.
• Java MaxHeapSize and MinHeapSize set to 4 GB for Kafka.
• CPU resource allocation of 1 CPU request and 2 CPU limits, and 5 GB memory requests and limits for Kafka (4 GB for the main service and 0.5 GB for the metrics exporter) and 3 GB for Zookeeper (2 GB for the main service and 0.5 GB for the metrics exporter).
• 100 GB of storage allocated to each Pod using the premium-rwo storageClass, 100 for Kafka Data and 90/10 for Zookeeper Data/Log.
• Tolerations, nodeAffinities, and podAntiAffinities configured for each workload, ensuring proper distribution across nodes, utilizing their respective node pools and different zones.
• Communication inside the cluster secured by self-signed certificates using a Certificate Authority that you provide.

This configuration represents the minimal setup required to create a production-ready Kafka cluster. The following sections demonstrate custom configurations to address aspects such as cluster security, Access Control Lists (ACLs), topic management, certificate management and more.

Create a basic Kafka cluster

1. Generate a CA pair:

   openssl genrsa -out ca-key.pem 2048
   openssl req -new -key ca-key.pem -x509 \
     -days 1000 \
     -out ca.pem \
     -subj "/C=US/ST=CA/L=Confluent/O=Confluent/OU=Operator/CN=MyCA"
   

   Confluent for Kubernetes provides auto-generated certificates for Confluent Platform components to use for TLS network encryption. You must generate and provide a Certificate Authority (CA).

2. Create a Kubernetes Secret for the certificate authority:

   kubectl create secret tls ca-pair-sslcerts --cert=ca.pem --key=ca-key.pem -n kafka
   

   The name of the Secret is predefined

3. Create a new Kafka cluster using the basic configuration:

   kubectl apply -n kafka -f kafka-confluent/manifests/01-basic-cluster/my-cluster.yaml
   

   This command creates a Kafka custom resource and Zookeeper custom resource of the CFK operator that include CPU and memory requests and limits, block storage requests, and taints and affinities to distribute the provisioned Pods across Kubernetes nodes.

4. Wait a few minutes while Kubernetes starts the required workloads:

   kubectl wait pods -l app=my-cluster --for condition=Ready --timeout=300s -n kafka
   

5. Verify that the Kafka workloads were created:

   kubectl get pod,svc,statefulset,deploy,pdb -n kafka
   

   The output is similar to the following:

   NAME                                    READY   STATUS  RESTARTS   AGE
   pod/confluent-operator-864c74d4b4-fvpxs   1/1   Running   0        49m
   pod/my-cluster-0                        1/1   Running   0        17m
   pod/my-cluster-1                        1/1   Running   0        17m
   pod/my-cluster-2                        1/1   Running   0        17m
   pod/zookeeper-0                         1/1   Running   0        18m
   pod/zookeeper-1                         1/1   Running   0        18m
   pod/zookeeper-2                         1/1   Running   0        18m
   
   NAME                          TYPE      CLUSTER-IP   EXTERNAL-IP   PORT(S)                                                        AGE
   service/confluent-operator    ClusterIP   10.52.13.164   <none>      7778/TCP                                                       49m
   service/my-cluster            ClusterIP   None         <none>      9092/TCP,8090/TCP,9071/TCP,7203/TCP,7777/TCP,7778/TCP,9072/TCP   17m
   service/my-cluster-0-internal   ClusterIP   10.52.2.242  <none>      9092/TCP,8090/TCP,9071/TCP,7203/TCP,7777/TCP,7778/TCP,9072/TCP   17m
   service/my-cluster-1-internal   ClusterIP   10.52.7.98   <none>      9092/TCP,8090/TCP,9071/TCP,7203/TCP,7777/TCP,7778/TCP,9072/TCP   17m
   service/my-cluster-2-internal   ClusterIP   10.52.4.226  <none>      9092/TCP,8090/TCP,9071/TCP,7203/TCP,7777/TCP,7778/TCP,9072/TCP   17m
   service/zookeeper             ClusterIP   None         <none>      2181/TCP,7203/TCP,7777/TCP,3888/TCP,2888/TCP,7778/TCP          18m
   service/zookeeper-0-internal  ClusterIP   10.52.8.52   <none>      2181/TCP,7203/TCP,7777/TCP,3888/TCP,2888/TCP,7778/TCP          18m
   service/zookeeper-1-internal  ClusterIP   10.52.12.44  <none>      2181/TCP,7203/TCP,7777/TCP,3888/TCP,2888/TCP,7778/TCP          18m
   service/zookeeper-2-internal  ClusterIP   10.52.12.134   <none>      2181/TCP,7203/TCP,7777/TCP,3888/TCP,2888/TCP,7778/TCP          18m
   
   NAME                        READY   AGE
   statefulset.apps/my-cluster   3/3   17m
   statefulset.apps/zookeeper  3/3   18m
   
   NAME                               READY   UP-TO-DATE   AVAILABLE   AGE
   deployment.apps/confluent-operator   1/1   1          1         49m
   
   NAME                                  MIN AVAILABLE   MAX UNAVAILABLE   ALLOWED DISRUPTIONS   AGE
   poddisruptionbudget.policy/my-cluster   N/A           1               1                   17m
   poddisruptionbudget.policy/zookeeper  N/A           1               1                   18m
   

The operator creates the following resources:

• Two StatefulSets for Kafka and ZooKeeper.
• Three Pods for Kafka broker replicas.
• Three Pods for ZooKeeper replicas.
• Two PodDisruptionBudget resources, ensuring a maximum one unavailable replica for cluster consistency.
• The Service my-cluster which serves as the bootstrap server for Kafka clients connecting from within the Kubernetes cluster. All internal Kafka listeners are available in this Service.
• The Service zookeeper which allows Kafka brokers to connect to ZooKeeper nodes as clients.

Authentication and user management

This section shows you how to enable the authentication and authorization to secure Kafka Listeners and share credentials with clients.

Confluent for Kubernetes supports various authentication methods for Kafka, such as:

• SASL/PLAIN authentication: Clients use a username and password for authentication. The username and password are stored server-side in a Kubernetes secret.
• SASL/PLAIN with LDAP authentication: Clients use a username and password for authentication. The credentials are stored in an LDAP server.
• mTLS authentication: Clients use TLS certificates for authentication.

Limitations

• CFK does not provide Custom Resources for user management. However, you can store credentials in Secrets and refer to Secrets to in listener specs.
• Although there's no Custom Resource to manage ACLs directly, the official Confluent for Kubernetes provides guidance on configuring ACLs using the Kafka CLI.

Create a user

This section shows you how to deploy a CFK operator that demonstrates user management capabilities, including:

• A Kafka cluster with password-based authentication (SASL/PLAIN) enabled on one of the listeners
• A KafkaTopicwith 3 replicas
• User credentials with read and write permissions

1. Create a Secret with user credentials:

   export USERNAME=my-user
   export PASSWORD=$(openssl rand -base64 12)
   kubectl create secret generic my-user-credentials -n kafka \
     --from-literal=plain-users.json="{\"$USERNAME\":\"$PASSWORD\"}"
   

   Credentials should be stored in the following format:

   {
   "username1": "password1",
   "username2": "password2",
   ...
   "usernameN": "passwordN"
   }
   

2. Configure Kafka cluster to use a listener with password-based authentication SCRAM-SHA-512 authentication on port 9094:

   kubectl apply -n kafka -f kafka-confluent/manifests/02-auth/my-cluster.yaml
   

3. Set up a topic and a client Pod to interact with your Kafka cluster and execute Kafka commands:

   kubectl apply -n kafka -f kafka-confluent/manifests/02-auth/my-topic.yaml
   kubectl apply -n kafka -f kafka-confluent/manifests/02-auth/kafkacat.yaml
   

   GKE mounts the Secret my-user-credentials to the client Pod as a Volume.

4. When the client Pod is ready, connect to it and start producing and consuming messages using the provided credentials:

   kubectl wait pod kafkacat --for=condition=Ready --timeout=300s -n kafka
   kubectl exec -it kafkacat -n kafka -- /bin/sh
   

5. Produce a message using the my-user credentials and then consume the message to verify its receipt.

   export USERNAME=$(cat /my-user/plain-users.json|cut -d'"' -f 2)
   export PASSWORD=$(cat /my-user/plain-users.json|cut -d'"' -f 4)
   echo "Message from my-user" |kcat \
     -b my-cluster.kafka.svc.cluster.local:9094 \
     -X security.protocol=SASL_SSL \
     -X sasl.mechanisms=PLAIN \
     -X sasl.username=$USERNAME \
     -X sasl.password=$PASSWORD  \
     -t my-topic -P
   kcat -b my-cluster.kafka.svc.cluster.local:9094 \
     -X security.protocol=SASL_SSL \
     -X sasl.mechanisms=PLAIN \
     -X sasl.username=$USERNAME \
     -X sasl.password=$PASSWORD  \
     -t my-topic -C
   

   The output is similar to the following:

   Message from my-user
   % Reached end of topic my-topic [1] at offset 1
   % Reached end of topic my-topic [2] at offset 0
   % Reached end of topic my-topic [0] at offset 0
   

   Type CTRL+C to stop the consumer process. If you get a Connect refused error, wait a few minutes and then try again.

6. Exit the Pod shell

   exit
   

Backups and disaster recovery

Using the Confluent operator, you can implement efficient backup strategies by following certain patterns.

You can use Backup for GKE to backup:

• Kubernetes resource manifests.
• Confluent API custom resources and their definitions extracted from the Kubernetes API server of the cluster undergoing backup.
• Volumes that correspond to PersistentVolumeClaim resources found in the manifests.

For more information about how to backup and restore Kafka clusters using Backup for GKE, see Prepare for disaster recovery.

You can also perform a manual backup of your Kafka cluster. You should backup:

• The Kafka configuration, which includes all custom resources of the Confluent API such as KafkaTopicsorConnect
• The data, which is stored in the PersistentVolumes of the Kafka brokers

Storing Kubernetes resource manifests, including Confluent configurations, in Git repositories can eliminate the need for a separate backup for Kafka configuration as the resources can be reapplied to a new Kubernetes cluster when necessary.

To safeguard Kafka data recovery in scenarios where a Kafka server instance, or Kubernetes cluster where Kafka is deployed, is lost, we recommend that you configure the Kubernetes storage class used for provisioning volumes for Kafka brokers with the reclaimPolicy option set to Retain. We also recommended that you take snapshots of Kafka broker volumes.

The following manifest describes a StorageClass that uses the reclaimPolicy option Retain:

apiVersion: storage.k8s.io/v1
kind: StorageClass
metadata:
  name: premium-rwo-retain
...
reclaimPolicy: Retain
volumeBindingMode: WaitForFirstConsumer

The following example shows the StorageClass added to the spec of a Kafka cluster custom resource:

...
spec:
  ...
  dataVolumeCapacity: 100Gi
  storageClass:
  name: premium-rwo-retain

With this configuration, PersistentVolumes provisioned using the storage class are not deleted even when the corresponding PersistentVolumeClaim is deleted.

To recover the Kafka instance on a new Kubernetes cluster using the existing configuration and broker instance data:

1. Apply the existing Confluent custom resources (Kafka, KafkaTopic, Zookeeper, etc.) to a new Kubernetes cluster
2. Update the PersistentVolumeClaims with the name of the new Kafka broker instances to the old PersistentVolumes using the spec.volumeName property on the PersistentVolumeClaim.