Introducción
Un grupo puede repartir las particiones de un tema entre varios consumidores. Cada partición tiene un único propietario activo dentro de un grupo estable, mientras que otro grupo puede leer los mismos registros retenidos de forma independiente. Esto permite compartir el trabajo sin que todas las aplicaciones compitan por una única posición global de lectura.
Iniciará dos clientes de consola de larga duración en un grupo, inspeccionará sus asignaciones reales antes de publicar y comparará las observaciones de otro grupo. Cuando se retire un miembro, esperará a que el restante tenga asignadas ambas particiones antes de enviar otro lote pequeño. El publicador preparado se dirige explícitamente a ambas particiones para que cada una tenga trabajo durante el experimento; se comprueba la entrega real al bróker.
Este laboratorio independiente proporciona un tema de dos particiones, datos de eventos accesibles y herramientas de espera de solo lectura con tiempo limitado. Basta con conocimientos básicos de terminal; las herramientas de datos no requieren editar Python. Los clientes de consola se ejecutan en segundo plano hasta que los detenga en los pasos finales. Mantenga la misma sesión de la Terminal para que las variables de sus procesos sigan disponibles.
Inicie dos miembros y espere a que la asignación se estabilice
En este paso observará la asignación antes de dar trabajo al grupo. El tema preparado tiene dos particiones y ningún registro. Inspeccione sus metadatos:
cd /home/labex/project
kafka-topics.sh --bootstrap-server localhost:9092 --describe --topic orders-shared
Los siguientes clientes utilizan el mismo grupo, orders-team. --command-property pasa cada valor de configuración al cliente consumidor. client.id los identifica como reader-a y reader-b en los informes. Los ajustes de confirmación automática ya conocidos permiten que su progreso de lectura se estabilice mientras siguen consultando.
El símbolo & final inicia un cliente en segundo plano en el intérprete de comandos para que pueda ejecutar comandos de inspección en la misma terminal. $! es el identificador del proceso recién iniciado en segundo plano; la asignación que aparece inmediatamente después lo guarda para detener ese proceso concreto más adelante.
El intérprete de comandos mantiene dos flujos de salida: la salida estándar para los registros y el error estándar para los diagnósticos. > escribe el primer flujo en el archivo JSONL; 2> escribe el flujo de errores en el archivo de registro. La asignación consumer_a_pid=$! guarda una variable sin espacios alrededor de =; "$consumer_a_pid" recupera después ese identificador de proceso concreto. Los registros normales y los diagnósticos van a archivos separados para no interrumpir el indicador de comandos. Estos clientes siguen ejecutándose hasta los comandos kill dirigidos a sus procesos cerca del final:
kafka-console-consumer.sh --bootstrap-server localhost:9092 --topic orders-shared --group orders-team --from-beginning --command-property client.id=reader-a --command-property enable.auto.commit=true --command-property auto.commit.interval.ms=1000 --command-property max.poll.records=1 > /home/labex/project/output/reader-a.jsonl 2> /home/labex/project/output/reader-a.log &
consumer_a_pid=$!
kafka-console-consumer.sh --bootstrap-server localhost:9092 --topic orders-shared --group orders-team --from-beginning --command-property client.id=reader-b --command-property enable.auto.commit=true --command-property auto.commit.interval.ms=1000 --command-property max.poll.records=1 > /home/labex/project/output/reader-b.jsonl 2> /home/labex/project/output/reader-b.log &
consumer_b_pid=$!
Espere a la asignación real de dos miembros y guarde su informe:
/home/labex/project/tools/wait-assignment 2 > /home/labex/project/output/two-members.txt
cat /home/labex/project/output/two-members.txt
La herramienta de solo lectura consulta repetidamente el informe de miembros del grupo hasta que los dos clientes identificados tienen asignaciones no vacías y sin solapamiento cuya unión son las particiones 0 y 1. Tiene un plazo límite y no inventa asignaciones ni cambia el progreso.
En el informe, CLIENT-ID identifica a reader-a o reader-b, y CURRENT-ASSIGNMENT enumera la asignación actual real. Por ejemplo, orders-shared:0 significa la partición 0 de ese tema. Una asignación de destino todavía puede estar pendiente, por lo que debe inspeccionar la columna actual. Cada miembro debe tener asignada una partición, sin exigir que reader-a reciba un número de partición concreto. Un reequilibrio redistribuye las particiones cuando los miembros se incorporan al grupo o se retiran. El grupo necesita tiempo para alcanzar la nueva asignación; que un proceso se inicie no significa que la asignación ya se haya estabilizado. Una asignación estable demuestra el reparto del trabajo; dos identificadores de proceso por sí solos no prueban qué particiones tiene cada miembro.
Publique trabajo y compárelo con un grupo independiente
En este paso proporcionará registros a ambas particiones asignadas y observará cómo otro grupo lee el mismo historial. Inspeccione los lotes accesibles. El lote initial contiene de share1 a share6; followup contiene share7 y share8:
cat /home/labex/project/data/group-batches.json
El publicador suministrado envía los datos seleccionados al bróker real, espera a las funciones de confirmación de entrega y falla si la entrega queda incompleta. Publique los seis eventos iniciales solo después de que la asignación se haya estabilizado:
/home/labex/project/tools/publish-shared-batch initial
/home/labex/project/tools/wait-caught-up orders-team 3 > /home/labex/project/output/team-initial-position.txt
cat /home/labex/project/output/team-initial-position.txt
La herramienta de solo lectura espera hasta que ambas particiones tienen el siguiente desplazamiento confirmado 3, el desplazamiento final 3 y el retraso 0. Cada partición ha recibido tres registros. Inspeccione lo que han observado los dos clientes:
cat /home/labex/project/output/reader-a.jsonl
cat /home/labex/project/output/reader-b.jsonl
Entre ambos archivos deben incluir los seis identificadores iniciales. El informe de asignación activa demuestra que la asignación es estable; un archivo también puede contener observaciones anteriores a un cambio de asignación.
Ahora inicie otro grupo con lectura acotada, orders-observer, desde el principio de los datos retenidos:
kafka-console-consumer.sh --bootstrap-server localhost:9092 --topic orders-shared --group orders-observer --from-beginning --max-messages 6 --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/observer-initial.jsonl
cat /home/labex/project/output/observer-initial.jsonl
/home/labex/project/tools/wait-caught-up orders-observer 3 > /home/labex/project/output/observer-initial-position.txt
cat /home/labex/project/output/observer-initial-position.txt
Debe recibir los seis registros de forma independiente. No quita particiones a orders-team porque pertenece a otro grupo. Ambos grupos pueden estar al día al mismo tiempo y conservar posiciones confirmadas independientes.
Detenga un miembro y observe la asignación del restante
En este paso dejará reader-a en ejecución y detendrá únicamente reader-b. kill envía una señal de terminación al proceso guardado; no detiene el bróker. wait permite que el intérprete de comandos termine de recoger el estado de ese proceso hijo. Un proceso que recibe una señal puede devolver un estado de salida distinto de cero, por lo que el || true final evita que este comando de cierre intencionado se interprete como un fallo de la aplicación:
kill "$consumer_b_pid"
wait "$consumer_b_pid" || true
/home/labex/project/tools/wait-assignment 1 > /home/labex/project/output/one-member.txt
cat /home/labex/project/output/one-member.txt
reader-a debe tener asignadas ambas particiones cuando el grupo se estabilice. La herramienta de espera comprueba el informe real antes de la siguiente publicación. Con dos particiones, añadir más de dos miembros dejaría a algunos sin partición en una asignación estable; este ejercicio no requiere un tercer cliente.
Publique un evento de seguimiento en cada partición y espere a que el miembro restante se ponga al día:
/home/labex/project/tools/publish-shared-batch followup
/home/labex/project/tools/wait-caught-up orders-team 4 > /home/labex/project/output/team-followup-position.txt
cat /home/labex/project/output/team-followup-position.txt
cat /home/labex/project/output/reader-a.jsonl
reader-a debe incluir ahora tanto share7 como share8. reader-b se ha detenido, por lo que no puede procesar estos registros nuevos. El grupo observador todavía conserva su posición confirmada anterior; reanude su lectura para los dos eventos nuevos:
kafka-console-consumer.sh --bootstrap-server localhost:9092 --topic orders-shared --group orders-observer --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/observer-followup.jsonl
cat /home/labex/project/output/observer-followup.jsonl
/home/labex/project/tools/wait-caught-up orders-observer 4 > /home/labex/project/output/observer-followup-position.txt
cat /home/labex/project/output/observer-followup-position.txt
Deben aparecer share7 y share8 y el siguiente desplazamiento 4 propio de cada grupo en ambas particiones. La opción de inicio desde el principio no sustituye el desplazamiento válido 3 que el grupo observador tiene guardado.
Un reequilibrio puede provocar entregas repetidas si un miembro que se retira ha procesado registros antes de que se confirme su progreso. Por ello, las observaciones de eventos pueden incluir repeticiones; una asignación estable sigue dando a cada partición un único propietario a la vez. Este laboratorio comprueba asignaciones reales y el contenido completo de los eventos sin afirmar que la entrega se produzca exactamente una vez de extremo a extremo.
Cierre el cliente restante y conserve las pruebas
En este paso terminará el experimento de larga duración deteniendo el único cliente que queda. Ya se han observado sus posiciones finales al llegar al final de los datos, por lo que no hay trabajo nuevo pendiente. Utilice su propio PID guardado:
kill "$consumer_a_pid"
wait "$consumer_a_pid" || true
/home/labex/project/tools/wait-assignment 0 > /home/labex/project/output/no-members.txt
cat /home/labex/project/output/no-members.txt
El grupo ya no debe tener miembros activos. Los desplazamientos guardados y los eventos del bróker permanecen; la duración de los clientes y el progreso almacenado se gestionan por separado. Los clientes observadores de lectura acotada ya han terminado automáticamente.
Inspeccione los informes de miembros conservados de antes y después de la reasignación:
cat /home/labex/project/output/two-members.txt /home/labex/project/output/one-member.txt
El primer informe demuestra que hay dos propietarios sin solapamiento y el segundo, un único propietario de ambas particiones. Son informes reales con identificadores de miembro propios de cada ejecución, no asignaciones fijas que deba memorizar. Los archivos de lectores y observadores conservan pruebas de eventos de ambos grupos. Ha cerrado únicamente sus dos clientes de consola y ha conservado el tema y las posiciones confirmadas para su revisión.
Resumen
Ha observado cómo dos consumidores comparten asignaciones reales de particiones, cómo otro grupo lee los mismos registros de forma independiente y cómo un único miembro restante asume ambas particiones antes de recibir trabajo nuevo. Las comprobaciones de estado con tiempo limitado han sustituido las suposiciones sobre tiempos de espera, y el cierre de procesos concretos ha detenido los clientes conservando el progreso y los eventos. A continuación, sustituirá la herramienta de publicación preparada por un productor Python cuya serialización y gestión de entregas podrá leer y modificar.
