Implantar um cluster Kafka altamente disponível no GKE

Autopilot Padrão

O Kafka é um sistema de mensagens de publicação/assinatura de código aberto e distribuído para lidar com dados de streaming de alto volume, alta capacidade e streaming em tempo real. É possível usar o Kafka para criar pipelines de dados de streaming que movam dados de maneira confiável em diferentes sistemas e aplicativos para processamento e análise.

Este tutorial é destinado a administradores de plataformas, arquitetos de nuvem e profissionais de operações interessados em implantar clusters Kafka altamente disponíveis no Google Kubernetes Engine (GKE).

Objetivos

Neste tutorial, você aprenderá a:

Use o Terraform para criar um cluster regional do GKE.
Implantar um cluster do Kafka altamente disponível.
Faça upgrade dos binários do Kafka.
Faça backup e restaure o cluster do Kafka.
Simule a interrupção do nó do GKE e o failover do agente do Kafka.

Arquitetura

Nesta seção, descrevemos a arquitetura da solução que você criará neste tutorial.

Um cluster do Kafka é um grupo de um ou mais servidores (chamados agentes) que trabalham juntos para lidar com os fluxos de dados de entrada e enviar mensagens de assinatura para clientes do Kafka (chamados consumidores).

Cada partição de dados em um cluster do Kafka tem um agente líder e pode ter um ou mais agentes seguidores. O agente líder processa todas as leituras e gravações na partição. Cada agente seguidor replica passivamente o agente líder.

Em uma configuração típica do Kafka, você também usa um serviço de código aberto chamado ZooKeeper para coordenar os clusters do Kafka. Esse serviço ajuda a selecionar um líder entre os agentes e acionar o failover em caso de falhas.

Neste tutorial, você implanta os clusters do Kafka no GKE configurando os agentes do Kafka e o serviço Zookeeper como StatefulSets individuais. Para provisionar clusters do Kafka altamente disponíveis e se preparar para recuperação de desastres, configure seus StatefulSets Kafka e Zookeeper para usar pools de nós e zonas separados.

O diagrama a seguir mostra como o StatefulSet do Kafka é executado em vários nós e zonas no cluster do GKE.

O diagrama mostra um exemplo de arquitetura de um StatefulSet do Kafka no GKE implantado em várias zonas. — **Figura 1**: como implantar o StatefulSet do Kafka em nós do GKE em três zonas diferentes.

O diagrama a seguir mostra como o StatefulSet do Zookeeper é executado em vários nós e zonas no cluster do GKE.

O diagrama mostra um exemplo de arquitetura de um StatefulSet do Zookeeper para o GKE implantado em várias zonas. — **Figura 2**: como implantar o Zookeeper do Kafka em nós do GKE em três zonas diferentes.

Provisionamento de nós e programação de pods

Se você estiver usando clusters do Autopilot, o Autopilot vai lidar com o provisionamento de nós e programar os pods para suas cargas de trabalho. Você vai usar a antiafinidade de pods para garantir que não haja dois pods do mesmo StatefulSet programados no mesmo nó e na mesma zona.

Se você estiver usando clusters padrão, precisará configurar a tolerância e a afinidade de nós do pod. Para saber mais, consulte Isolar suas cargas de trabalho em pools de nós dedicados.

Custos

Neste documento, você usará os seguintes componentes faturáveis do Google Cloud:

Para gerar uma estimativa de custo baseada na projeção de uso deste tutorial, use a calculadora de preços. Novos usuários do Google Cloud podem estar qualificados para uma avaliação gratuita.

Ao concluir as tarefas descritas neste documento, é possível evitar o faturamento contínuo excluindo os recursos criados. Saiba mais em Limpeza.

Antes de começar

Configurar o projeto

Faça login na sua conta do Google Cloud. Se você começou a usar o Google Cloud agora, crie uma conta para avaliar o desempenho de nossos produtos em situações reais. Clientes novos também recebem US$ 300 em créditos para executar, testar e implantar cargas de trabalho.

No console do Google Cloud, na página do seletor de projetos, clique em Criar projeto para começar a criar um novo projeto do Google Cloud.

Acessar o seletor de projetos

Verifique se a cobrança está ativada para o seu projeto do Google Cloud.

Ative as APIs Google Kubernetes Engine, Backup for GKE, Artifact Registry, Compute Engine, and IAM.

Ative as APIs

