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

Kafka è un sistema di messaggistica di pubblicazione/sottoscrizione distribuito e open source per la gestione di dati di streaming in tempo reale ad alto volume e velocità effettiva. Puoi utilizzare Kafka per creare pipeline di dati di streaming che spostano i dati in modo affidabile tra diversi sistemi e applicazioni per l'elaborazione e l'analisi.

Questo tutorial è destinato ad amministratori di piattaforme, architetti cloud e professionisti delle operazioni interessati a eseguire il deployment di cluster Kafka a disponibilità elevata su Google Kubernetes Engine (GKE).

Crea l'infrastruttura del cluster

In questa sezione eseguirai uno script Terraform per creare due cluster GKE regionali. Il cluster primario verrà implementato in us-central1.

Per creare il cluster:

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 eseguire il deployment dell'infrastruttura:

  • Crea un repository Artifact Registry per archiviare le immagini Docker.
  • Crea la rete VPC e la subnet per l'interfaccia di rete della VM.
  • Crea due cluster GKE.

Terraform crea un cluster privato nelle due regioni e attiva Backup per GKE per ilripristino di emergenzay.

Esegui il deployment di Kafka sul cluster

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

Per utilizzare il grafico Helm per eseguire il deployment di Kafka:

  1. Configura l'accesso a Docker.

    gcloud auth configure-docker us-docker.pkg.dev
    
  2. Popola 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
    
  3. Configura l'accesso alla riga di comando kubectl al cluster primario.

    gcloud container clusters get-credentials gke-kafka-us-central1 \
        --location=${REGION} \
        --project=${PROJECT_ID}
    
  4. Crea uno spazio dei nomi.

    export NAMESPACE=kafka
    kubectl create namespace $NAMESPACE
    
  5. Installa Kafka utilizzando la versione 20.0.6 del grafico Helm.

    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
    
  6. Verifica che le repliche Kafka siano in esecuzione (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
    

Creare dati di test

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

  1. Crea 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
    
  2. 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
    
  3. 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, nota 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 su più broker, consentendo maggiore scalabilità e tolleranza agli errori. Il fattore di replica 3 garantisce che ogni partizione abbia tre repliche, in modo che i dati siano comunque disponibili anche se uno o due broker non funzionano.

  4. Esegui questo comando per generare in blocco i numeri di messaggio 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=":"
    
  5. 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 la procedura consumer.

Benchmark Kafka

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

  1. 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
    
  2. Crea 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 il giorno 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 visualizzare altre metriche, simili a quelle riportate di seguito:

    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.

  3. Esci dalla shell del pod.

    exit
    

Gestire gli upgrade

Gli aggiornamenti delle versioni di Kafka e Kubernetes vengono rilasciati regolarmente. Segui le best practice operative per eseguire l'upgrade regolare del tuo ambiente software.

Pianifica gli upgrade dei binari Kafka

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

Per eseguire l'upgrade dalla versione precedente di Kafka dal grafico Helm utilizzato in Esegui il deployment di Kafka sul cluster, segui questi passaggi:

  1. Popola 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
    
  2. Segui questi passaggi per eseguire il deployment di un grafico Helm con le immagini di Kafka e Zookeeper aggiornate. Per indicazioni specifiche per la versione, consulta le istruzioni di Kafka per gli upgrade di 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 Kafka:

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

    Per uscire dallo smartwatch, digita CTRL + C.

  3. 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
    
  4. 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 nel passaggio precedente. Digita CTRL+C per uscire dalla procedura.

  5. Esci dal pod client.

    exit
    

Prepararsi per il ripristino di emergenza

Per assicurarti che i tuoi carichi di lavoro di produzione rimangano disponibili in caso di interruzione del servizio, devi preparare un piano di ripristino di emergenza. Per saperne di più sulla pianificazione RE, consulta la Guida alla pianificazione del disaster recovery.

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

Scenario di esempio di backup e ripristino di Kafka

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

Il seguente diagramma illustra i componenti della soluzione di ripristino di emergenza e il loro rapporto.

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

Per prepararti a eseguire il backup e il ripristino del cluster Kafka:

  1. 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
    
  2. Verifica che il cluster sia nello stato RUNNING.

    gcloud container clusters describe $CLUSTER_NAME --location us-central1 --format='value(status)'
    
  3. 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
    
  4. Crea manualmente un backup. Anche se i backup pianificati sono in genere 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
    
  5. 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"
    
  6. Ripristina manualmente 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
    
  7. Guarda l'applicazione ripristinata apparire nel cluster di backup. Potrebbero essere necessari alcuni minuti prima che tutti i pod siano in esecuzione e pronti.

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

    Digita CTRL+C per uscire dall'orologio quando tutti i pod sono in esecuzione.

  8. Verifica che gli argomenti precedenti possano essere recuperati da un consumer.

    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.

  9. 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 si applica solo a Standard. Autopilot gestisce i nodi per te, quindi non è possibile simulare l'errore del nodo.

  1. Crea un pod client per connetterti 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
    
  2. Crea l'argomento topic-failover-test e genera traffico di test.

    kafka-topics.sh \
      --create \
      --topic topic-failover-test \
      --partitions 1  \
      --replication-factor 3  \
      --bootstrap-server kafka-headless.kafka.svc.cluster.local:9092
    
  3. Determina quale broker è il 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 riportato sopra, Leader: 1 significa che il leader per topic-failover-test è il broker 1. Corrisponde al pod kafka-1.

  4. Apri un nuovo terminale e connettiti allo stesso cluster.

    gcloud container clusters get-credentials gke-kafka-us-west1 --location us-west1 --project PROJECT_ID
    
  5. Trova il nodo su cui è in esecuzione il pod 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.

  6. Svuota il nodo per espellere i pod.

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

    Sostituisci NODE con il nodo su cui è in esecuzione il pod kafka-1. 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
    
  7. Trova il nodo su cui è in esecuzione il pod 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 precedente, vedi che l'applicazione è in esecuzione su un nuovo nodo.

  8. Nel terminale connesso al pod kafka-client, determina quale broker è il 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 viene eseguito su un nuovo nodo.

Test del failover del leader Kafka

  1. In Cloud Shell, connettiti al client Kafka e utilizza describe per visualizzare il leader eletto 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
    
  2. In Cloud Shell non connesso al client Kafka, elimina il broker leader kafka-0 per forzare una nuova elezione del leader. Devi eliminare l'indice che corrisponde a uno dei leader nell'output precedente.

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

    L'output è simile al seguente:

    pod "kafka-0" force deleted
    
  3. In Cloud Shell connessa al client Kafka, utilizza describe per visualizzare il leader eletto.

    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.