Esegui il deployment di un cluster Kafka ad alta disponibilità su GKE

Autopilot Standard

Kafka è un sistema di messaggistica open source distribuito, di sottoscrizione e di pubblicazione, per la gestione di flussi di dati in tempo reale, a elevata velocità effettiva e volumi elevati. Puoi usare Kafka per creare pipeline di dati in modalità flusso che trasferiscono i dati in modo affidabile tra diversi sistemi e applicazioni per l'elaborazione e l'analisi.

Questo tutorial è destinato agli amministratori di piattaforma, ai Cloud Architect e ai professionisti operativi interessati al deployment di cluster Kafka ad alta disponibilità su Google Kubernetes Engine (GKE).

Obiettivi

In questo tutorial imparerai a:

Utilizza Terraform per creare un cluster GKE a livello di regione.
Esegui il deployment di un cluster Kafka ad alta disponibilità.
Esegui l'upgrade dei file binari di Kafka.
Esegui il backup e ripristina il cluster Kafka.
Simula l'interruzione dei nodi GKE e il failover del broker Kafka.

Architettura

Questa sezione descrive l'architettura della soluzione che creerai in questo tutorial.

Un cluster Kafka è un gruppo di uno o più server (chiamati broker) che lavorano insieme per gestire gli stream di dati in entrata e la messaggistica "pubblica-sottoscrizione" per i client Kafka (detti consumer).

Ogni partizione di dati in un cluster Kafka ha un broker leader e può avere uno o più broker di follower. Il broker leader gestisce tutte le operazioni di lettura e scrittura nella partizione. Ogni broker di follower replica passivamente il broker leader.

In una configurazione tipica di Kafka, puoi utilizzare anche un servizio open source chiamato ZooKeeper per coordinare i cluster Kafka. Questo servizio consente di scegliere un leader tra i broker e di attivare il failover in caso di errori.

In questo tutorial, eseguirai il deployment dei cluster Kafka su GKE configurando i broker Kafka e il servizio Zookeeper come singoli StatefulSets. Per eseguire il provisioning di cluster Kafka ad alta disponibilità e prepararti per il ripristino di emergenza, devi configurare gli StatefulSet Kafka e Zookeeper in modo da utilizzare pool di nodi e zone separati.

Il seguente diagramma mostra l'esecuzione dello StatefulSet Kafka su più nodi e zone nel cluster GKE.

Il diagramma mostra un'architettura di esempio di uno StatefulSet Kafka su GKE di cui è stato eseguito il deployment in più zone. — **Figura 1**: deployment di Kafka StatefulSet su nodi GKE in tre zone diverse.

Il seguente diagramma mostra l'esecuzione dello StatefulSet Zookeeper su più nodi e zone nel cluster GKE.

Il diagramma mostra un'architettura di esempio di uno StatefulSet Zookeeper su GKE di cui è stato eseguito il deployment in più zone. — **Figura 2**: deployment di Kafka Zookeeper su nodi GKE in tre zone diverse.

Provisioning dei nodi e pianificazione dei pod

Se utilizzi cluster Autopilot, Autopilot gestisce i nodi di provisioning e la pianificazione dei pod per i tuoi carichi di lavoro. Userai l'anti-affinità dei pod per assicurarti che non vengano pianificati due pod dello stesso StatefulSet sullo stesso nodo e nella stessa zona.

Se utilizzi cluster standard, devi configurare la tolleranza e l'affinità dei nodi dei pod. Per saperne di più, consulta Isolare i carichi di lavoro in pool di nodi dedicati.

Costi

In questo documento vengono utilizzati i seguenti componenti fatturabili di Google Cloud:

Per generare una stima dei costi in base all'utilizzo previsto, utilizza il Calcolatore prezzi. I nuovi utenti di Google Cloud possono essere idonei a una prova senza costi aggiuntivi.

Una volta completate le attività descritte in questo documento, puoi evitare la fatturazione continua eliminando le risorse che hai creato. Per ulteriori informazioni, consulta la pagina Pulizia.

Prima di iniziare

Configura il progetto

Accedi al tuo account Google Cloud. Se non conosci Google Cloud, crea un account per valutare le prestazioni dei nostri prodotti in scenari reali. I nuovi clienti ricevono anche 300 $di crediti gratuiti per l'esecuzione, il test e il deployment dei carichi di lavoro.

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

Assicurati che la fatturazione sia attivata per il tuo progetto Google Cloud.

Abilita le API Google Kubernetes Engine, Backup for GKE, Artifact Registry, Compute Engine, and IAM.

Abilita le API

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

Assicurati che la fatturazione sia attivata per il tuo progetto Google Cloud.

Abilita le API Google Kubernetes Engine, Backup for GKE, Artifact Registry, Compute Engine, and IAM.