No console do Google Cloud, na página do seletor de projetos, clique em Criar projeto para começar a criar um novo projeto do Google Cloud.

Acessar o seletor de projetos

Verifique se a cobrança está ativada para o seu projeto do Google Cloud.

Ative as APIs Google Kubernetes Engine, Backup for GKE, Artifact Registry, Compute Engine, and IAM.

Ative as APIs

Configurar papéis

Atribua os papéis à sua Conta do Google. Execute uma vez o seguinte comando para cada um dos seguintes papéis do 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
```
- Substitua PROJECT_ID pela ID do seu projeto.
- Substitua EMAIL_ADDRESS pelo seu endereço de e-mail.
- Substitua ROLE por cada papel individual.

Configure seu ambiente

Neste tutorial, você usará o Cloud Shell para gerenciar recursos hospedados no Google Cloud. O Cloud Shell vem pré-instalado com o software necessário para este tutorial, incluindo o Docker, o kubectl, a CLI gcloud, Helm e Terraform.

Para configurar o ambiente com o Cloud Shell, siga estas etapas:

Inicie uma sessão do Cloud Shell no Console do Google Cloud clicando em Ativar o Cloud Shell no Console do Google Cloud. Isso inicia uma sessão no painel inferior do Cloud Console.
Defina variáveis de ambiente.
```
export PROJECT_ID=PROJECT_ID
export REGION=us-central1
```
Substitua os seguintes valores:
- PROJECT_ID: é o ID do projeto do Google Cloud.
Defina as variáveis de ambiente padrão.
```
gcloud config set project PROJECT_ID
```

Clone o repositório do código.

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

Mude para o diretório de trabalho.

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

Criar a infraestrutura do cluster

Nesta seção, você vai executar um script do Terraform para criar dois clusters regionais do GKE. O cluster primário será implantado em us-central1.

Para criar o cluster, siga estas etapas:

Piloto automático

No Cloud Shell, execute os seguintes comandos:

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

Quando solicitado, digite yes.

Standard

No Cloud Shell, execute os seguintes comandos:

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

Quando solicitado, digite yes.

Os arquivos de configuração do Terraform criam os seguintes recursos para implantar a infraestrutura:

Crie um repositório do Artifact Registry para armazenar as imagens do Docker.
Crie a rede VPC e a sub-rede da interface de rede da VM.
Crie dois clusters do GKE.

O Terraform cria um cluster particular nas duas regiões e ativa o Backup para GKE para a recuperação de desastres.

Implantar Kafka no cluster

Nesta seção, você implantará o Kafka no GKE usando um gráfico do Helm. A operação cria os seguintes recursos:

Os StatefulSets Kafka e Zookeeper.
Uma implantação de exportador Kafka. O exportador coleta métricas do Kafka para o consumo do Prometheus.
Um orçamento de interrupção do pod (PDB) que limita o número de pods off-line durante uma interrupção voluntária.

Para usar o gráfico do Helm para implantar o Kafka, siga estas etapas:

Configurar o acesso ao Docker

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

Preencha o Artifact Registry com as imagens do Kafka e do 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

Configure o acesso da linha de comando kubectl ao cluster principal.

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

Crie um namespace.

export NAMESPACE=kafka
kubectl create namespace $NAMESPACE

Instale o Kafka usando a versão 20.0.6 do gráfico do 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"

O resultado será assim:

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

Verifique se as réplicas do Kafka estão em execução. Isso pode levar alguns minutos.

kubectl get all -n kafka

O resultado será assim:

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

Crie dados de teste

Nesta seção, você vai testar o aplicativo Kafka e gerar mensagens.

Criar um pod cliente do cliente para interagir com o aplicativo 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

Crie um tópico chamado topic1 com três partições e um fator de replicação de três.

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

Verifique se as partições de tópico estão replicadas nos três agentes.

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

O resultado será assim:

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

No exemplo de saída, observe que topic1 tem três partições, cada uma com um líder e um conjunto de réplicas diferentes. Isso ocorre porque o Kafka usa o particionamento para distribuir os dados entre vários agentes, permitindo maior escalonabilidade e tolerância a falhas. O fator de replicação de três garante que cada partição tenha três réplicas, para que os dados ainda estejam disponíveis, mesmo que um ou dois agentes falhem.

Execute o comando a seguir para gerar números de mensagem em massa em 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=":"

Execute o comando a seguir para consumir topic1 de todas as partições.

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

Digite CTRL+C para interromper o processo do consumidor.

Comparativo de mercado do Kafka

Para modelar com precisão um caso de uso, execute uma simulação da carga esperada no cluster. Para testar a performance, use as ferramentas incluídas no pacote Kafka, ou seja, os scripts kafka-producer-perf-test.sh e kafka-consumer-perf-test.sh na pasta bin.

Crie um tópico para comparação.

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

Criar carga no cluster do 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

O produtor vai gerar 10.000.000 registros em topic-benchmark. A resposta será parecida com esta:

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.

Depois que todos os registros forem enviados, serão exibidas outras métricas na saída, como a seguir:

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

Para sair da exibição, digite CTRL + C.

Saia do shell do pod.
```
exit
```

Gerenciar upgrades

As atualizações de versão do Kafka e do Kubernetes são lançadas regularmente. Siga as práticas recomendadas operacionais para atualizar o ambiente de software regularmente.

Planejar upgrades binários do Kafka

Nesta seção, você atualizará a imagem do Kafka usando o Helm e verificará se os temas ainda estão disponíveis.

Para fazer upgrade da versão anterior do Kafka do gráfico Helm que você usou em Implantar o Kafka no seu cluster, siga estas etapas:

Preencha o Artifact Registry com a seguinte imagem:

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

Siga estas etapas para implantar um gráfico do Helm com as imagens atualizadas do Kafka e do Zookeeper. Para orientações específicas da versão, consulte as instruções do Kafka para upgrades de versão.
1. Atualize a versão da dependência Chart.yaml:
```
../scripts/chart.sh kafka 20.1.0
```
1. Implante o gráfico do Helm com as novas imagens do Kafka e do Zookeeper, conforme mostrado no seguinte exemplo:
```
rm -rf Chart.lock charts && \
helm dependency update && \
helm -n kafka upgrade --install kafka ./ \
      --set global.imageRegistry="$REGION-docker.pkg.dev/$PROJECT_ID/main"
