Déployer un cluster Kafka à haute disponibilité sur GKE

Autopilot Standard

---

Kafka est un système de messagerie Pub/Sub distribué Open Source permettant de gérer des données en streaming volumineuses, à haut débit et en temps réel. Kafka vous permet de créer des pipelines de données en streaming qui déplacent des données de manière fiable entre différents systèmes et applications à des fins de traitement et d'analyse.

Ce tutoriel est destiné aux administrateurs de plate-forme, aux architectes cloud et aux professionnels des opérations qui souhaitent déployer des clusters Kafka à disponibilité élevée sur Google Kubernetes Engine (GKE).

Objectifs

Dans ce tutoriel, vous allez apprendre à effectuer les opérations suivantes :

• Utiliser Terraform pour créer un cluster GKE régional.
• Déployer un cluster Kafka à haute disponibilité
• Mettez à niveau les binaires Kafka.
• Sauvegardez et restaurez le cluster Kafka.
• Simulez une interruption de nœud GKE et un basculement pour l'agent Kafka.

Architecture

Cette section décrit l'architecture de la solution que vous allez créer dans ce tutoriel.

Un cluster Kafka est un groupe d'un ou de plusieurs serveurs (appelés agents ou brokers) qui fonctionnent ensemble pour gérer les flux de données entrants et la messagerie Pub/Sub pour les clients Kafka (appelés consommateurs ou consumers).

Chaque partition de données d'un cluster Kafka possède un agent principal et peut avoir un ou plusieurs agents subordonnés. L'agent principal gère toutes les lectures et écritures sur la partition. Chaque agent subordonné réplique l'agent principal de manière passive.

Dans une configuration Kafka classique, vous utilisez également un service Open Source appelé ZooKeeper pour coordonner vos clusters Kafka. Ce service permet de désigner un agent principal parmi les agents et de déclencher un basculement en cas d'échec.

Dans ce tutoriel, vous allez déployer les clusters Kafka sur GKE en configurant les agents Kafka et le service Zookeeper en tant qu'objets StatefulSets individuels. Pour provisionner des clusters Kafka à haute disponibilité et préparer la reprise après sinistre, vous devez configurer vos StatefulSets Kafka et Zookeeper pour utiliser des pools de nœuds et des zones distincts.

Le schéma suivant montre comment le StatefulSet Kafka s'exécute sur plusieurs nœuds et zones de votre cluster GKE.

Figure 1 : Déploiement de votre StatefulSet Kafka sur des nœuds GKE dans trois zones différentes.

Le schéma suivant montre comment votre StatefulSet Zookeeper s'exécute sur plusieurs nœuds et zones de votre cluster GKE.

Figure 2 : Déploiement de votre ZooKeeper Kafka sur des nœuds GKE dans trois zones différentes

Provisionnement des nœuds et planification des pods

Si vous utilisez des clusters Autopilot, Autopilot gère le provisionnement des nœuds et la planification des pods pour vos charges de travail. Vous allez utiliser l'anti-affinité de pod pour vous assurer que deux pods du même StatefulSet ne sont pas programmés sur le même nœud et la même zone.

Si vous utilisez des clusters standards, vous devez configurer la tolérance et l'affinité de nœud du pod. Pour en savoir plus, consultez la page Isoler des charges de travail dans des pools de nœuds dédiés.

Coûts

Dans ce document, vous utilisez les composants facturables suivants de Google Cloud :

• Artifact Registry
• Backup for GKE
• Compute Engine
• GKE
• Google Cloud Managed Service for Prometheus

Obtenez une estimation des coûts en fonction de votre utilisation prévue à l'aide du simulateur de coût. Les nouveaux utilisateurs de Google Cloud peuvent bénéficier d'un essai gratuit.

Une fois que vous avez terminé les tâches décrites dans ce document, vous pouvez éviter de continuer à payer des frais en supprimant les ressources que vous avez créées. Pour en savoir plus, consultez la section Effectuer un nettoyage.

Avant de commencer

Configurer votre projet

1. Sign in to your Google Cloud account. If you're new to Google Cloud, create an account to evaluate how our products perform in real-world scenarios. New customers also get $300 in free credits to run, test, and deploy workloads.

2. In the Google Cloud console, on the project selector page, click Create project to begin creating a new Google Cloud project.

   Go to project selector

3. Make sure that billing is enabled for your Google Cloud project.

