Introducción
El consumidor que terminó correctamente escribió un archivo y después confirmó el progreso de lectura. Son dos operaciones independientes. Este laboratorio hace visibles dos casos de fallo: un registro JSON mal formado debe detener el procesamiento antes de que el progreso lo sobrepase, y un proceso que termina después de escribir, pero antes de confirmar, puede recibir de nuevo el mismo registro.
Primero establecerá una entrega normal, después inspeccionará la posición exacta de un registro incorrecto y, por último, utilizará una opción de entrenamiento suministrada para detenerse entre el vaciado del búfer y la confirmación. Al reiniciar el mismo grupo, la entrega repetida quedará visible en el archivo mediante el mismo identificador de evento y la misma posición de Kafka. Observará ese comportamiento sin implementar deduplicación ni un sistema de reintentos.
Este entorno independiente prepara tres temas de una partición y el consumidor completo que conoce del laboratorio anterior. La opción adicional de salida controlada ya está implementada. Necesita los conceptos anteriores de JSON, archivos y confirmación manual en Python; no requiere reconfigurar el bróker ni diseñar una nueva arquitectura de recuperación. Utilice la Terminal.
Establezca la ruta de procesamiento correcto
En este paso confirmará que la aplicación funciona con una entrada válida antes de diagnosticar un defecto. Cada tema tiene una partición para que las posiciones de los registros sean directas. orders-normal contiene dos registros válidos. El consumidor utiliza la misma secuencia explícita de escritura, vaciado del búfer, confirmación síncrona del mensaje y cierre que antes.
Inspeccione la aplicación suministrada, incluida su opción de salida controlada, sin modificarla:
cd /home/labex/project
cat /home/labex/project/consumer.py
La opción está desactivada de forma predeterminada. Ejecute la ruta normal de dos registros con su propio grupo y archivo:
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
Deben aparecer normal1 y normal2 en los desplazamientos 0 y 1, con el siguiente desplazamiento confirmado 2 después de ambas escrituras. max-messages es la cantidad solicitada en esta invocación; el archivo al que solo se añaden datos puede conservar observaciones de varias invocaciones en los pasos posteriores.
Esta referencia de funcionamiento correcto permite distinguir un problema de conexión general de la aplicación de un fallo propio de un registro. Utilizará temas, grupos y archivos diferentes para los fallos deliberados, de modo que las pruebas resulten claras.
Deténgase ante JSON mal formado sin confirmar más allá
En este paso inspeccionará un registro inicial válido, un registro JSON no válido y otro registro válido posterior. Lea las pruebas locales:
cat /home/labex/project/data/bad-events.txt
En la partición 0 de orders-bad, el registro inicial válido está en el desplazamiento 0, el texto mal formado en el 1 y el evento válido posterior en el 2. La política consiste en detenerse inmediatamente ante el fallo: informar de la posición incorrecta y terminar. No capture el error para continuar y confirmar registros posteriores de esa partición.
Solicite la entrega de los tres registros. Se espera que el siguiente comando falle.
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
Debe observar un estado distinto de cero y Decode failure que identifique topic=orders-bad partition=0 offset=1. El archivo debe contener únicamente bad-prefix en el desplazamiento 0. Su escritura correcta se ha confirmado, por lo que el siguiente desplazamiento del grupo es 1; el registro incorrecto y el posterior siguen sin procesar.
Reinicie el mismo grupo y añada datos al mismo archivo. Se espera que el siguiente comando falle.
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
Debe volver a fallar el mismo desplazamiento 1. El registro inicial válido no debe escribirse por segunda vez porque ya se había confirmado su siguiente posición 1. El evento válido posterior no debe aparecer en el archivo. Así queda visible la política: un operador debe decidir cómo tratar los datos incorrectos, mientras que esta aplicación mínima no los omite silenciosamente ni confirma más allá de ellos.
Termine después de escribir correctamente y antes de confirmar
En este paso creará otro intervalo de fallo mediante el tema preparado orders-replay. Solo contiene repeat1 en el desplazamiento 0. La opción de entrenamiento --exit-before-commit termina con el estado 73 inmediatamente después del vaciado del búfer del archivo y antes de la confirmación explícita. Registra la posición real y aun así cierra el consumidor con las confirmaciones automáticas desactivadas.
Utilice un grupo nuevo y otro archivo. Se espera que el siguiente comando falle.
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
Debe observar el estado 73 y una estructura guardada de repeat1 en la partición 0, desplazamiento 0. El archivo de registro identifica CONTROLLED_EXIT_AFTER_WRITE_BEFORE_COMMIT. Es una demostración de fallo suministrada, no un cambio que deba implementar.
El archivo ya contiene el registro, mientras que el grupo no tiene una posición confirmada para él. Por tanto, se aplicarán las reglas anteriores de política de inicio cuando ese mismo grupo vuelva a arrancar. Esta separación es precisamente el intervalo que observará en el siguiente paso.
Reinicie e identifique el registro de Kafka entregado de nuevo
En este paso reiniciará replay-reader sin la opción de entrenamiento. La ausencia de una posición confirmada y su política earliest permiten volver a leer el registro retenido. Mantenga el modo de adición y el mismo archivo de salida para que ambas entregas queden visibles:
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
Deben aparecer dos estructuras con el mismo event_id, clave, valor, tema, partición y desplazamiento. Sigue habiendo un único registro original en el bróker. La segunda invocación confirma correctamente el siguiente desplazamiento 1, mientras que el archivo conserva pruebas de ambas entregas.
Utilice las siguientes pruebas para distinguir situaciones habituales:
| Observación | Primera prueba útil |
|---|---|
| No hay eventos nuevos | Tema válido y progreso del grupo al final; una consulta acotada puede no devolver registros |
| Nombre de tema incorrecto | Compare el nombre suministrado con la lista y la descripción de temas; este cliente desactiva la creación accidental de temas |
| No hay una posición confirmada válida | Inspeccione el progreso del grupo y la política inicial earliest configurada; puede leer el historial retenido |
| Fallo de decodificación | Consulte el tema, la partición y el desplazamiento indicados y las pruebas del valor; el grupo se detiene antes de superar esa posición incorrecta |
Escribir un archivo y confirmar una posición de Kafka no constituyen una única transacción. Una interrupción entre ambas operaciones puede provocar una entrega repetida, como demuestra esta posición real del registro. Este ejercicio conserva ambas entregas y no implementa deduplicación, una cola de mensajes no procesables ni un diseño transaccional. El desafío final aplica la ruta correcta con un procedimiento explícito para reintentar desde cero.
Resumen
Ha establecido una entrega normal, detenido el procesamiento ante JSON mal formado sin confirmar más allá y observado que un registro escrito con el búfer ya vaciado, pero sin confirmar, se entrega de nuevo tras reiniciar. La coincidencia del contenido del evento y de la posición de Kafka ha explicado las observaciones duplicadas del archivo sin inventar un segundo evento en el bróker. El desafío final combina ahora la publicación correcta, el consumo acotado, la salida a archivos y la escritura antes de confirmar con datos nuevos.
