Introdução
Um grupo pode dividir as partições de um tópico entre vários consumidores. Cada partição tem um responsável ativo dentro de um grupo estável, enquanto outro grupo pode ler os mesmos registros retidos de forma independente. Isso permite compartilhar trabalho sem fazer todas as aplicações disputarem uma única posição global de leitura.
Você iniciará dois clientes de console de longa duração em um grupo, inspecionará suas atribuições reais antes de publicar e comparará as observações de um grupo separado. Após a saída de um membro, aguardará até que o membro restante assuma as duas partições antes de enviar outro pequeno lote. O publicador preparado direciona explicitamente os dados para ambas as partições para que cada uma tenha trabalho durante o experimento; sua entrega real ao broker é verificada.
Este laboratório independente fornece um tópico com duas partições, arquivos públicos de eventos e ferramentas de espera somente de leitura com prazo limitado. Conhecimentos básicos de terminal bastam; as ferramentas dos dados de exemplo não exigem editar Python. Os clientes de console executam em segundo plano até você interrompê-los nas etapas finais. Mantenha o mesmo shell do Terminal para que suas variáveis de processo continuem disponíveis.
Inicie dois membros e aguarde uma atribuição estável
Nesta etapa, você observará a atribuição antes de dar trabalho ao grupo. O tópico preparado tem duas partições e nenhum registro. Inspecione seus metadados:
cd /home/labex/project
kafka-topics.sh --bootstrap-server localhost:9092 --describe --topic orders-shared
Os clientes a seguir usam o mesmo grupo, orders-team. --command-property passa cada valor de configuração ao cliente consumidor. client.id os identifica como reader-a e reader-b nos relatórios. As configurações conhecidas de confirmação automática permitem que o progresso da leitura se estabilize enquanto eles continuam consultando novos registros.
O & ao final inicia um cliente em segundo plano no shell para que você possa executar comandos de inspeção no mesmo terminal. $! é o ID do processo recém-iniciado em segundo plano; cada atribuição logo em seguida salva esse ID para o encerramento específico mais adiante.
O shell mantém dois fluxos de saída: saída padrão para registros e saída de erro padrão para diagnósticos. > grava o primeiro fluxo no arquivo JSONL; 2> grava o fluxo de erros no log. A atribuição consumer_a_pid=$! armazena uma variável do shell sem espaços ao redor de =; "$consumer_a_pid" recupera depois esse ID de processo específico. Registros comuns e diagnósticos vão para arquivos separados para não interromper seu prompt. Esses clientes continuam executando até os comandos kill específicos perto do fim:
kafka-console-consumer.sh --bootstrap-server localhost:9092 --topic orders-shared --group orders-team --from-beginning --command-property client.id=reader-a --command-property enable.auto.commit=true --command-property auto.commit.interval.ms=1000 --command-property max.poll.records=1 > /home/labex/project/output/reader-a.jsonl 2> /home/labex/project/output/reader-a.log &
consumer_a_pid=$!
kafka-console-consumer.sh --bootstrap-server localhost:9092 --topic orders-shared --group orders-team --from-beginning --command-property client.id=reader-b --command-property enable.auto.commit=true --command-property auto.commit.interval.ms=1000 --command-property max.poll.records=1 > /home/labex/project/output/reader-b.jsonl 2> /home/labex/project/output/reader-b.log &
consumer_b_pid=$!
Aguarde a atribuição real dos dois membros e salve seu relatório:
/home/labex/project/tools/wait-assignment 2 > /home/labex/project/output/two-members.txt
cat /home/labex/project/output/two-members.txt
O utilitário somente de leitura consulta repetidamente o relatório de membros do grupo até que os dois clientes identificados tenham atribuições não vazias e sem sobreposição, cuja união seja as partições 0 e 1. Ele tem um prazo e não fabrica atribuições nem altera o progresso.
No relatório, CLIENT-ID identifica reader-a ou reader-b, e CURRENT-ASSIGNMENT lista a responsabilidade atual real. Por exemplo, orders-shared:0 significa a partição 0 desse tópico. Uma atribuição de destino ainda pode estar pendente, portanto inspecione a coluna atual. Espere que cada membro assuma uma partição, sem exigir que reader-a receba um número específico. Um rebalanceamento redistribui a responsabilidade pelas partições quando membros entram ou saem do grupo. O grupo precisa de tempo para chegar à nova atribuição; o início de um processo não significa que a responsabilidade já tenha se estabilizado. Uma atribuição estável é a evidência de compartilhamento de trabalho; dois IDs de processo sozinhos não comprovam quais partições eles assumiram.
Publique trabalho e compare um grupo independente
Nesta etapa, você fornecerá registros às duas partições atribuídas e observará um segundo grupo ler o mesmo histórico. Inspecione os lotes públicos. O lote initial contém de share1 a share6; followup contém share7 e share8:
cat /home/labex/project/data/group-batches.json
O publicador fornecido envia os dados de exemplo selecionados ao broker real, aguarda os callbacks de entrega e falha se a entrega ficar incompleta. Publique os seis eventos iniciais apenas após a atribuição estável:
/home/labex/project/tools/publish-shared-batch initial
/home/labex/project/tools/wait-caught-up orders-team 3 > /home/labex/project/output/team-initial-position.txt
cat /home/labex/project/output/team-initial-position.txt
O utilitário somente de leitura que aguarda a leitura completa espera até que ambas as partições tenham próximo offset confirmado 3, offset final 3 e lag 0. Cada partição recebeu três registros. Inspecione o que os dois clientes observaram:
cat /home/labex/project/output/reader-a.jsonl
cat /home/labex/project/output/reader-b.jsonl
Juntos, seus arquivos devem incluir os seis IDs iniciais. O relatório de atribuição ativa é a evidência de responsabilidade estável; um arquivo também pode conter observações anteriores a uma mudança de atribuição.
Agora inicie um grupo separado com limites, orders-observer, desde o início dos registros retidos:
kafka-console-consumer.sh --bootstrap-server localhost:9092 --topic orders-shared --group orders-observer --from-beginning --max-messages 6 --timeout-ms 30000 --command-property enable.auto.commit=true --command-property auto.commit.interval.ms=1000 --command-property max.poll.records=1 > /home/labex/project/output/observer-initial.jsonl
cat /home/labex/project/output/observer-initial.jsonl
/home/labex/project/tools/wait-caught-up orders-observer 3 > /home/labex/project/output/observer-initial-position.txt
cat /home/labex/project/output/observer-initial-position.txt
Ele deve receber os seis registros de forma independente. Ele não retira partições de orders-team porque pertence a outro grupo. Ambos os grupos podem ter lido tudo ao mesmo tempo, mantendo posições confirmadas separadas.
Interrompa um membro e observe a atribuição restante
Nesta etapa, você deixará reader-a em execução e interromperá apenas reader-b. kill envia um sinal de término ao processo salvo; ele não interrompe o broker. wait permite que o shell conclua a coleta do status desse processo filho. Um processo que recebeu um sinal pode informar um status de saída diferente de zero, portanto o || true ao final evita que este comando de encerramento intencional seja tratado como falha da aplicação:
kill "$consumer_b_pid"
wait "$consumer_b_pid" || true
/home/labex/project/tools/wait-assignment 1 > /home/labex/project/output/one-member.txt
cat /home/labex/project/output/one-member.txt
Espere que reader-a assuma ambas as partições depois que o grupo se estabilizar. A ferramenta de espera verifica o relatório real antes da próxima publicação. Com duas partições, acrescentar mais de dois membros deixaria alguns sem partição em uma atribuição estável; este exercício não exige um terceiro cliente.
Publique um evento adicional em cada partição e aguarde o membro restante concluir a leitura:
/home/labex/project/tools/publish-shared-batch followup
/home/labex/project/tools/wait-caught-up orders-team 4 > /home/labex/project/output/team-followup-position.txt
cat /home/labex/project/output/team-followup-position.txt
cat /home/labex/project/output/reader-a.jsonl
reader-a agora deve incluir tanto share7 quanto share8. reader-b foi interrompido, portanto não pode processar esses novos registros. O grupo observador ainda tem sua posição confirmada anterior; retome-o para os dois novos eventos:
kafka-console-consumer.sh --bootstrap-server localhost:9092 --topic orders-shared --group orders-observer --from-beginning --max-messages 2 --timeout-ms 30000 --command-property enable.auto.commit=true --command-property auto.commit.interval.ms=1000 --command-property max.poll.records=1 > /home/labex/project/output/observer-followup.jsonl
cat /home/labex/project/output/observer-followup.jsonl
/home/labex/project/tools/wait-caught-up orders-observer 4 > /home/labex/project/output/observer-followup-position.txt
cat /home/labex/project/output/observer-followup-position.txt
Espere share7 e share8 e o próximo offset 4 de cada grupo em ambas as partições. A opção de início não substitui o offset válido 3 salvo pelo grupo observador.
Um rebalanceamento pode causar entrega repetida se um membro que está saindo tiver tratado registros antes da confirmação de seu progresso. Por isso, as observações de eventos podem incluir repetições; uma atribuição estável ainda dá à partição um único responsável por vez. Este laboratório verifica atribuições reais e o conteúdo completo dos eventos sem afirmar uma entrega exatamente uma vez de ponta a ponta.
Encerre o cliente restante e preserve as evidências
Nesta etapa, você concluirá o experimento de longa duração interrompendo o único cliente restante. Suas posições finais já foram observadas no fim, portanto não há novo trabalho pendente. Use o próprio PID salvo dele:
kill "$consumer_a_pid"
wait "$consumer_a_pid" || true
/home/labex/project/tools/wait-assignment 0 > /home/labex/project/output/no-members.txt
cat /home/labex/project/output/no-members.txt
Agora o grupo não deve ter membros ativos. Os offsets salvos e os eventos no broker permanecem; o tempo de vida do cliente e o progresso armazenado são separados. Os clientes observadores com limites já encerraram automaticamente.
Inspecione os relatórios de membros preservados de antes e depois da recuperação das atribuições:
cat /home/labex/project/output/two-members.txt /home/labex/project/output/one-member.txt
O primeiro relatório demonstra dois responsáveis sem sobreposição e o segundo, um responsável pelas duas partições. São relatórios reais com identificadores de membros específicos da execução, não atribuições fixas para memorizar. Os arquivos dos leitores e dos observadores preservam evidências de eventos para os dois grupos. Você encerrou apenas seus dois clientes de console, preservando o tópico e as posições confirmadas para revisão.
Resumo
Você observou dois consumidores compartilharem atribuições reais de partições, um grupo separado ler os mesmos registros de forma independente e um membro restante assumir ambas as partições antes da chegada de novo trabalho. Verificações de estado com prazo limitado substituíram estimativas de tempo, e o encerramento específico dos clientes preservou o progresso e os eventos. A seguir, você substituirá a ferramenta de publicação preparada por um produtor Python cuja serialização e cujo tratamento de entrega poderá ler e modificar.