4. Enable the Google Kubernetes Engine, Backup for GKE, Artifact Registry, Compute Engine, and IAM APIs.

   Enable the APIs

5. In the Google Cloud console, on the project selector page, click Create project to begin creating a new Google Cloud project.

   Go to project selector

6. Make sure that billing is enabled for your Google Cloud project.

7. Enable the Google Kubernetes Engine, Backup for GKE, Artifact Registry, Compute Engine, and IAM APIs.

   Enable the APIs

Configurer les rôles

1. Grant roles to your user account. Run the following command once for each of the following IAM roles: role/storage.objectViewer, role/logging.logWriter, role/artifactregistry.Admin, roles/container.clusterAdmin, role/container.serviceAgent, roles/iam.serviceAccountAdmin, roles/serviceusage.serviceUsageAdmin, roles/iam.serviceAccountAdmin

   gcloud projects add-iam-policy-binding PROJECT_ID --member="user:USER_IDENTIFIER" --role=ROLE

   • Replace PROJECT_ID with your project ID.

   • Replace USER_IDENTIFIER with the identifier for your user account. For example, user:[e-mail].

   • Replace ROLE with each individual role.

Configurer votre environnement

Dans ce tutoriel, vous utilisez Cloud Shell pour gérer les ressources hébergées sur Google Cloud. Cloud Shell est préinstallé avec les logiciels dont vous avez besoin pour ce tutoriel, y compris Docker, kubectl, gcloud CLI, Helm et Terraform.

Pour configurer votre environnement avec Cloud Shell, procédez comme suit :

1. Lancez une session Cloud Shell depuis la console Google Cloud en cliquant sur  Activer Cloud Shell dans la console Google Cloud. Une session s'ouvre dans le volet inférieur de la console Google Cloud.

2. Définissez les variables d'environnement.

   export PROJECT_ID=PROJECT_ID
   export REGION=us-central1
   

   Remplacez les valeurs suivantes :

   • PROJECT_ID : ID de votre projet Google Cloud.

3. Définissez les variables d'environnement par défaut.

   gcloud config set project PROJECT_ID
   

4. Clonez le dépôt de code.

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

5. Accédez au répertoire de travail.

   cd kubernetes-engine-samples/streaming/gke-stateful-kafka
   

Créer l'infrastructure de votre cluster

Dans cette section, vous allez exécuter un script Terraform pour créer deux clusters GKE régionaux. Le cluster principal sera déployé dans us-central1.

Remarque : Comme vous utiliserez Google Cloud Managed Service pour Prometheus et Sauvegarde pour GKE pour ce tutoriel, la création et le rapprochement du cluster peuvent prendre jusqu'à huit minutes.

Pour créer le cluster, procédez comme suit :

Autopilot

Dans Cloud Shell, exécutez les commandes suivantes :

terraform -chdir=terraform/gke-autopilot init
terraform -chdir=terraform/gke-autopilot apply -var project_id=$PROJECT_ID

Lorsque vous y êtes invité, saisissez yes.

Standard

Dans Cloud Shell, exécutez les commandes suivantes :

terraform -chdir=terraform/gke-standard init
terraform -chdir=terraform/gke-standard apply -var project_id=$PROJECT_ID 

Lorsque vous y êtes invité, saisissez yes.

Les fichiers de configuration Terraform créent les ressources suivantes pour déployer votre infrastructure :

• Ils créent un dépôt Artifact Registry pour stocker les images Docker.
• Ils créent le réseau et le sous-réseau VPC pour l'interface réseau de la VM.
• Ils créent deux clusters GKE.

Terraform crée un cluster privé dans les deux régions et active le service de Sauvegarde pour GKE pour la reprise après sinistre.

Conseil : Pour déboguer les problèmes liés à l'exécution de Terraform, vous pouvez capturer la sortie de débogage en définissant la variable d'environnement de niveau de journalisation Terraform TF_LOG. Exemple : export TF_LOG="DEBUG". Les niveaux de journalisation valides sont (par ordre décroissant de verbosité) : TRACE, DEBUG, INFO, WARN ou ERROR.

Déployer Kafka sur votre cluster

Dans cette section, vous allez déployer Kafka sur GKE à l'aide d'un chart Helm. L'opération crée les ressources suivantes :

• Les StatefulSets de Kafka et Zookeeper.
• Un déploiement de l'exportateur Kafka. L'exportateur collecte les métriques Kafka pour la consommation de Prometheus.
• Un budget d'interruptions de pod qui limite le nombre de pods hors connexion lors d'une interruption volontaire.

