Introdução
O Kafka mantém registros independentemente de quem os leu. Um grupo de consumidores também precisa lembrar seu progresso para que um processo posterior possa continuar em vez de recomeçar. Um offset confirmado é a próxima posição que esse grupo deve ler em uma partição, não o número do último registro.
Você lerá dois de quatro registros preparados, inspecionará o progresso confirmado e o lag, retomará a leitura com o mesmo grupo e comparará um novo grupo. Depois, visualizará e executará uma redefinição enquanto o grupo original estiver inativo e lerá novamente os registros retidos. O tópico tem uma partição para facilitar o raciocínio sobre as posições.
Este laboratório independente fornece o broker, quatro registros e um utilitário que consulta repetidamente o estado real do coordenador até que apareça o offset esperado. Conhecimentos básicos de terminal bastam; não é necessário escrever código Python nem configurar serviços. O utilitário não altera offsets nem fabrica um relatório. Use o Terminal.
Leia dois registros e inspecione a próxima posição confirmada
Nesta etapa, você relacionará uma leitura com limites à posição real salva pelo grupo. Inspecione os quatro valores preparados:
cd /home/labex/project
cat /home/labex/project/data/progress-events.txt
Seus offsets neste novo tópico de uma partição vão de 0 a 3. Um novo grupo chamado orders-review não tem posição salva, portanto --from-beginning seleciona o início dos registros retidos para esta leitura inicial.
--command-property passa um valor de configuração ao cliente consumidor, em vez de alterar o formato de exibição dos eventos. O cliente de console confirma automaticamente o progresso neste exercício. max.poll.records=1 limita cada consulta a um registro retornado, para que encerrar após dois registros exibidos não avance sua posição por um lote retornado maior. O intervalo define o comportamento de confirmação periódica; ao fechar, o cliente também conclui seu tratamento normal de progresso. Você inspecionará o resultado real em vez de presumir um atraso específico:
kafka-console-consumer.sh --bootstrap-server localhost:9092 --topic orders-progress --group orders-review --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/first-two.txt
cat /home/labex/project/output/first-two.txt
Espere os dois eventos created; em seguida, o cliente encerra. Execute o utilitário fornecido de consulta com tempo limitado, passando o nome do grupo e o próximo offset esperado:
/home/labex/project/tools/wait-group-offset orders-review 2 > /home/labex/project/output/after-first.txt
cat /home/labex/project/output/after-first.txt
O utilitário consulta repetidamente a descrição do grupo real até que seu offset confirmado seja 2, ou falha após o prazo. Ele não confirma nem redefine nada. Localize CURRENT-OFFSET 2, LOG-END-OFFSET 4 e LAG 2 para a partição 0. Uma mensagem informando que o grupo não tem membros ativos é normal depois que o cliente com limites encerra; seu progresso salvo continua existindo.
O offset 2 significa que a próxima leitura começa no terceiro registro armazenado. O offset final 4 é a próxima posição após os quatro registros existentes; o lag 4 − 2 = 2 mede os registros restantes neste exemplo simples e estabilizado. Os dois primeiros registros não foram excluídos.
Retome o mesmo grupo e compare um novo grupo
Nesta etapa, você preverá quais registros um grupo recebe a partir de sua posição salva. Execute o mesmo grupo novamente, mantendo deliberadamente a opção de início:
kafka-console-consumer.sh --bootstrap-server localhost:9092 --topic orders-progress --group orders-review --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/resumed-two.txt
cat /home/labex/project/output/resumed-two.txt
Espere os dois eventos paid, e não os eventos created anteriores. --from-beginning fornece uma política inicial quando não existe um offset confirmado válido; ela não substitui o offset válido 2 salvo por este grupo. Confirme a conclusão pelo estado real do coordenador:
/home/labex/project/tools/wait-group-offset orders-review 4 > /home/labex/project/output/after-resume.txt
cat /home/labex/project/output/after-resume.txt
Espere offset confirmado 4, offset final 4 e lag 0. Agora preveja o comportamento de outro grupo chamado orders-independent. Ele não tem posição salva e é independente de orders-review, portanto uma leitura desde o início deve retornar os quatro registros retidos:
kafka-console-consumer.sh --bootstrap-server localhost:9092 --topic orders-progress --group orders-independent --from-beginning --max-messages 4 --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/new-group-all.txt
cat /home/labex/project/output/new-group-all.txt
/home/labex/project/tools/wait-group-offset orders-independent 4 > /home/labex/project/output/new-group-position.txt
cat /home/labex/project/output/new-group-position.txt
Espere os quatro valores e o offset confirmado 4 do próprio novo grupo. Ambos os grupos podem chegar ao fim seguindo históricos de leitura diferentes. O progresso de um grupo não moveu a posição inicial do outro, e o consumo de nenhum deles removeu os registros do tópico.
A seguir, você alterará deliberadamente a posição salva do grupo original, em vez de tratar a opção de início como um comando para voltar a leitura.
Visualize uma redefinição de grupo inativo e releia o histórico
Nesta etapa, você redefinirá apenas orders-review. Seu consumidor com limites já encerrou, portanto o grupo não tem membro ativo. Interrompa qualquer membro desse mesmo grupo que tenha ficado em execução por engano antes de redefinir; um consumidor ativo poderia continuar alterando o progresso.
--reset-offsets seleciona a operação de gerenciamento de progresso, --topic limita seu alcance e --to-earliest seleciona o início dos registros retidos. Primeiro, use --dry-run para visualizar o novo offset proposto sem aplicá-lo:
kafka-consumer-groups.sh --bootstrap-server localhost:9092 --group orders-review --reset-offsets --to-earliest --topic orders-progress --dry-run > /home/labex/project/output/reset-preview.txt
cat /home/labex/project/output/reset-preview.txt
Espere um NEW-OFFSET proposto de 0. Confirme que a visualização não alterou a posição salva atual:
/home/labex/project/tools/wait-group-offset orders-review 4 > /home/labex/project/output/after-preview.txt
cat /home/labex/project/output/after-preview.txt
Agora aplique essa redefinição restrita com --execute:
kafka-consumer-groups.sh --bootstrap-server localhost:9092 --group orders-review --reset-offsets --to-earliest --topic orders-progress --execute
/home/labex/project/tools/wait-group-offset orders-review 0 > /home/labex/project/output/after-reset.txt
cat /home/labex/project/output/after-reset.txt
O próximo offset confirmado real agora deve ser 0, com offset final 4 e lag 4. Os registros permaneceram no lugar; apenas a posição salva deste grupo mudou. Leia os quatro novamente usando o início agora confirmado e válido, sem depender de uma opção de política inicial:
kafka-console-consumer.sh --bootstrap-server localhost:9092 --topic orders-progress --group orders-review --max-messages 4 --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/replayed-all.txt
cat /home/labex/project/output/replayed-all.txt
/home/labex/project/tools/wait-group-offset orders-review 4 > /home/labex/project/output/after-replay.txt
cat /home/labex/project/output/after-replay.txt
Espere os quatro valores originais e o offset confirmado 4 novamente. orders-independent ainda deve ter sua própria posição de conclusão; a redefinição não fez esse grupo voltar.
A releitura lê intencionalmente registros existentes outra vez. Uma falha entre o tratamento de um registro e a confirmação do progresso também pode fazer com que um registro seja entregue novamente. Este exercício de console demonstra posições de leitura, não uma transação atômica de gravação em arquivo e confirmação nem uma garantia de processamento exatamente uma vez de ponta a ponta. Os próximos laboratórios de Python tornarão explícita a ordem entre processamento e confirmação.
Resumo
Você inspecionou próximos offsets confirmados e lag, retomou a partir do progresso salvo apesar da política inicial e comparou um grupo independente. A simulação e a execução de uma redefinição com o grupo inativo permitiram uma releitura deliberada, enquanto os registros reais do tópico permaneceram inalterados. A seguir, você observará como vários membros de um grupo compartilham partições e recuperam uma atribuição após a saída de um membro.