```
Acompanhe o upgrade dos pods do Kafka:
```
kubectl get pod -l app.kubernetes.io/component=kafka -n kafka --watch
```
Para sair da exibição, digite CTRL + C.

Conectar-se ao cluster do Kafka usando um pod cliente.

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

Verifique se você pode acessar mensagens de topic1.
```
kafka-console-consumer.sh \
  --topic topic1 \
  --from-beginning \
  --bootstrap-server kafka-headless.kafka.svc.cluster.local:9092
```
A saída vai mostrar as mensagens geradas na etapa anterior. Digite CTRL+C para sair do processo.
Saia do pod cliente.
```
exit
```

Prepare-se para a recuperação de desastres

Para garantir que suas cargas de trabalho de produção permaneçam disponíveis no caso de um evento de interrupção de serviço, prepare um plano de recuperação de desastres (DR). Para saber mais sobre o planejamento de DR, consulte o Guia de planejamento de recuperação de desastres.

Para fazer backup e restaurar suas cargas de trabalho em clusters do GKE, use o Backup para o GKE.

Exemplo de cenário de backup e restauração do Kafka

Nesta seção, você fará um backup do cluster em gke-kafka-us-central1 e o restaurará em gke-kafka-us-west1. Você realizará a operação de backup e restauração no escopo do aplicativo usando o recurso personalizado ProtectedApplication.

O diagrama a seguir ilustra os componentes da solução de recuperação de desastres e como eles se relacionam.

O diagrama mostra um exemplo de solução de backup e recuperação de um cluster do Kafka altamente disponível. — **Figura 3**: exemplo de solução de backup e recuperação para um cluster do Kafka altamente disponível.

Para se preparar para o backup e a restauração do cluster do Kafka, siga estas etapas:

Configurar as variáveis de 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

Verifique se o cluster está em um estado RUNNING.
```
gcloud container clusters describe $CLUSTER_NAME --region us-central1 --format='value(status)'
```
Observação: o cluster precisa estar no estado RUNNING para criar um backup.

Crie um plano de 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

Crie um backup manualmente. Embora os backups programados geralmente sejam regidos pelo cron-schedule no plano de backup, o exemplo a seguir mostra como iniciar uma operação de backup única.
```
gcloud beta container backup-restore backups create $BACKUP_NAME \
    --project=$PROJECT_ID \
    --location=$DR_REGION \
    --backup-plan=$BACKUP_PLAN_NAME \
    --wait-for-completion