Pour déployer Kafka à l'aide du chart Helm, procédez comme suit :

1. Configurez l'accès à Docker.

   gcloud auth configure-docker us-docker.pkg.dev
   

2. Renseignez Artifact Registry avec les images Kafka et Zookeeper.

   ./scripts/gcr.sh bitnami/kafka 3.3.2-debian-11-r0
   ./scripts/gcr.sh bitnami/kafka-exporter 1.6.0-debian-11-r52
   ./scripts/gcr.sh bitnami/jmx-exporter 0.17.2-debian-11-r41
   ./scripts/gcr.sh bitnami/zookeeper 3.8.0-debian-11-r74
   

3. Configurez l'accès via la ligne de commande kubectl au cluster principal.

   gcloud container clusters get-credentials gke-kafka-us-central1 \
       --region=${REGION} \
       --project=${PROJECT_ID}
   

4. Créez un espace de noms.

   export NAMESPACE=kafka
   kubectl create namespace $NAMESPACE
   

5. Installez Kafka à l'aide du chart Helm version 20.0.6.

   cd helm
   ../scripts/chart.sh kafka 20.0.6 && \
   rm -rf Chart.lock charts && \
   helm dependency update && \
   helm -n kafka upgrade --install kafka . \
   --set global.imageRegistry="us-docker.pkg.dev/$PROJECT_ID/main"
   

   Le résultat ressemble à ce qui suit :

   NAME: kafka
   LAST DEPLOYED: Thu Feb 16 03:29:39 2023
   NAMESPACE: kafka
   STATUS: deployed
   REVISION: 1
   TEST SUITE: None
   

6. Vérifiez que vos instances dupliquées Kafka sont en cours d'exécution (cela peut prendre quelques minutes).

   kubectl get all -n kafka
   

   Le résultat ressemble à ce qui suit :

   ---
   NAME                    READY   STATUS    RESTARTS        AGE
   pod/kafka-0             1/1     Running   2 (3m51s ago)   4m28s
   pod/kafka-1             1/1     Running   3 (3m41s ago)   4m28s
   pod/kafka-2             1/1     Running   2 (3m57s ago)   4m28s
   pod/kafka-zookeeper-0   1/1     Running   0               4m28s
   pod/kafka-zookeeper-1   1/1     Running   0               4m28s
   pod/kafka-zookeeper-2   1/1     Running   0               4m28s
   
   NAME                                   TYPE        CLUSTER-IP        EXTERNAL-IP   PORT(S)                      AGE
   service/kafka                          ClusterIP   192.168.112.124   <none>        9092/TCP                     4m29s
   service/kafka-app                      ClusterIP   192.168.75.57     <none>        9092/TCP                     35m
   service/kafka-app-headless             ClusterIP   None              <none>        9092/TCP,9093/TCP            35m
   service/kafka-app-zookeeper            ClusterIP   192.168.117.102   <none>        2181/TCP,2888/TCP,3888/TCP   35m
   service/kafka-app-zookeeper-headless   ClusterIP   None              <none>        2181/TCP,2888/TCP,3888/TCP   35m
   service/kafka-headless                 ClusterIP   None              <none>        9092/TCP,9093/TCP            4m29s
   service/kafka-zookeeper                ClusterIP   192.168.89.249    <none>        2181/TCP,2888/TCP,3888/TCP   4m29s
   service/kafka-zookeeper-headless       ClusterIP   None              <none>        2181/TCP,2888/TCP,3888/TCP   4m29s
   
   NAME                               READY   AGE
   statefulset.apps/kafka             3/3     4m29s
   statefulset.apps/kafka-zookeeper   3/3     4m29s
   

Créer des données test

Dans cette section, vous allez tester l'application Kafka et générer des messages.

1. Créez un pod client consommateur pour interagir avec l'application Kafka.

   kubectl run kafka-client -n kafka --rm -ti \
       --image us-docker.pkg.dev/$PROJECT_ID/main/bitnami/kafka:3.3.2-debian-11-r0 -- bash
   

2. Créez un sujet nommé topic1 avec trois partitions et un facteur de réplication de trois.

   kafka-topics.sh \
       --create \
       --topic topic1 \
       --partitions 3  \
       --replication-factor 3 \
       --bootstrap-server kafka-headless.kafka.svc.cluster.local:9092
   

