Introdução
Um consumidor que exibe um evento não necessariamente salvou uma entrega útil. Esta aplicação decodificará cada valor, preservará sua chave e posição no Kafka em um arquivo JSONL local e confirmará o progresso somente depois que a gravação correspondente tiver sido bem-sucedida e seus dados tiverem sido descarregados do buffer.
Você mapeará o modelo de grupos e offsets já conhecido para Python: subscribe escolhe o tópico, poll solicita registros e uma confirmação síncrona explícita salva o próximo offset da mensagem tratada. A confirmação automática e o armazenamento automático de offsets ficam desativados para não avançarem o progresso independentemente do seu código. A gravação em arquivo e as confirmações no Kafka continuam sendo operações separadas; esta não é uma transação atômica entre sistemas.
Este laboratório independente prepara os seis eventos iniciais de pedidos conhecidos do laboratório do produtor em um novo tópico. Nenhum estado anterior da VM é necessário. Você precisa conhecer funções, laços, dicionários, JSON e execução de scripts em Python. O analisador de argumentos e a estrutura de prazo são fornecidos. Use o Terminal e o ambiente bigdata-python instalado.
Assine o tópico, consulte e decodifique um registro real
Nesta etapa, você inspecionará um registro recebido sem confirmar o progresso da prévia. Primeiro, leia os valores de eventos preparados:
cd /home/labex/project
cat /home/labex/project/data/events.jsonl
Um Consumer é configurado com a identidade de um grupo e uma política inicial earliest. subscribe registra o tópico desejado; poll retorna uma mensagem, uma mensagem com erro ou None quando nenhum registro está disponível, inclusive durante a atribuição. O laço continua consultando dentro de um prazo geral.
A estrutura de fluxo de controle fornecida mantém esta leitura limitada. time.monotonic() mede o tempo decorrido sem depender da data do relógio. continue inicia a próxima consulta quando nenhuma mensagem está pronta; break sai do laço após a exibição de um registro. O else do laço while executa apenas se o laço terminar sem esse break, o que aqui significa que nenhum registro chegou antes do prazo.
finally fecha o consumidor tanto quando a leitura tem sucesso quanto quando gera um erro. Entenda essa estrutura como espera e encerramento ao redor das chamadas ao Kafka; você não precisa projetar uma estrutura de novas tentativas.
Esta prévia usa um grupo separado, file-preview, com ambas as configurações automáticas desativadas. O valor decodificado é um dicionário Python; o objeto que o envolve também preserva tópico, partição, offset e uma chave decodificada. Nenhuma chamada a commit é feita:
cat > /home/labex/project/preview_consumer.py <<'PYTHON'
import json,time
from confluent_kafka import Consumer
consumer = Consumer({"bootstrap.servers":"localhost:9092", "group.id":"file-preview", "auto.offset.reset":"earliest", "enable.auto.commit":False, "enable.auto.offset.store":False, "allow.auto.create.topics":False})
deadline = time.monotonic() + 20
try:
consumer.subscribe(["orders-file"])
while time.monotonic() < deadline:
message = consumer.poll(1)
if message is None:
continue
if message.error() is not None:
raise RuntimeError(str(message.error()))
value = json.loads(message.value().decode("utf-8"))
record = {"topic":message.topic(), "partition":message.partition(), "offset":message.offset(), "key":message.key().decode("utf-8"), "value":value}
print(json.dumps(record, ensure_ascii=False))
break
else:
raise SystemExit("No preview record arrived before the deadline.")
finally:
consumer.close()
PYTHON
bigdata-python /home/labex/project/preview_consumer.py > /home/labex/project/output/one-event.json
cat /home/labex/project/output/one-event.json
Espere um envelope JSON cujo value contenha um dos eventos de py1 a py6 e cuja key corresponda ao order_id desse evento. Entre três partições, o primeiro evento observado não é fixo. A partição e o offset identificam seu registro real no broker.
Fechar esta prévia não confirma o progresso porque as confirmações automáticas estão desativadas e o script não fez nenhuma confirmação explícita. A próxima aplicação usa seu próprio grupo para a entrega em arquivo.
Grave e descarregue cada envelope antes de confirmar sua mensagem
Nesta etapa, você criará o consumidor de arquivo com limites. Leia com atenção sua ordem de processamento: decodificar e validar o objeto JSON e a chave, gravar um envelope JSONL, descarregar o buffer do arquivo, confirmar a mensagem de forma síncrona e, então, contá-la como concluída. Se uma gravação, decodificação ou confirmação falhar, a aplicação encerra em vez de ignorar o erro e depois fingir que o lote foi bem-sucedido.
A instrução with fecha o arquivo local quando seu bloco termina. Path identifica o nome do arquivo local. A saída abre em modo de acréscimo para que um reinício não apague silenciosamente evidências de entregas anteriores. A primeira execução limpa começa sem arquivo. max-messages é a meta desta execução; o prazo geral faz uma execução incompleta encerrar com status diferente de zero. JSONL armazena um objeto JSON completo por linha, sem um array ao redor. flush() transfere os dados do buffer de arquivo do Python para o sistema operacional antes da solicitação de confirmação. Isso não garante a sobrevivência a uma falha de energia ou de disco e não torna a gravação em arquivo e a confirmação no Kafka uma transação atômica. Este exercício observa a interrupção de processos e o progresso salvo na VM em execução.
commit(message=..., asynchronous=False) confirma o próximo offset da mensagem tratada para sua partição. Os erros retornados por partição são verificados. enable.auto.offset.store=false impede o armazenamento automático de progresso não relacionado, enquanto o argumento explícito message identifica exatamente qual posição tratada está sendo confirmada.
Salve a estrutura completa fornecida de analisador e laço com esta sequência de processamento:
cat > /home/labex/project/consumer.py <<'PYTHON'
import argparse,json,sys,time
from pathlib import Path
from confluent_kafka import Consumer
parser = argparse.ArgumentParser(description="Save a finite JSON event delivery and commit only written records.")
parser.add_argument("--topic", required=True)
parser.add_argument("--group", required=True)
parser.add_argument("--output", required=True)
parser.add_argument("--max-messages", type=int, default=6)
parser.add_argument("--timeout-seconds", type=float, default=30)
parser.add_argument("--bootstrap-server", default="localhost:9092")
args = parser.parse_args()
if args.max_messages <= 0 or args.timeout_seconds <= 0:
parser.error("Message count and timeout must be positive.")
consumer = Consumer({
"bootstrap.servers": args.bootstrap_server,
"group.id": args.group,
"auto.offset.reset": "earliest",
"enable.auto.commit": False,
"enable.auto.offset.store": False,
"allow.auto.create.topics": False,
})
received = 0
deadline = time.monotonic() + args.timeout_seconds
try:
consumer.subscribe([args.topic])
with Path(args.output).open("a", encoding="utf-8") as sink:
while received < args.max_messages and time.monotonic() < deadline:
message = consumer.poll(1)
if message is None:
continue
position = f"topic={message.topic()} partition={message.partition()} offset={message.offset()}"
if message.error() is not None:
raise RuntimeError(f"Kafka error {position}: {message.error()}")
try:
key = message.key().decode("utf-8") if message.key() is not None else None
value = json.loads(message.value().decode("utf-8"))
if not isinstance(value, dict) or not isinstance(value.get("event_id"), str) or not isinstance(value.get("order_id"), str):
raise ValueError("Expected an object with string event_id and order_id.")
if key != value["order_id"]:
raise ValueError("Record key does not match order_id.")
except (UnicodeDecodeError, json.JSONDecodeError, ValueError) as error:
print(f"Decode failure {position}: {error}", file=sys.stderr)
raise SystemExit(1)
record = {"topic": message.topic(), "partition": message.partition(), "offset": message.offset(), "key": key, "value": value}
sink.write(json.dumps(record, ensure_ascii=False) + "\n")
sink.flush()
committed = consumer.commit(message=message, asynchronous=False)
if not committed or any(item.error is not None for item in committed):
raise RuntimeError(f"Commit failed {position}: {committed}")
received += 1
print(f"SAVED_AND_COMMITTED event_id={value['event_id']} {position} next_offset={message.offset()+1}")
if received < args.max_messages:
print(f"Only received {received} of {args.max_messages} requested records before the deadline.", file=sys.stderr)
raise SystemExit(2)
print("Saved and committed records:", received)
finally:
consumer.close()
PYTHON
bigdata-python /home/labex/project/consumer.py --topic orders-file --group file-writer --output /home/labex/project/output/received-events.jsonl --max-messages 6 --timeout-seconds 30 > /home/labex/project/output/consumer.log 2>&1
cat /home/labex/project/output/consumer.log
cat /home/labex/project/output/received-events.jsonl
Espere Saved and committed records: 6 e seis envelopes JSONL. Cada envelope deve preservar o valor completo, a chave correspondente e a posição real do registro. Um callback ou log de console não substitui o arquivo salvo; a próxima etapa compara as evidências do arquivo com o progresso confirmado real.
Relacione o arquivo aos próximos offsets confirmados por partição
Nesta etapa, você inspecionará o progresso no coordenador usando a ferramenta de linha de comando que já conhece:
kafka-consumer-groups.sh --bootstrap-server localhost:9092 --group file-writer --describe > /home/labex/project/output/file-writer-position.txt
cat /home/labex/project/output/file-writer-position.txt
Para cada partição que contém registros entregues, CURRENT-OFFSET deve alcançar o LOG-END-OFFSET dessa partição, com lag zero. As atribuições exatas de partições e a distribuição não são fixadas nesta lição. Uma partição sem registros pode ainda não ter posição confirmada; nela não há mensagem tratada para confirmar.
Compare o offset de um envelope salvo com o próximo offset confirmado de sua partição. Se o último registro tratado de uma partição estiver no offset n, a próxima posição confirmada será n+1. Essa próxima posição não é uma contagem global entre todas as partições.
O consumidor já fechou, portanto uma mensagem de ausência de membros ativos é normal. O arquivo armazenado e o progresso salvo do grupo permanecem. Você tem evidências separadas da saída do processamento e do estado do coordenador, sem afirmar que foram gravados em uma única transação.
Reinicie o mesmo grupo e observe a ausência de novas entregas
Nesta etapa, você testará a posição salva em vez de presumir que earliest sempre volta ao início. Os seis registros continuam retidos, mas file-writer tem posições finais confirmadas válidas. Preveja outra inicialização com esse mesmo grupo: ele não deve encontrar registros novos, mesmo com auto.offset.reset definido como earliest.
Use um caminho de saída separado para preservar o arquivo original de seis registros. Solicite um registro com um prazo geral curto. Como a quantidade solicitada não pode ser alcançada, esta aplicação informa deliberadamente um status de lote incompleto diferente de zero. O próximo comando deve falhar.
bigdata-python /home/labex/project/consumer.py --topic orders-file --group file-writer --output /home/labex/project/output/no-new-events.jsonl --max-messages 1 --timeout-seconds 5 > /home/labex/project/output/same-group-restart.log 2>&1
Capture seu status imediatamente:
printf '%s\n' "$?" > /home/labex/project/output/same-group-exit-status.txt
cat /home/labex/project/output/same-group-exit-status.txt
cat /home/labex/project/output/same-group-restart.log
Espere um status diferente de zero e Only received 0 of 1 requested records before the deadline. O novo arquivo deve estar vazio; o received-events.jsonl original ainda contém seis envelopes. Os offsets confirmados válidos do mesmo grupo não foram substituídos por sua política inicial earliest.
Para ler intencionalmente o histórico retido outra vez, use o fluxo já ensinado de grupo distinto ou redefinição de grupo inativo. Se um processo parar depois de gravar, mas antes de confirmar, a relação será diferente: o mesmo grupo poderá receber o registro novamente. O próximo laboratório tornará visível essa janela de falha.
Resumo
Você assinou um tópico e consultou registros com Python, decodificou registros JSON/UTF-8 e preservou chaves e posições do Kafka em JSONL. Os flushes bem-sucedidos precederam confirmações síncronas explícitas, e o estado real do coordenador correspondeu ao arquivo concluído. Um reinício do mesmo grupo não encontrou registros novos, sem excluir o histórico retido do tópico. A seguir, uma falha de decodificação e uma interrupção entre a gravação e a confirmação mostrarão por que registros podem ser entregues novamente.