Abilita le API

Configura i ruoli

Concedi i ruoli al tuo Account Google. Esegui questo comando una volta per ciascuno dei seguenti ruoli IAM: 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:EMAIL_ADDRESS" --role=ROLE
```
- Sostituisci PROJECT_ID con l'ID progetto.
- Sostituisci EMAIL_ADDRESS con il tuo indirizzo email.
- Sostituisci ROLE con ogni singolo ruolo.

configura l'ambiente

In questo tutorial utilizzerai Cloud Shell per gestire le risorse ospitate su Google Cloud. Cloud Shell è preinstallato con il software necessario per questo tutorial, tra cui Docker, kubectl, gcloud CLI, Helm e Terraform.

Per configurare il tuo ambiente con Cloud Shell, segui questi passaggi:

Avvia una sessione di Cloud Shell dalla console Google Cloud, facendo clic su Attiva Cloud Shell nella console Google Cloud. In questo modo viene avviata una sessione nel riquadro inferiore della console Google Cloud.
Imposta le variabili di ambiente.
```
export PROJECT_ID=PROJECT_ID
export REGION=us-central1
```
Sostituisci i seguenti valori:
- PROJECT_ID: il tuo ID progetto Google Cloud.
Imposta le variabili di ambiente predefinite.
```
gcloud config set project PROJECT_ID
```

clona il repository di codice.

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

Passa alla directory di lavoro.

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

Crea l'infrastruttura del cluster

In questa sezione eseguirai uno script Terraform per creare due cluster GKE a livello di regione. Il deployment del cluster primario verrà eseguito in us-central1.

Per creare il cluster, segui questi passaggi:

Autopilot

In Cloud Shell, esegui questi comandi:

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

Quando richiesto, digita yes.

Standard

In Cloud Shell, esegui questi comandi:

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

Quando richiesto, digita yes.

I file di configurazione Terraform creano le seguenti risorse per il deployment dell'infrastruttura:

Crea un repository Artifact Registry per l'archiviazione delle immagini Docker.
Crea la rete VPC e la subnet per l'interfaccia di rete della VM.
Creare due cluster GKE.

Terraform crea un cluster privato nelle due regioni e abilita Backup per GKE per il ripristino di emergenza.

Esegui il deployment di Kafka sul tuo cluster

In questa sezione, eseguirai il deployment di Kafka su GKE utilizzando un grafico Helm. L'operazione crea le seguenti risorse:

Gli StatefulSet Kafka e Zookeeper.
Implementazione di un esportatore Kafka. L'esportatore raccoglie le metriche di Kafka per l'utilizzo di Prometheus.
Un budget di interruzione dei pod (PDB) che limita il numero di pod offline durante un'interruzione volontaria.

Per utilizzare il grafico Helm per il deployment di Kafka:

Configurare l'accesso a Docker.

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

Completare Artifact Registry con le immagini di Kafka e 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

Configura l'accesso da riga di comando di kubectl al cluster primario.

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

Crea uno spazio dei nomi.

export NAMESPACE=kafka
kubectl create namespace $NAMESPACE

Installare Kafka utilizzando il grafico Helm versione 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"

L'output è simile al seguente:

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

Verifica che le repliche Kafka siano in esecuzione. L'operazione potrebbe richiedere alcuni minuti.

kubectl get all -n kafka

L'output è simile al seguente:

---
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

Crea dati di test

In questa sezione testerai l'applicazione Kafka e genererai messaggi.

Creare un pod client consumer per interagire con l'applicazione 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

Crea un argomento denominato topic1 con tre partizioni e un fattore di replica di tre.

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

Verifica che le partizioni dell'argomento siano replicate in tutti e tre i broker.

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

L'output è simile al seguente:

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

Nell'output di esempio, vedrai che topic1 ha tre partizioni, ognuna con un leader e un insieme di repliche diversi. Questo perché Kafka utilizza il partizionamento per distribuire i dati tra più broker, consentendo una maggiore scalabilità e tolleranza agli errori. Il fattore di replica pari a 3 garantisce che ogni partizione contenga tre repliche, in modo che i dati siano ancora disponibili anche in caso di errore di uno o due broker.

Esegui questo comando per generare collettivamente i numeri dei messaggi in 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=":"

Esegui questo comando per utilizzare topic1 da tutte le partizioni.

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

Digita CTRL+C per interrompere il processo del consumatore.

Benchmark Kafka

Per modellare con precisione un caso d'uso, puoi eseguire una simulazione del carico previsto sul cluster. Per verificare le prestazioni, utilizza gli strumenti inclusi nel pacchetto Kafka, ovvero gli script kafka-producer-perf-test.sh e kafka-consumer-perf-test.sh nella cartella bin.

Crea un argomento per il benchmarking.

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

Crea un carico sul 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

Il produttore genererà 10.000.000 di record su topic-benchmark. L'output è simile al seguente:

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.

Una volta inviati tutti i record, nell'output dovresti vedere metriche aggiuntive, simili alle seguenti:

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

Per uscire dallo smartwatch, digita CTRL + C.

Esci dalla shell del pod.
```
exit
```

Gestisci gli upgrade

Gli aggiornamenti della versione per Kafka e Kubernetes vengono rilasciati con un programma regolare. Segui le best practice operative per eseguire regolarmente l'upgrade del tuo ambiente software.

Pianifica gli upgrade dei file binari di Kafka

In questa sezione, aggiornerai l'immagine di Kafka utilizzando Helm e verificherai che i tuoi argomenti siano ancora disponibili.

Per eseguire l'upgrade dalla versione precedente di Kafka dal grafico Helm che hai utilizzato nel deployment di Kafka nel cluster, segui questi passaggi:

Compila Artifact Registry con l'immagine seguente:

../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

Esegui questi passaggi per eseguire il deployment di un grafico Helm con le immagini Kafka e Zoookeeper aggiornate. Per indicazioni specifiche per la versione, consulta le istruzioni di Kafka per gli upgrade della versione.
1. Aggiorna la versione della dipendenza Chart.yaml:
```
../scripts/chart.sh kafka 20.1.0
```
1. Esegui il deployment del grafico Helm con le nuove immagini di Kafka e Zookeeper, come mostrato nell'esempio seguente:
```
rm -rf Chart.lock charts && \
helm dependency update && \
helm -n kafka upgrade --install kafka ./ \
      --set global.imageRegistry="$REGION-docker.pkg.dev/$PROJECT_ID/main"