3. Vérifiez que les partitions du sujet sont bien répliquées sur les trois agents.

   kafka-topics.sh \
       --describe \
       --topic topic1 \
       --bootstrap-server kafka-headless.kafka.svc.cluster.local:9092
   

   Le résultat ressemble à ce qui suit :

   Topic: topic1     TopicId: 1ntc4WiFS4-AUNlpr9hCmg PartitionCount: 3       ReplicationFactor: 3    Configs: flush.ms=1000,segment.bytes=1073741824,flush.messages=10000,max.message.bytes=1000012,retention.bytes=1073741824
          Topic: topic1    Partition: 0    Leader: 2       Replicas: 2,0,1 Isr: 2,0,1
          Topic: topic1    Partition: 1    Leader: 1       Replicas: 1,2,0 Isr: 1,2,0
          Topic: topic1    Partition: 2    Leader: 0       Replicas: 0,1,2 Isr: 0,1,2
   

   Dans l'exemple de résultat, notez que topic1 comporte trois partitions, chacune avec un agent principal et un ensemble d'instances dupliquées différents. En effet, Kafka utilise le partitionnement pour distribuer les données sur plusieurs agents, ce qui permet une plus grande évolutivité et une tolérance aux pannes. Le facteur de réplication de trois garantit que chaque partition possède trois instances dupliquées, de sorte que les données soient toujours disponibles même en cas de défaillance d'un ou deux agents.

4. Exécutez la commande suivante pour générer des numéros de message de façon groupée dans topic1.

   ALLOW_PLAINTEXT_LISTENER=yes
   for x in $(seq 0 200); do
     echo "$x: Message number $x"
   done | kafka-console-producer.sh \
       --topic topic1 \
       --bootstrap-server kafka-headless.kafka.svc.cluster.local:9092 \
       --property parse.key=true \
       --property key.separator=":"
   

5. Exécutez la commande suivante pour consommer topic1 à partir de toutes les partitions.

   kafka-console-consumer.sh \
       --bootstrap-server kafka.kafka.svc.cluster.local:9092 \
       --topic topic1 \
       --property print.key=true \
       --property key.separator=" : " \
       --from-beginning;
   

   Saisissez CTRL+C pour arrêter le processus destiné au consommateur.

Benchmark Kafka

Pour modéliser un cas d'utilisation avec précision, vous pouvez exécuter une simulation de la charge attendue sur le cluster. Pour tester les performances, vous utiliserez les outils inclus dans le package Kafka, à savoir les scripts kafka-producer-perf-test.sh et kafka-consumer-perf-test.sh du dossier bin.

1. Créez un sujet pour l'analyse comparative.

   kafka-topics.sh \
     --create \
     --topic topic-benchmark \
     --partitions 3  \
     --replication-factor 3 \
     --bootstrap-server kafka-headless.kafka.svc.cluster.local:9092
   

2. Créez une charge sur le cluster Kafka.

   KAFKA_HEAP_OPTS="-Xms4g -Xmx4g" kafka-producer-perf-test.sh \
       --topic topic-benchmark \
       --num-records 10000000 \
       --throughput -1 \
       --producer-props bootstrap.servers=kafka.kafka.svc.cluster.local:9092 \
             batch.size=16384 \
             acks=all \
             linger.ms=500 \
             compression.type=none \
       --record-size 100 \
       --print-metrics
   

   Le producteur génère 10 000 000 d'enregistrements sur topic-benchmark. Le résultat ressemble à ce qui suit :

   623821 records sent, 124316.7 records/sec (11.86 MB/sec), 1232.7 ms avg latency, 1787.0 ms max latency.
   1235948 records sent, 247140.2 records/sec (23.57 MB/sec), 1253.0 ms avg latency, 1587.0 ms max latency.
   1838898 records sent, 367779.6 records/sec (35.07 MB/sec), 793.6 ms avg latency, 1185.0 ms max latency.
   2319456 records sent, 463242.7 records/sec (44.18 MB/sec), 54.0 ms avg latency, 321.0 ms max latency.
   

   Une fois tous les enregistrements envoyés, des métriques supplémentaires devraient s'afficher dans le résultat, semblables à ce qui suit :

   producer-topic-metrics:record-send-rate:{client-id=perf-producer-client, topic=topic-benchmark}     : 173316.233
   producer-topic-metrics:record-send-total:{client-id=perf-producer-client, topic=topic-benchmark}    : 10000000.000
   

   Pour quitter la lecture, appuyez sur CTRL + C.

3. Quittez l'interface système du pod.

   exit
   