```

Crie um plano de restauração.

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"

Restaure manualmente de um 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

Veja o aplicativo restaurado aparecer no cluster de backup. Pode levar alguns minutos para que todos os pods estejam em execução e prontos.
```
gcloud container clusters get-credentials gke-kafka-us-west1 \
    --region us-west1
kubectl get pod -n kafka --watch
```
Digite CTRL+C para sair da exibição quando todos os pods estiverem em execução.

Valide se os temas anteriores podem ser buscados por um consumidor.

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;

O resultado será assim:

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

Digite CTRL+C para sair do processo.

Saia do pod.
```
exit
```

Simular uma interrupção do serviço Kafka

Nesta seção, você simulará uma falha de nó substituindo um nó do Kubernetes que hospeda o agente. Esta seção se aplica somente ao padrão. O Autopilot gerencia os nós para você, então a falha não pode ser simulada.

Criar um pod de cliente para se conectar ao aplicativo Kafka.

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

Crie o tópico topic-failover-test e gere tráfego de teste.

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

Determine qual agente é o líder do tópico topic-failover-test.

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

O resultado será assim:

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

Na saída acima, Leader: 1 significa que o líder de topic-failover-test é agente 1. Corresponde ao pod kafka-1.

Abra um novo terminal e conecte-se ao mesmo cluster.

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

Descubra em qual nó o pod kafka-1 está sendo executado.

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

O resultado será assim:

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>

Na saída acima, você vê que o pod kafka-1 está em execução no nó gke-gke-kafka-us-west1-pool-system-a0d0d395-nx72.

Drene o nó para remover os pods.

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

Substitua NODE pelo pod do nó kafka-1 em execução. Neste exemplo, o nó é gke-gke-kafka-us-west1-pool-system-a0d0d395-nx72.

O resultado será assim:

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

Descubra em qual nó o pod kafka-1 está sendo executado.

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

A resposta será parecida com esta:

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>

Na saída acima, você vê que o aplicativo está em execução em um novo nó.

No terminal conectado ao pod kafka-client, determine qual agente é líder para topic-failover-test.

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

A resposta será parecida com esta:

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

No exemplo de saída, o líder ainda é 1. No entanto, ele está em execução em um novo nó.

Teste de falha do líder do Kafka

No Cloud Shell, conecte-se ao cliente Kafka e use describe para ver o líder escolhido para cada partição em topic1.

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

O resultado será assim:

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

No Cloud Shell, não conectado ao cliente Kafka, exclua o agente líder kafka-0 para forçar uma nova eleição de líder. Exclua o índice que mapeia para um dos líderes na saída anterior.
```
kubectl delete pod -n kafka kafka-0 --force
```
O resultado será assim:
```
pod "kafka-0" force deleted
```

No Cloud Shell conectado ao cliente Kafka e use describe para ver o líder escolhido.

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

O resultado será assim:

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

Na saída, o novo líder para cada partição será alterado, se ele tiver sido atribuído ao líder que foi interrompido (kafka-0). Isso indica que o líder original foi substituído quando o pod foi excluído e recriado.

Limpar

Para evitar cobranças na sua conta do Google Cloud pelos recursos usados no tutorial, exclua o projeto que os contém ou mantenha o projeto e exclua os recursos individuais.

Excluir o projeto

A maneira mais fácil de evitar o faturamento é excluir o projeto criado para o tutorial.

Cuidado: excluir um projeto tem os seguintes efeitos:

Tudo no projeto é excluído. Se você tiver usado um projeto existente para as tarefas neste documento, a exclusão dele incluirá a exclusão de quaisquer outros trabalhos feitos no projeto.
Os IDs do projeto personalizados são perdidos. Ao criar o projeto, você pode ter criado um código do projeto personalizado para ser usado no futuro. Para preservar os URLs que usam o ID do projeto, como um URL appspot.com, exclua recursos específicos do projeto, em vez de excluir o projeto inteiro.

Se você planeja passar por várias arquiteturas, tutoriais ou guias de início rápido, a reutilização de projetos pode evitar que você exceda os limites da cota do projeto.

Exclua um projeto do Google Cloud:

gcloud projects delete PROJECT_ID

A seguir

Para um serviço de mensagens totalmente gerenciado e escalonável, consulte Migrar do Kafka para o Pub/Sub.