Introducción
Kafka conserva los registros con independencia de quién los haya leído. Un grupo de consumidores también necesita recordar su progreso para que un proceso posterior pueda continuar en lugar de empezar de nuevo. Un desplazamiento confirmado es la siguiente posición que ese grupo debe leer en una partición, no el número del último registro leído.
Leerá dos de los cuatro registros preparados, inspeccionará el progreso confirmado y el retraso, reanudará la lectura con el mismo grupo y lo comparará con un grupo nuevo. Después previsualizará y ejecutará un restablecimiento mientras el grupo original está inactivo, y volverá a leer los registros retenidos. El tema tiene una partición para facilitar el razonamiento sobre las posiciones.
Este laboratorio independiente proporciona el bróker, cuatro registros y una herramienta auxiliar que consulta repetidamente el estado real del coordinador hasta que aparece el desplazamiento esperado. Basta con conocimientos básicos de terminal; no necesita programar en Python ni configurar servicios. La herramienta no modifica desplazamientos ni inventa informes. Utilice la Terminal.
Lea dos registros e inspeccione la siguiente posición confirmada
En este paso relacionará una lectura acotada con la posición real guardada del grupo. Inspeccione los cuatro valores preparados:
cd /home/labex/project
cat /home/labex/project/data/progress-events.txt
Sus desplazamientos en este tema nuevo de una partición van de 0 a 3. Un grupo nuevo llamado orders-review no tiene una posición guardada, por lo que --from-beginning selecciona el principio de los datos retenidos para esta lectura inicial.
--command-property pasa un valor de configuración al cliente consumidor, en lugar de cambiar el formato de presentación de los eventos. En este ejercicio, el cliente de consola confirma el progreso automáticamente. max.poll.records=1 limita cada consulta a un registro devuelto, de modo que detenerse después de mostrar dos registros no avance la posición por un lote devuelto más grande. El intervalo establece el comportamiento de confirmación periódica; al cerrarse, el cliente también completa su gestión normal del progreso. Inspeccionará el resultado real en lugar de suponer un tiempo de espera concreto:
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
Deben aparecer los dos eventos created y después el cliente termina. Ejecute la herramienta de consulta con tiempo limitado suministrada, indicando el nombre del grupo y el siguiente desplazamiento 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
La herramienta consulta repetidamente la descripción del grupo real hasta que su desplazamiento confirmado es 2, o falla al alcanzar su plazo límite. No confirma ni restablece nada. Localice CURRENT-OFFSET 2, LOG-END-OFFSET 4 y LAG 2 para la partición 0. Es normal que un mensaje indique que el grupo no tiene miembros activos después de terminar el cliente de lectura acotada; el progreso guardado sigue existiendo.
El desplazamiento 2 significa que la siguiente lectura comienza con el tercer registro almacenado. El desplazamiento final 4 es la siguiente posición después de los cuatro registros existentes; el retraso 4 − 2 = 2 mide los registros pendientes en este ejemplo sencillo con el estado ya estabilizado. Los dos primeros registros no se han eliminado.
Reanude el mismo grupo y compárelo con uno nuevo
En este paso anticipará qué registros recibe un grupo a partir de su posición guardada. Ejecute el mismo grupo de nuevo y conserve deliberadamente la opción de inicio desde el principio:
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
Deben aparecer los dos eventos paid, no los eventos created anteriores. --from-beginning proporciona una política de inicio cuando no existe un desplazamiento confirmado válido; no sustituye el desplazamiento válido 2 que este grupo tiene guardado. Confirme que la lectura ha terminado mediante el estado real del coordinador:
/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
Debe observar el desplazamiento confirmado 4, el desplazamiento final 4 y el retraso 0. Ahora anticipe qué ocurrirá con otro grupo llamado orders-independent. No tiene una posición guardada y es independiente de orders-review, por lo que una lectura desde el principio debe devolver los cuatro registros retenidos:
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
Deben aparecer los cuatro valores y el desplazamiento confirmado 4 propio de ese grupo nuevo. Ambos grupos pueden llegar al final tras seguir historiales de lectura diferentes. El progreso de un grupo no ha movido la posición inicial del otro, y el consumo de ninguno de los grupos ha eliminado los registros del tema.
A continuación, cambiará deliberadamente la posición guardada del grupo original, en lugar de interpretar la opción de inicio desde el principio como un comando para retroceder.
Previsualice el restablecimiento de un grupo inactivo y repita la lectura del historial
En este paso restablecerá únicamente orders-review. Su consumidor de lectura acotada ha terminado, por lo que el grupo no tiene miembros activos. Detenga cualquier miembro del mismo grupo que haya dejado ejecutándose por error antes de restablecerlo; un consumidor activo podría seguir cambiando el progreso.
--reset-offsets selecciona la operación de gestión del progreso, --topic limita su alcance y --to-earliest selecciona el principio de los datos retenidos. Primero utilice --dry-run para previsualizar el nuevo desplazamiento propuesto sin aplicarlo:
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
El NEW-OFFSET propuesto debe ser 0. Confirme que la previsualización no ha cambiado la posición guardada actual:
/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
Ahora aplique ese restablecimiento de alcance limitado con --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
El siguiente desplazamiento confirmado real debe ser ahora 0, con el desplazamiento final 4 y el retraso 4. Los registros permanecen en su sitio; solo ha cambiado la posición guardada de este grupo. Vuelva a leer los cuatro utilizando el desplazamiento inicial confirmado, que ahora es válido, sin depender de una opción de política de inicio:
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
Deben aparecer los cuatro valores originales y, de nuevo, el desplazamiento confirmado 4. orders-independent debe conservar su propia posición de lectura completada; el restablecimiento no ha hecho retroceder a ese grupo.
La repetición de lectura vuelve a leer registros existentes de forma intencionada. Un fallo entre el procesamiento de un registro y la confirmación del progreso también puede hacer que el registro se entregue otra vez. Este ejercicio de consola demuestra posiciones de lectura, no una transacción atómica que escriba un archivo y confirme el progreso ni una garantía de procesamiento exactamente una vez de extremo a extremo. Los laboratorios posteriores de Python harán explícito el orden de procesamiento y confirmación.
Resumen
Ha inspeccionado los siguientes desplazamientos confirmados y el retraso, reanudado la lectura desde el progreso guardado pese a una política de inicio desde el principio y comparado un grupo independiente. La simulación y posterior ejecución del restablecimiento de un grupo inactivo permitieron repetir la lectura de forma deliberada mientras los registros reales del tema permanecían intactos. A continuación, observará cómo varios miembros de un grupo comparten particiones y recuperan una asignación cuando un miembro se retira.