Gérer les mises à jour

Les mises à jour de Kafka et de Kubernetes sont publiées selon un calendrier régulier. Suivez les bonnes pratiques opérationnelles pour mettre régulièrement à jour votre environnement logiciel.

Remarque : Par défaut, GKE gère automatiquement les mises à jour des clusters et des pools de nœuds. Les clusters Autopilot sont automatiquement mis à jour en fonction de la version disponible sélectionnée. Si vous utilisez des clusters standards et que vous avez désactivé les mises à jour automatiques, consultez la page Mettre à jour manuellement un cluster ou un pool de nœuds.

Planifier les mises à jour des binaires Kafka

Dans cette section, vous allez mettre à jour l'image Kafka à l'aide de Helm et vérifier que vos sujets sont toujours disponibles.

Pour effectuer une mise à jour depuis la version précédente de Kafka à partir du chart Helm que vous avez utilisé dans la section Déployer Kafka sur votre cluster, procédez comme suit :

1. Renseignez Artifact Registry avec l'image suivante :

   ../scripts/gcr.sh bitnami/kafka 3.4.0-debian-11-r2
   ../scripts/gcr.sh bitnami/kafka-exporter 1.6.0-debian-11-r61
   ../scripts/gcr.sh bitnami/jmx-exporter 0.17.2-debian-11-r49
   ../scripts/gcr.sh bitnami/zookeeper 3.8.1-debian-11-r0
   

2. Procédez comme suit pour déployer un chart Helm avec les images Kafka et Zookeeper mises à jour. Pour obtenir des conseils spécifiques à la version, reportez-vous aux instructions Kafka pour les mises à jour de version.

   1. Mettez à jour la version de la dépendance Chart.yaml :

   ../scripts/chart.sh kafka 20.1.0
   

   1. Déployez le chart Helm avec les nouvelles images Kafka et Zookeeper, comme illustré dans l'exemple suivant :

      rm -rf Chart.lock charts && \
      helm dependency update && \
      helm -n kafka upgrade --install kafka ./ \
            --set global.imageRegistry="$REGION-docker.pkg.dev/$PROJECT_ID/main"
      

   Regardez la mise à jour des pods Kafka :

   kubectl get pod -l app.kubernetes.io/component=kafka -n kafka --watch
   

   Pour quitter la lecture, appuyez sur CTRL + C.

3. Connectez-vous au cluster Kafka à l'aide d'un pod client.

   kubectl run kafka-client -n kafka --rm -ti \
     --image us-docker.pkg.dev/$PROJECT_ID/main/bitnami/kafka:3.4.0-debian-11-r2 -- bash
   

4. Vérifiez que vous pouvez accéder aux messages de topic1.

   kafka-console-consumer.sh \
     --topic topic1 \
     --from-beginning \
     --bootstrap-server kafka-headless.kafka.svc.cluster.local:9092
   

   Le résultat doit afficher les messages générés à l'étape précédente. Saisissez CTRL+C pour quitter le processus.

5. Quittez le pod client.

   exit
   

Préparer la reprise après sinistre

Pour garantir la disponibilité de vos charges de travail de production en cas d'interruption de service, vous devez préparer un plan de reprise après sinistre. Pour en savoir plus sur la planification de reprise après sinistre, consultez le Guide de planification de reprise après sinistre.

Pour sauvegarder et restaurer vos charges de travail sur des clusters GKE, vous pouvez utiliser le service de Sauvegarde pour GKE.

Conseil : Si vous utilisez des clusters Autopilot, vérifiez que votre infrastructure fonctionne avec ces Restrictions de Sauvegarde pour GKE.

Exemple de scénario de sauvegarde et de restauration Kafka

Dans cette section, vous allez effectuer une sauvegarde de votre cluster à partir de gke-kafka-us-central1 et restaurer la sauvegarde dans gke-kafka-us-west1. Vous allez effectuer l'opération de sauvegarde et de restauration au niveau de l'application à l'aide de la ressource personnalisée ProtectedApplication.

Le schéma suivant montre les composants de la solution de reprise après sinistre et décrit leur relation.

Figure 3 : Exemple de solution de sauvegarde et de récupération pour un cluster Kafka à haute disponibilité.

Pour préparer la sauvegarde et la restauration de votre cluster Kafka, procédez comme suit :