```
Guarda l'upgrade dei pod di Kafka:
```
kubectl get pod -l app.kubernetes.io/component=kafka -n kafka --watch
```
Per uscire dallo smartwatch, digita CTRL + C.

Connettiti al cluster Kafka utilizzando 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

Verifica di poter accedere ai messaggi di topic1.
```
kafka-console-consumer.sh \
  --topic topic1 \
  --from-beginning \
  --bootstrap-server kafka-headless.kafka.svc.cluster.local:9092
```
L'output dovrebbe mostrare i messaggi generati dal passaggio precedente. Digita CTRL+C per uscire dalla procedura.
Esci dal pod del client.
```
exit
```

Preparati per il ripristino di emergenza

Per garantire che i carichi di lavoro di produzione rimangano disponibili in caso di evento che interrompe il servizio, devi preparare un piano di ripristino di emergenza (RE). Per scoprire di più sulla pianificazione RE, consulta la Guida alla pianificazione del ripristino di emergenza.

Per eseguire il backup e il ripristino dei carichi di lavoro sui cluster GKE, puoi utilizzare Backup per GKE.

Scenario di esempio di backup e ripristino di Kafka

In questa sezione, eseguirai un backup del tuo cluster da gke-kafka-us-central1 e ripristinerai il backup in gke-kafka-us-west1. Eseguirai l'operazione di backup e ripristino nell'ambito dell'applicazione, utilizzando la risorsa personalizzata ProtectedApplication.

Il seguente diagramma illustra i componenti della soluzione di ripristino di emergenza e la loro correlazione tra loro.

Il diagramma mostra un esempio di soluzione di backup e ripristino per un cluster Kafka ad alta disponibilità. — **Figura 3**: esempio di soluzione di backup e ripristino per un cluster Kafka ad alta disponibilità.

Per prepararti a eseguire il backup e il ripristino del cluster Kafka, segui questi passaggi:

Configura le variabili di ambiente.

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

Verifica che il cluster sia in stato RUNNING.
```
gcloud container clusters describe $CLUSTER_NAME --region us-central1 --format='value(status)'
```
Nota: il cluster deve essere in stato RUNNING prima di poter creare un backup.

Crea un piano di backup.

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

Crea manualmente un backup. Sebbene i backup pianificati siano generalmente regolati dalla pianificazione cron nel piano di backup, l'esempio seguente mostra come avviare un'operazione di backup una tantum.
```
gcloud beta container backup-restore backups create $BACKUP_NAME \
    --project=$PROJECT_ID \
    --location=$DR_REGION \
    --backup-plan=$BACKUP_PLAN_NAME \
    --wait-for-completion
```

Crea un piano di ripristino.

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"

Ripristino manuale da un backup.

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

Osserva l'applicazione ripristinata che appare nel cluster di backup. Potrebbero essere necessari alcuni minuti perché tutti i pod siano in esecuzione e pronti.
```
gcloud container clusters get-credentials gke-kafka-us-west1 \
    --region us-west1
