Deploy Apache Kafka to GKE using Strimzi

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 strimzi-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 strimzi-cluster --region us-central1"

Connecting to the cluster

Configure kubectl to communicate with the cluster:

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

Deploy the Strimzi operator to your cluster

In this section, you deploy the Strimzi operator using a Helm chart. There are also several other ways to deploy Strimzi.

1. Add the Strimzi Helm Chart repository:

   helm repo add strimzi https://strimzi.io/charts/
   

2. Add a namespace for the Strimzi Operator and the Kafka cluster:

   kubectl create ns kafka
   

3. Deploy the Strimzi cluster operator using Helm:

   helm install strimzi-operator strimzi/strimzi-kafka-operator -n kafka
   

   To deploy Strimzi Cluster Operator and Kafka clusters to different namespaces, add the parameter --set watchNamespaces="{kafka-namespace,kafka-namespace-2,...}" to the helm command.

4. Verify that the Strimzi Cluster 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
   strimzi-operator    kafka      1       2023-06-27 11:22:15.850545 +0200 CEST    deployed    strimzi-kafka-operator-0.35.0    0.35.0
   

Deploy Kafka

After the operator is deployed to the cluster, you are ready to deploy a Kafka cluster instance.

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 Strimzi.
• Java MaxHeapSize and MinHeapSize set to 4 GB for Kafka and 2 GB for ZooKeeper.
• CPU resource allocation of 1 CPU request and 2 CPU limits both for Kafka and ZooKeeper, along with 5 GB memory requests and limits for Kafka (4 GB for the main service and 0.5 GB for the metrics exporter) and 2.5 GB for ZooKeeper (2 GB for the main service and 0.5 GB for the metrics exporter).
• Entity-operator with the following requests and limits:
  • tlsSidecar: 100 m/500 m CPU and 128 Mi memory.
  • topicOperator: 100 m/500 m CPU and 512 Mi memory.
  • userOperator: 500 m CPU and 2 Gi memory.
• 100 GB of storage allocated to each Pod using the premium-rwo storageClass.
• 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: separate Certificate Authorities (CAs) for cluster and clients (mTLS). You can also configure to use a different Certificate Authority.

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. Create a new Kafka cluster using the basic configuration:

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

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

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

   kubectl wait kafka/my-cluster --for=condition=Ready --timeout=600s -n kafka
   

3. Verify that the Kafka workloads were created:

   kubectl get pod,service,deploy,pdb -l=strimzi.io/cluster=my-cluster -n kafka
   

   The output is similar to the following:

   NAME                                            READY   STATUS  RESTARTS   AGE
   pod/my-cluster-entity-operator-848698874f-j5m7f   3/3   Running   0        44m
   pod/my-cluster-kafka-0                          1/1   Running   0        5m
   pod/my-cluster-kafka-1                          1/1   Running   0        5m
   pod/my-cluster-kafka-2                          1/1   Running   0        5m
   pod/my-cluster-zookeeper-0                      1/1   Running   0        6m
   pod/my-cluster-zookeeper-1                      1/1   Running   0        6m
   pod/my-cluster-zookeeper-2                      1/1   Running   0        6m
   
   NAME                                TYPE      CLUSTER-IP   EXTERNAL-IP   PORT(S)                             AGE
   service/my-cluster-kafka-bootstrap  ClusterIP   10.52.8.80   <none>      9091/TCP,9092/TCP,9093/TCP          5m
   service/my-cluster-kafka-brokers    ClusterIP   None         <none>      9090/TCP,9091/TCP,9092/TCP,9093/TCP   5m
   service/my-cluster-zookeeper-client   ClusterIP   10.52.11.144   <none>      2181/TCP                            6m
   service/my-cluster-zookeeper-nodes  ClusterIP   None         <none>      2181/TCP,2888/TCP,3888/TCP          6m
   
   NAME                                       READY   UP-TO-DATE   AVAILABLE   AGE
   deployment.apps/my-cluster-entity-operator   1/1   1          1         44m
   
   NAME                                            MIN AVAILABLE   MAX UNAVAILABLE   ALLOWED DISRUPTIONS   AGE
   poddisruptionbudget.policy/my-cluster-kafka     2             N/A             1                   5m
   poddisruptionbudget.policy/my-cluster-zookeeper   2             N/A             1                   6m
   

The operator creates the following resources:

• Two StrimziPodSets for Kafka and ZooKeeper.
• Three Pods for Kafka broker replicas.
• Three Pods for ZooKeeper replicas.
• Two PodDisruptionBudgets, ensuring a minimum availability of two replicas for cluster consistency.
• A Service named my-cluster-kafka-bootstrap, which serves as the bootstrap server for Kafka clients connecting from within the Kubernetes cluster. All internal Kafka listeners are available in this Service.
• A headless Service named my-cluster-kafka-brokers that enables DNS resolution of Kafka broker Pod IP addresses directly. This service is used for inter broker communication.
• A Service named my-cluster-zookeeper-client that lets Kafka brokers connect to ZooKeeper nodes as clients.
• A headless Service named my-cluster-zookeeper-nodes that enables DNS resolution of ZooKeeper Pod IP addresses directly. This service is used to connect between ZooKeeper replicas.
• A Deployment named my-cluster-entity-operator that contains the topic-operator and user-operator and facilitates the management of custom resources KafkaTopics and KafkaUsers.

You can also configure two NetworkPolicies to facilitate connectivity to Kafka listeners from any Pod and Namespace. These policies would also restrict connections to ZooKeeper to brokers, and enable communication between the cluster Pods and internal Service ports exclusive to cluster communication.

Authentication and user management

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

Strimzi provides a Kubernetes-native method for user management using a separate User Operator and its corresponding Kubernetes custom resource, KafkaUser, which defines the user configuration. The user configuration includes settings for authentication and authorization, and provisions the corresponding user in Kafka.

Strimzi can create Kafka listeners and users that support several authentication mechanisms such as username and password-based authentication (SCRAM-SHA-512) or TLS. You can also use OAuth 2.0 authentication, which is often considered a better approach compared to using passwords or certificates for authentication because of security and external credential management.

Deploy a Kafka cluster

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

• A Kafka cluster with password-based authentication (SCRAM-SHA-512) enabled on one of the listeners.
• A KafkaTopicwith 3 replicas.
• A KafkaUser with an ACL that specifies that user has read and write permissions to the topic.

1. Configure your Kafka cluster to use a listener with password-based SCRAM-SHA-512 authentication on port 9094 and simple authorization:

   kubectl apply -n kafka -f kafka-strimzi/manifests/03-auth/my-cluster.yaml
   

2. Create a Topic, User and a client Pod to execute commands against the Kafka cluster:

   kubectl apply -n kafka -f kafka-strimzi/manifests/03-auth/topic.yaml
   kubectl apply -n kafka -f kafka-strimzi/manifests/03-auth/my-user.yaml
   

   The Secret my-user with the user credentials is mounted to the client Pod as a Volume.

   These credentials confirm that the user has permissions to publish messages to the topic using the listener with the password-based authentication (SCRAM-SHA-512) enabled.

3. Create a client pod:

   kubectl apply -n kafka -f kafka-strimzi/manifests/03-auth/kafkacat.yaml
   

4. Wait a few minutes for the client Pod becomes Ready then connect to it:

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

5. Produce a new message with my-user credentials and try to consume it:

   echo "Message from my-user" |kcat \
     -b my-cluster-kafka-bootstrap.kafka.svc.cluster.local:9094 \
     -X security.protocol=SASL_SSL \
     -X sasl.mechanisms=SCRAM-SHA-512 \
     -X sasl.username=my-user \
     -X sasl.password=$(cat /my-user/password) \
     -t my-topic -P
   kcat -b my-cluster-kafka-bootstrap.kafka.svc.cluster.local:9094 \
     -X security.protocol=SASL_SSL \
     -X sasl.mechanisms=SCRAM-SHA-512 \
     -X sasl.username=my-user \
     -X sasl.password=$(cat /my-user/password) \
     -t my-topic -C
   

   The output is similar to the following:

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

   Type CTRL+C to stop the consumer process.

6. Exit the Pod shell

   exit
   

Backups and disaster recovery

Although the Strimzi operator does not offer built-in backup functionality, you can implement efficient backup strategies by following certain patterns.

You can use Backup for GKE to backup:

• Kubernetes resource manifests.
• Strimzi 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 backup of a Kafka cluster that has been deployed using the Strimzi operator. You should backup:

• The Kafka configuration, which includes all custom resources of the Strimzi API such as KafkaTopicsand KafkaUsers.
• The data, which is stored in the PersistentVolumes of the Kafka brokers.

Storing Kubernetes resource manifests, including Strimzi configurations, in Git repositories can eliminate the need for a separate backup for Kafka configuration, because 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 a 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 recommend 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:
  kafka:
    # ...
    storage:
      type: persistent-claim
      size: 100Gi
      class: 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 Strimzi Kafka custom resources (Kakfa, KafkaTopic, KafkaUser, 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.