1. Configurez les variables d'environnement.

   export BACKUP_PLAN_NAME=kafka-protected-app
   export BACKUP_NAME=protected-app-backup-1
   export RESTORE_PLAN_NAME=kafka-protected-app
   export RESTORE_NAME=protected-app-restore-1
   export REGION=us-central1
   export DR_REGION=us-west1
   export CLUSTER_NAME=gke-kafka-$REGION
   export DR_CLUSTER_NAME=gke-kafka-$DR_REGION
   

2. Vérifiez que le cluster est à l'état RUNNING.

   gcloud container clusters describe $CLUSTER_NAME --region us-central1 --format='value(status)'
   

   Remarque : Le cluster doit être à l'état RUNNING pour que vous puissiez créer une sauvegarde.

3. Créez un plan de sauvegarde.

   gcloud beta container backup-restore backup-plans create $BACKUP_PLAN_NAME \
       --project=$PROJECT_ID \
       --location=$DR_REGION \
       --cluster=projects/$PROJECT_ID/locations/$REGION/clusters/$CLUSTER_NAME \
       --selected-applications=kafka/kafka,kafka/zookeeper \
       --include-secrets \
       --include-volume-data \
       --cron-schedule="0 3 * * *" \
       --backup-retain-days=7 \
       --backup-delete-lock-days=0
   

4. Créez manuellement une sauvegarde. Bien que les sauvegardes planifiées soient généralement régies par la planification Cron dans le plan de sauvegarde, l'exemple suivant montre comment lancer une opération de sauvegarde ponctuelle.

   gcloud beta container backup-restore backups create $BACKUP_NAME \
       --project=$PROJECT_ID \
       --location=$DR_REGION \
       --backup-plan=$BACKUP_PLAN_NAME \
       --wait-for-completion
   

5. Créez un plan de restauration.

   gcloud beta container backup-restore restore-plans create $RESTORE_PLAN_NAME \
       --project=$PROJECT_ID \
       --location=$DR_REGION \
       --backup-plan=projects/$PROJECT_ID/locations/$DR_REGION/backupPlans/$BACKUP_PLAN_NAME \
       --cluster=projects/$PROJECT_ID/locations/$DR_REGION/clusters/$DR_CLUSTER_NAME \
       --cluster-resource-conflict-policy=use-existing-version \
       --namespaced-resource-restore-mode=delete-and-restore \
       --volume-data-restore-policy=restore-volume-data-from-backup \
       --selected-applications=kafka/kafka,kafka/zookeeper \
       --cluster-resource-scope-selected-group-kinds="storage.k8s.io/StorageClass"
   

6. Effectuez une restauration manuelle à partir d'une sauvegarde.

   gcloud beta container backup-restore restores create $RESTORE_NAME \
       --project=$PROJECT_ID \
       --location=$DR_REGION \
       --restore-plan=$RESTORE_PLAN_NAME \
       --backup=projects/$PROJECT_ID/locations/$DR_REGION/backupPlans/$BACKUP_PLAN_NAME/backups/$BACKUP_NAME
   

7. Regardez l'application restaurée apparaître dans le cluster de sauvegarde. L'exécution et la préparation de tous les pods peuvent prendre quelques minutes.

   gcloud container clusters get-credentials gke-kafka-us-west1 \
       --region us-west1
   kubectl get pod -n kafka --watch
   

   Appuyez sur CTRL+C pour quitter la lecture lorsque tous les pods sont opérationnels.

8. Vérifiez que les sujets précédents peuvent être récupérés par un consommateur.

   kubectl run kafka-client -n kafka --rm -ti \
       --image us-docker.pkg.dev/$PROJECT_ID/main/bitnami/kafka:3.4.0 -- bash
   

   kafka-console-consumer.sh \
       --bootstrap-server kafka.kafka.svc.cluster.local:9092 \
       --topic topic1 \
       --property print.key=true \
       --property key.separator=" : " \
       --from-beginning;
   

   Le résultat ressemble à ce qui suit :

   192 :  Message number 192
   193 :  Message number 193
   197 :  Message number 197
   200 :  Message number 200
   Processed a total of 201 messages
   

   Saisissez CTRL+C pour quitter le processus.

9. Fermez le pod.

   exit
   

Simuler une interruption de service Kafka

Dans cette section, vous allez simuler une défaillance d'un nœud en remplaçant un nœud Kubernetes hébergeant l'agent. Cette section ne s'applique qu'à l'édition standard. Autopilot gère vos nœuds automatiquement. Par conséquent, les défaillances de nœuds ne peuvent pas être simulées.

