Introdução
O consumidor bem-sucedido gravou um arquivo e depois confirmou o progresso de leitura. Essas são duas operações separadas. Este laboratório torna visíveis dois casos de falha: um registro JSON malformado deve interromper o processamento antes que o progresso avance além dele, e um processo que encerra depois de gravar, mas antes de confirmar, pode receber o mesmo registro novamente.
Primeiro, você estabelecerá uma entrega normal, depois inspecionará a posição exata de um registro inválido e, por fim, usará uma opção de treinamento fornecida para parar entre flush e commit. Reiniciar o mesmo grupo revelará a reentrega no arquivo pelo mesmo ID de evento e pela mesma posição no Kafka. Você observará esse comportamento sem implementar desduplicação nem uma estrutura de novas tentativas.
Este ambiente independente prepara três tópicos com uma partição e o consumidor completo conhecido do laboratório anterior. A opção adicional de saída controlada já está implementada. Você precisa dos conceitos anteriores de JSON, arquivos e confirmação manual em Python; não é necessário reconfigurar o broker nem criar uma nova arquitetura de recuperação. Use o Terminal.
Estabeleça o fluxo de processamento bem-sucedido
Nesta etapa, você confirmará que a aplicação funciona com uma entrada válida antes de diagnosticar um defeito. Cada tópico tem uma partição, portanto as posições dos registros são diretas. orders-normal contém dois registros válidos. O consumidor usa a mesma sequência explícita de gravação, flush, confirmação síncrona da mensagem e fechamento de antes.
Inspecione a aplicação fornecida, incluindo sua opção de saída controlada, sem alterá-la:
cd /home/labex/project
cat /home/labex/project/consumer.py
A opção está desativada por padrão. Execute o fluxo normal de dois registros com seu próprio grupo e arquivo:
bigdata-python /home/labex/project/consumer.py --topic orders-normal --group normal-reader --output /home/labex/project/output/normal-delivery.jsonl --max-messages 2 --timeout-seconds 20 > /home/labex/project/output/normal.log 2>&1
cat /home/labex/project/output/normal.log
cat /home/labex/project/output/normal-delivery.jsonl
Espere normal1 e normal2 nos offsets 0 e 1, com próximo offset confirmado 2 após ambas as gravações. max-messages é a quantidade solicitada nesta execução; o arquivo que só recebe acréscimos pode preservar observações de várias execuções nas etapas posteriores.
Esta referência bem-sucedida separa um problema de conexão de toda a aplicação de uma falha específica de registro. Você usará tópicos, grupos e arquivos diferentes para as falhas deliberadas, mantendo as evidências claras.
Pare no JSON malformado sem confirmar além dele
Nesta etapa, você inspecionará um prefixo válido, um registro JSON inválido e um registro válido posterior. Leia as evidências locais:
cat /home/labex/project/data/bad-events.txt
Na partição 0 de orders-bad, o prefixo válido está no offset 0, o texto malformado no offset 1 e o evento válido posterior no offset 2. A política é falhar imediatamente: informar a posição inválida e encerrar. Não capture o erro para continuar confirmando registros posteriores dessa partição.
Solicite a entrega de três registros. O próximo comando deve falhar.
bigdata-python /home/labex/project/consumer.py --topic orders-bad --group bad-reader --output /home/labex/project/output/bad-prefix.jsonl --max-messages 3 --timeout-seconds 20 > /home/labex/project/output/bad-first.log 2>&1
printf '%s\n' "$?" > /home/labex/project/output/bad-first-exit.txt
cat /home/labex/project/output/bad-first-exit.txt
cat /home/labex/project/output/bad-first.log
cat /home/labex/project/output/bad-prefix.jsonl
Espere um status diferente de zero e Decode failure identificando topic=orders-bad partition=0 offset=1. O arquivo deve conter apenas bad-prefix no offset 0. Sua gravação bem-sucedida foi confirmada, portanto o próximo offset do grupo é 1; o registro inválido e o posterior continuam sem tratamento.
Reinicie o mesmo grupo e acrescente ao mesmo arquivo. O próximo comando deve falhar.
bigdata-python /home/labex/project/consumer.py --topic orders-bad --group bad-reader --output /home/labex/project/output/bad-prefix.jsonl --max-messages 3 --timeout-seconds 20 > /home/labex/project/output/bad-retry.log 2>&1
printf '%s\n' "$?" > /home/labex/project/output/bad-retry-exit.txt
cat /home/labex/project/output/bad-retry-exit.txt
cat /home/labex/project/output/bad-retry.log
kafka-consumer-groups.sh --bootstrap-server localhost:9092 --group bad-reader --describe
O mesmo offset 1 deve falhar novamente. O prefixo válido não deve ser gravado uma segunda vez porque sua próxima posição 1 já foi confirmada. O evento válido posterior não deve aparecer no arquivo. Isso torna a política visível: um operador precisa escolher como resolver o problema nos dados, enquanto esta aplicação mínima não os ignora silenciosamente nem confirma além deles.
Encerre após uma gravação bem-sucedida, mas antes da confirmação
Nesta etapa, você criará uma janela de falha diferente usando o tópico preparado orders-replay. Ele contém apenas repeat1 no offset 0. A opção de treinamento --exit-before-commit encerra com status 73 imediatamente após o flush do arquivo e antes da confirmação explícita. Ela registra a posição real e ainda fecha o consumidor com as confirmações automáticas desativadas.
Use um novo grupo e um arquivo separado. O próximo comando deve falhar.
bigdata-python /home/labex/project/consumer.py --topic orders-replay --group replay-reader --output /home/labex/project/output/redelivery.jsonl --max-messages 1 --timeout-seconds 20 --exit-before-commit > /home/labex/project/output/before-commit-exit.log 2>&1
printf '%s\n' "$?" > /home/labex/project/output/before-commit-exit-status.txt
cat /home/labex/project/output/before-commit-exit-status.txt
cat /home/labex/project/output/before-commit-exit.log
cat /home/labex/project/output/redelivery.jsonl
Espere status 73 e um envelope salvo para repeat1 na partição 0, offset 0. O log identifica CONTROLLED_EXIT_AFTER_WRITE_BEFORE_COMMIT. Esta é uma demonstração de falha fornecida, não uma alteração que você precisa implementar.
O arquivo agora tem o registro, enquanto o grupo não tem uma posição confirmada para ele. Portanto, as regras anteriores de política inicial se aplicam quando esse mesmo grupo inicia novamente. Essa separação é exatamente a janela que a próxima etapa observará.
Reinicie e identifique o registro Kafka repetido
Nesta etapa, você reiniciará replay-reader sem a opção de treinamento. A ausência de posição confirmada e a política earliest permitem ler o registro retido novamente. Mantenha o modo de acréscimo e o mesmo arquivo de saída para que ambas as entregas permaneçam visíveis:
bigdata-python /home/labex/project/consumer.py --topic orders-replay --group replay-reader --output /home/labex/project/output/redelivery.jsonl --max-messages 1 --timeout-seconds 20 > /home/labex/project/output/after-restart.log 2>&1
cat /home/labex/project/output/after-restart.log
cat /home/labex/project/output/redelivery.jsonl
kafka-consumer-groups.sh --bootstrap-server localhost:9092 --group replay-reader --describe
Espere dois envelopes com os mesmos event_id, chave, valor, tópico, partição e offset. Ainda há apenas um registro original no broker. A segunda execução confirma com sucesso o próximo offset 1, enquanto o arquivo preserva evidências das duas entregas.
Use as evidências a seguir para distinguir situações comuns:
| Observação | Primeira evidência útil |
|---|---|
| Nenhum evento novo | Tópico válido com o progresso do grupo no fim; uma consulta com prazo limitado pode não retornar registros |
| Nome de tópico incorreto | Compare o nome recebido com a listagem/descrição de tópicos; este cliente desativa a criação acidental de tópicos |
| Nenhuma posição confirmada válida | Inspecione o progresso do grupo e a política inicial earliest configurada; ela pode ler o histórico retido |
| Falha de decodificação | Leia o tópico/partição/offset informado e as evidências do valor; o grupo para antes dessa posição inválida |
Gravar um arquivo e confirmar uma posição no Kafka não formam uma única transação. Uma interrupção entre essas operações pode causar reentrega, como demonstra esta posição real de registro. Este exercício mantém as duas entregas e não implementa desduplicação, uma fila de mensagens mortas ou um projeto transacional. O desafio final aplica o fluxo bem-sucedido com um procedimento explícito de nova tentativa em execução limpa.
Resumo
Você estabeleceu uma entrega normal, interrompeu o processamento de JSON malformado sem confirmar além dele e observou um registro gravado com flush, mas não confirmado, ser entregue novamente após o reinício. A correspondência do conteúdo do evento e da posição no Kafka explicou as observações duplicadas no arquivo sem inventar um segundo evento no broker. O desafio final agora combina publicação bem-sucedida, consumo com limites, saída em arquivo e gravação antes da confirmação com novos dados.
