Introducción
Un consumidor que muestra un evento no necesariamente ha guardado una entrega útil. Esta aplicación decodificará cada valor, conservará su clave y posición de Kafka en un archivo JSONL local y confirmará el progreso solo después de que la escritura correspondiente haya terminado correctamente y se haya vaciado su búfer.
Aplicará en Python el modelo de grupos y desplazamientos que ya conoce: subscribe selecciona el tema, poll solicita registros y una confirmación síncrona explícita guarda el siguiente desplazamiento del mensaje procesado. La confirmación automática y el almacenamiento automático de desplazamientos están desactivados para que no avancen el progreso al margen de su código. La escritura del archivo y las confirmaciones de Kafka siguen siendo operaciones independientes; no constituyen una transacción atómica entre sistemas.
Este laboratorio independiente prepara en un tema nuevo los seis eventos iniciales de pedidos que conoce del laboratorio del productor. No requiere ningún estado de una máquina virtual anterior. Necesita conocimientos de funciones, bucles, diccionarios y JSON en Python, así como de ejecución de scripts. La estructura del analizador de argumentos y del plazo límite ya se proporciona. Utilice la Terminal y el entorno bigdata-python instalado.
Suscríbase, consulte y decodifique un registro real
En este paso inspeccionará un registro recibido sin confirmar el progreso de la vista previa. Lea primero los valores de los eventos preparados:
cd /home/labex/project
cat /home/labex/project/data/events.jsonl
Un Consumer se configura con una identidad de grupo y una política inicial earliest. subscribe registra el tema deseado; poll devuelve un mensaje, un mensaje que contiene un error o None cuando no hay ningún registro disponible, incluso durante la asignación. El bucle sigue consultando dentro de un plazo límite general.
La estructura de control suministrada mantiene esta lectura acotada. time.monotonic() mide el tiempo transcurrido sin depender de la fecha del reloj del sistema. continue inicia la siguiente consulta cuando no hay un mensaje listo; break sale del bucle después de mostrar un registro. El else del bucle while solo se ejecuta si el bucle termina sin ese break, lo que aquí significa que no ha llegado ningún registro antes del plazo límite.
finally cierra el consumidor tanto si la lectura tiene éxito como si genera un error. Interprete esta estructura como la espera y el cierre que rodean las llamadas a Kafka; no necesita diseñar un sistema de reintentos.
Esta vista previa utiliza un grupo independiente, file-preview, con ambos ajustes automáticos desactivados. El valor decodificado es un diccionario Python; la estructura que lo contiene también conserva topic, partition, offset y una clave decodificada. No se realiza ninguna llamada de confirmación:
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
Debe aparecer una estructura JSON cuyo value contenga uno de los eventos de py1 a py6 y cuya key coincida con el order_id de ese evento. Al haber tres particiones, el primer evento observado no es fijo. La partición y el desplazamiento identifican su registro real en el bróker.
Cerrar esta vista previa no confirma el progreso porque las confirmaciones automáticas están desactivadas y el script no ha realizado ninguna confirmación explícita. La siguiente aplicación utiliza su propio grupo para la entrega al archivo.
Escriba cada registro con sus metadatos y vacíe el búfer antes de confirmar el mensaje
En este paso creará el consumidor de archivos con lectura acotada. Lea con atención su orden de procesamiento: decodificar y validar el objeto JSON y la clave, escribir una estructura JSONL, vaciar el búfer del archivo, confirmar ese mensaje de forma síncrona y después contarlo como completado. Si falla una escritura, una decodificación o una confirmación, la aplicación termina en lugar de saltarse el error y fingir después que el lote se ha completado correctamente.
La instrucción with cierra el archivo local cuando termina su bloque. Path identifica el nombre del archivo local. La salida se abre en modo de adición para que un reinicio no trunque silenciosamente las pruebas de entrega anteriores. La primera ejecución desde cero comienza sin archivo. max-messages es el objetivo de esta invocación; el plazo límite general hace que una ejecución incompleta termine con un estado distinto de cero. JSONL almacena un objeto JSON completo por línea, sin un arreglo que los englobe. flush() transfiere los datos del búfer de archivo de Python al sistema operativo antes de solicitar la confirmación. No garantiza que sobrevivan a un fallo de alimentación o de disco, ni convierte el archivo y la confirmación de Kafka en una única transacción atómica. Este ejercicio observa la interrupción de procesos y el progreso guardado en la máquina virtual en ejecución.
commit(message=..., asynchronous=False) confirma el siguiente desplazamiento del mensaje procesado para su partición. Se comprueban los errores por partición devueltos. enable.auto.offset.store=false impide el almacenamiento automático de progreso ajeno al procesamiento, mientras que el argumento message explícito identifica exactamente qué posición procesada se confirma.
Guarde la estructura completa del analizador y el bucle suministrados con esta secuencia de procesamiento:
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
Debe observar Saved and committed records: 6 y seis estructuras JSONL. Cada una debe conservar el valor completo, la clave correspondiente y la posición real del registro. Una función de retorno o un registro de consola no sustituyen al archivo guardado; el siguiente paso compara las pruebas del archivo con el progreso confirmado real.
Relacione el archivo con los siguientes desplazamientos confirmados por partición
En este paso inspeccionará el progreso del coordinador con la herramienta de línea de comandos que ya conoce:
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
En cada partición que contenga registros entregados, CURRENT-OFFSET debe alcanzar el LOG-END-OFFSET de esa partición, con retraso cero. Las asignaciones exactas y la distribución entre particiones no están fijadas en esta lección. Una partición sin registros puede no tener aún una posición confirmada; no hay ningún mensaje procesado que confirmar en ella.
Compare el offset de una estructura guardada con el siguiente desplazamiento confirmado de su partición. Si el último registro procesado de una partición está en el desplazamiento n, la siguiente posición confirmada es n+1. Esa posición siguiente no es un recuento global de todas las particiones.
El consumidor se ha cerrado, por lo que es normal que aparezca un mensaje de ausencia de miembros activos. El archivo almacenado y el progreso guardado del grupo permanecen. Dispone de pruebas independientes de la salida del procesamiento y del estado del coordinador, sin afirmar que se hayan escrito en una única transacción.
Reinicie el mismo grupo y observe que no hay nuevas entregas
En este paso pondrá a prueba la posición guardada en lugar de suponer que earliest siempre hace retroceder la lectura. Los seis registros siguen retenidos, pero file-writer tiene posiciones finales confirmadas válidas. Anticipe qué ocurrirá al iniciar otra vez ese mismo grupo: no debe encontrar registros nuevos, aunque auto.offset.reset sea earliest.
Utilice otra ruta de salida para conservar el archivo original de seis registros. Solicite un registro con un plazo límite general breve. Como no se puede alcanzar la cantidad solicitada, esta aplicación informa deliberadamente de un lote incompleto mediante un estado distinto de cero. Se espera que el siguiente comando falle.
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 su estado inmediatamente:
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
Debe observar un estado distinto de cero y Only received 0 of 1 requested records before the deadline. El archivo nuevo debe estar vacío; el received-events.jsonl original sigue conteniendo seis estructuras. La política inicial earliest no ha sustituido los desplazamientos confirmados válidos del mismo grupo.
Para volver a leer intencionadamente el historial retenido, utilice el procedimiento ya explicado de otro grupo o de restablecimiento de un grupo inactivo. Si un proceso se detiene después de escribir, pero antes de confirmar, la relación es distinta: el mismo grupo puede recibir de nuevo el registro. El siguiente laboratorio hará visible ese intervalo de fallo.
Resumen
Ha realizado suscripciones y consultas con Python, decodificado registros JSON en UTF-8 y conservado las claves y posiciones de Kafka en JSONL. El vaciado correcto del búfer ha precedido a las confirmaciones síncronas explícitas, y el estado real del coordinador ha coincidido con el archivo completo. Al reiniciar el mismo grupo no se han encontrado registros nuevos, sin eliminar el historial retenido del tema. A continuación, un fallo de decodificación y una interrupción entre la escritura y la confirmación mostrarán por qué los registros pueden entregarse de nuevo.