1. Créez un pod client pour vous connecter à l'application Kafka.

   kubectl run kafka-client -n kafka --restart='Never' -it \
   --image us-docker.pkg.dev/$PROJECT_ID/main/bitnami/kafka:3.4.0 -- bash
   

2. Créez le sujet topic-failover-test et générez du trafic de test.

   kafka-topics.sh \
     --create \
     --topic topic-failover-test \
     --partitions 1  \
     --replication-factor 3  \
     --bootstrap-server kafka-headless.kafka.svc.cluster.local:9092
   

3. Déterminez l'agent principal du sujet topic-failover-test.

   kafka-topics.sh --describe \
     --topic topic-failover-test \
     --bootstrap-server kafka-headless.kafka.svc.cluster.local:9092
   

   Le résultat ressemble à ce qui suit :

   Topic: topic-failover-test     Partition: 0    Leader: 1       Replicas: 1,0,2 Isr: 1,0,2
   

   Dans le résultat ci-dessus, Leader: 1 signifie que l'agent principal pour topic-failover-test est l'agent 1. Cela correspond au pod kafka-1.

4. Ouvrez un nouveau terminal et connectez-vous au même cluster.

   gcloud container clusters get-credentials gke-kafka-us-west1 --region us-west1 --project PROJECT_ID
   

5. Recherchez le nœud sur lequel le pod kafka-1 s'exécute.

   kubectl get pod -n kafka kafka-1 -o wide
   

   Le résultat ressemble à ce qui suit :

   NAME      READY   STATUS    RESTARTS      AGE   IP              NODE                                               NOMINATED NODE   READINESS GATES
   kafka-1   2/2     Running   1 (35m ago)   36m   192.168.132.4   gke-gke-kafka-us-west1-pool-system-a0d0d395-nx72   <none>           <none>
   

   Dans le résultat ci-dessus, le pod kafka-1 s'exécute sur le nœud gke-gke-kafka-us-west1-pool-system-a0d0d395-nx72.

6. Drainez le nœud pour évincer les pods.

   kubectl drain NODE \
     --delete-emptydir-data \
     --force \
     --ignore-daemonsets
   

   Remplacez NODE par le nœud sur lequel le pod kafka-1 s'exécute. Dans cet exemple, le nœud est gke-gke-kafka-us-west1-pool-system-a0d0d395-nx72.

   Le résultat ressemble à ce qui suit :

   node/gke-gke-kafka-us-west1-pool-system-a0d0d395-nx72 cordoned
   Warning: ignoring DaemonSet-managed Pods: gmp-system/collector-gjzsd, kube-system/calico-node-t28bj, kube-system/fluentbit-gke-lxpft, kube-system/gke-metadata-server-kxw78, kube-system/ip-masq-agent-kv2sq, kube-system/netd-h446k, kube-system/pdcsi-node-ql578
   evicting pod kafka/kafka-1
   evicting pod kube-system/kube-dns-7d48cb57b-j4d8f
   evicting pod kube-system/calico-typha-56995c8d85-5clph
   pod/calico-typha-56995c8d85-5clph evicted
   pod/kafka-1 evicted
   pod/kube-dns-7d48cb57b-j4d8f evicted
   node/gke-gke-kafka-us-west1-pool-system-a0d0d395-nx72 drained
   

7. Recherchez le nœud sur lequel le pod kafka-1 s'exécute.

   kubectl get pod -n kafka kafka-1 -o wide
   

   Le résultat doit ressembler à ce qui suit :

   NAME      READY   STATUS    RESTARTS   AGE     IP              NODE                                              NOMINATED NODE   READINESS GATES
   kafka-1   2/2     Running   0          2m49s   192.168.128.8   gke-gke-kafka-us-west1-pool-kafka-700d8e8d-05f7   <none>           <none>
   

   D'après le résultat ci-dessus, l'application s'exécute sur un nouveau nœud.

8. Dans le terminal connecté au pod kafka-client, déterminez quel est l'agent principal pour topic-failover-test.

   kafka-topics.sh --describe \
     --topic topic-failover-test \
     --bootstrap-server kafka-headless.kafka.svc.cluster.local:9092
   

   Le résultat doit ressembler à ce qui suit :

   Topic: topic-failover-test     TopicId: bemKyqmERAuKZC5ymFwsWg PartitionCount: 1       ReplicationFactor: 3    Configs: flush.ms=1000,segment.bytes=1073741824,flush.messages=10000,max.message.bytes=1000012,retention.bytes=1073741824
       Topic: topic-failover-test     Partition: 0    Leader: 1       Replicas: 1,0,2 Isr: 0,2,1
   

   Dans l'exemple de résultat, l'agent principal est toujours 1. Toutefois, il s'exécute désormais sur un nouveau nœud.