kubectl get pod -n kafka --watch
```
Digita CTRL+C per uscire dallo smartwatch quando tutti i pod sono in esecuzione.

Verifica che gli argomenti precedenti possano essere recuperati da un consumatore.

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;

L'output è simile al seguente:

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

Digita CTRL+C per uscire dalla procedura.

Esci dal pod.
```
exit
```

Simulare un'interruzione del servizio Kafka

In questa sezione simulerai un errore del nodo sostituendo un nodo Kubernetes che ospita il broker. Questa sezione riguarda solo gli annunci standard. Autopilot gestisce i nodi per te, quindi non è possibile simulare l'errore dei nodi.

Creare un pod client per la connessione all'applicazione Kafka.

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

Crea l'argomento topic-failover-test e produci traffico di prova.

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

Determina quale broker è leader per l'argomento topic-failover-test.
```
kafka-topics.sh --describe \
  --topic topic-failover-test \
  --bootstrap-server kafka-headless.kafka.svc.cluster.local:9092
```
L'output è simile al seguente:
```
Topic: topic-failover-test     Partition: 0    Leader: 1       Replicas: 1,0,2 Isr: 1,0,2
```
Nell'output precedente, Leader: 1 significa che il leader per topic-failover-test è il broker 1. Corrisponde al pod kafka-1.

Apri un nuovo terminale e connettiti allo stesso cluster.

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

Scopri su quale pod di nodo è in esecuzione kafka-1.

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

L'output è simile al seguente:

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>

Nell'output riportato sopra, vedi che il pod kafka-1 è in esecuzione sul nodo gke-gke-kafka-us-west1-pool-system-a0d0d395-nx72.

Svuota il nodo per rimuovere i pod.

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

Sostituisci NODE con il pod del nodo kafka-1 in esecuzione. In questo esempio, il nodo è gke-gke-kafka-us-west1-pool-system-a0d0d395-nx72.

L'output è simile al seguente:

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

Scopri su quale pod di nodo è in esecuzione kafka-1.

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

L'output dovrebbe essere simile al seguente:

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>

Dall'output riportato sopra puoi vedere che l'applicazione è in esecuzione su un nuovo nodo.

Nel terminale collegato al pod kafka-client, determina quale broker è leader per topic-failover-test.

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

L'output dovrebbe essere simile al seguente:

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

Nell'output di esempio, il leader è ancora 1 . Tuttavia, ora è in esecuzione su un nuovo nodo.

Test del fallimento del leader Kafka

In Cloud Shell, connettiti al client Kafka e usa describe per visualizzare la leader scelta per ogni partizione in topic1.

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

L'output è simile al seguente:

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

In Cloud Shell non collegato al client Kafka, elimina il broker leader kafka-0 per forzare una nuova elezione di leader. Devi eliminare l'indice mappato a una delle leader nell'output precedente.
```
kubectl delete pod -n kafka kafka-0 --force
```
L'output è simile al seguente:
```
pod "kafka-0" force deleted
```

Nella sessione di Cloud Shell collegata al client Kafka e utilizza describe per visualizzare il leader selezionato.

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

L'output è simile al seguente:

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

Nell'output, il nuovo leader per ogni partizione cambia, se è stato assegnato al leader che è stato interrotto (kafka-0). Ciò indica che il leader originale è stato sostituito quando il pod è stato eliminato e ricreato.

Esegui la pulizia

Per evitare che al tuo Account Google Cloud vengano addebitati costi relativi alle risorse utilizzate in questo tutorial, elimina il progetto che contiene le risorse oppure mantieni il progetto ed elimina le singole risorse.

Elimina il progetto

Il modo più semplice per evitare la fatturazione è eliminare il progetto che hai creato per il tutorial.

Attenzione: l'eliminazione di un progetto ha i seguenti effetti:

L'intero contenuto del progetto viene eliminato. Se hai utilizzato un progetto esistente per le attività in questo documento, quando lo elimini elimini anche qualsiasi altro lavoro svolto nel progetto.
Gli ID progetto personalizzati non sono più disponibili. Quando hai creato questo progetto, potresti aver creato un ID progetto personalizzato che vuoi utilizzare in futuro. Per conservare gli URL che utilizzano l'ID progetto, ad esempio un URL appspot.com, elimina le risorse selezionate all'interno del progetto anziché eliminare l'intero progetto.

Se intendi esplorare più architetture, tutorial o guide rapide, puoi riutilizzare i progetti ed evitare così di superare i limiti di quota.

Elimina un progetto Google Cloud:

gcloud projects delete PROJECT_ID

Passaggi successivi

Per un servizio di messaggistica completamente gestito e scalabile, vedi Eseguire la migrazione da Kafka a Pub/Sub.