Tester la défaillance de l'agent principal Kafka

1. Dans Cloud Shell, connectez-vous au client Kafka et utilisez describe pour afficher l'agent principal désigné pour chaque partition dans topic1.

   kafka-topics.sh --describe \
     --topic topic1 \
     --bootstrap-server kafka-headless.kafka.svc.cluster.local:9092
   

   Le résultat ressemble à ce qui suit :

   Topic: topic1   TopicId: B3Jr_5t2SPq7F1jVHu4r0g PartitionCount: 3       ReplicationFactor: 3    Configs: flush.ms=1000,segment.bytes=1073741824,flush.messages=10000,max.message.bytes=1000012,retention.bytes=1073741824
       Topic: topic1   Partition: 0    Leader: 0       Replicas: 0,2,1 Isr: 0,2,1
       Topic: topic1   Partition: 1    Leader: 0       Replicas: 2,1,0 Isr: 0,2,1
       Topic: topic1   Partition: 2    Leader: 0       Replicas: 1,0,2 Isr: 0,2,1
   

2. Dans Cloud Shell qui n'est pas connecté au client Kafka, supprimez l'agent principal kafka-0 pour forcer une nouvelle élection de l'agent principal. Vous devez supprimer l'index correspondant à l'un des agents principaux dans le résultat précédent.

   kubectl delete pod -n kafka kafka-0 --force
   

   Le résultat ressemble à ce qui suit :

   pod "kafka-0" force deleted
   

3. Dans Cloud Shell connecté au client Kafka, utilisez describe pour afficher l'agent principal désigné.

   kafka-topics.sh --describe \
     --topic topic1 \
     --bootstrap-server kafka-headless.kafka.svc.cluster.local:9092
   

   Le résultat ressemble à ce qui suit :

   Topic: topic1   TopicId: B3Jr_5t2SPq7F1jVHu4r0g PartitionCount: 3       ReplicationFactor: 3    Configs: flush.ms=1000,segment.bytes=1073741824,flush.messages=10000,max.message.bytes=1000012,retention.bytes=1073741824
       Topic: topic1   Partition: 0    Leader: 2       Replicas: 0,1,2 Isr: 2,0,1
       Topic: topic1   Partition: 1    Leader: 2       Replicas: 2,0,1 Isr: 2,0,1
       Topic: topic1   Partition: 2    Leader: 2       Replicas: 1,2,0 Isr: 2,0,1
   

   Dans le résultat, le nouvel agent principal de chaque partition change si elle a été attribuée à l'agent principal interrompu (kafka-0). Cela indique que l'agent principal d'origine a été remplacé lors de la suppression et de la recréation du pod.

Effectuer un nettoyage

Pour éviter que les ressources utilisées lors de ce tutoriel soient facturées sur votre compte Google Cloud, supprimez le projet contenant les ressources, ou conservez le projet et supprimez les ressources individuelles.

Supprimer le projet

Le moyen le plus simple d'empêcher la facturation est de supprimer le projet que vous avez créé pour ce tutoriel.

Attention : La suppression d'un projet entraîne les effets décrits ci-dessous :

• Tout le contenu du projet est supprimé. Si vous avez utilisé un projet existant pour les tâches décrites dans ce document et que vous le supprimez, vous supprimerez également tout autre travail effectué dans le projet.
• Les ID de projets personnalisés sont perdus. Lorsque vous avez créé ce projet, vous avez peut-être créé un ID de projet personnalisé que vous souhaitez utiliser à l'avenir. Pour conserver les URL qui utilisent l'ID de projet, telle qu'une URL appspot.com, supprimez les ressources sélectionnées dans le projet au lieu de supprimer l'ensemble du projet.

Si vous envisagez d'explorer plusieurs architectures, tutoriels et guides de démarrage rapide, réutiliser des projets peut vous aider à ne pas dépasser les limites de quotas des projets.

Delete a Google Cloud project:

gcloud projects delete PROJECT_ID

Étapes suivantes

• Pour découvrir un service de messagerie entièrement géré et évolutif, consultez la page Migrer de Kafka vers Pub/Sub.