Introduction
Un groupe peut répartir les partitions d'un topic entre plusieurs consommateurs. Dans un groupe stable, chaque partition a un seul propriétaire actif, tandis qu'un autre groupe peut lire indépendamment les mêmes enregistrements conservés. Cela permet de partager le travail sans mettre toutes les applications en concurrence pour une seule position de lecture globale.
Vous allez démarrer deux clients en console de longue durée dans un même groupe, inspecter leurs affectations réelles avant de publier, puis comparer les observations d'un groupe distinct. Après le départ d'un membre, vous attendrez que le membre restant possède les deux partitions avant d'envoyer un autre petit lot. L'outil de publication préparé cible explicitement les deux partitions afin que chacune ait du travail pendant l'expérience ; sa livraison réelle au broker est vérifiée.
Ce lab indépendant fournit un topic à deux partitions, des jeux d'événements consultables et des outils d'attente en lecture seule à durée limitée. Des connaissances élémentaires du terminal suffisent ; les outils fournis ne nécessitent aucune modification de code Python. Les clients en console s'exécutent en arrière-plan jusqu'à leur arrêt dans les dernières étapes. Conservez le même shell du terminal pour que les variables de leurs processus restent disponibles.
Démarrer deux membres et attendre une affectation stable
Dans cette étape, vous allez observer l'affectation avant de donner du travail au groupe. Le topic préparé possède deux partitions et aucun enregistrement. Inspectez ses métadonnées :
cd /home/labex/project
kafka-topics.sh --bootstrap-server localhost:9092 --describe --topic orders-shared
Les clients suivants utilisent le même groupe, orders-team. --command-property transmet chaque valeur de configuration au client consommateur. client.id les désigne par reader-a et reader-b dans les rapports. Les paramètres de validation automatique déjà connus permettent à leur progression de lecture de se stabiliser pendant qu'ils continuent à interroger le broker.
Le & final lance un client en arrière-plan du shell pour que vous puissiez exécuter des commandes d'inspection dans le même terminal. $! est l'identifiant du processus qui vient d'être lancé en arrière-plan ; l'affectation qui suit immédiatement chaque lancement le conserve pour arrêter précisément ce processus plus tard.
Le shell distingue deux flux de sortie : la sortie standard pour les enregistrements et la sortie d'erreur standard pour les diagnostics. > écrit le premier flux dans le fichier JSONL ; 2> écrit le flux d'erreur dans le journal. L'affectation consumer_a_pid=$! enregistre une variable du shell sans espaces autour de = ; "$consumer_a_pid" permet ensuite de récupérer l'identifiant de ce processus précis. Les enregistrements ordinaires et les diagnostics vont dans des fichiers distincts pour ne pas interrompre votre invite. Ces clients continuent à s'exécuter jusqu'aux commandes kill ciblées vers la fin du lab :
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=$!
Attendez l'affectation réelle des deux membres et enregistrez son rapport :
/home/labex/project/tools/wait-assignment 2 > /home/labex/project/output/two-members.txt
cat /home/labex/project/output/two-members.txt
L'outil en lecture seule consulte à plusieurs reprises le rapport des membres du groupe jusqu'à ce que les deux clients nommés possèdent des affectations non vides, sans chevauchement, dont l'union couvre les partitions 0 et 1. Son attente est limitée ; il ne fabrique aucune affectation et ne modifie pas la progression.
Dans le rapport, CLIENT-ID identifie reader-a ou reader-b, et CURRENT-ASSIGNMENT indique les partitions actuellement possédées. Par exemple, orders-shared:0 désigne la partition 0 de ce topic. Une affectation cible peut encore être en attente ; examinez donc la colonne actuelle. Chaque membre doit posséder une partition, sans exiger que reader-a reçoive un numéro de partition particulier. Un rééquilibrage redistribue les partitions lorsque des membres rejoignent ou quittent le groupe. Le groupe a besoin de temps pour atteindre sa nouvelle affectation ; le démarrage d'un processus ne signifie pas que cette répartition est déjà stable. Une affectation stable prouve le partage du travail ; deux identifiants de processus ne suffisent pas à établir les partitions qu'ils possèdent.
Publier du travail et comparer avec un groupe indépendant
Dans cette étape, vous allez fournir des enregistrements aux deux partitions affectées et observer un second groupe lire le même historique. Inspectez les lots consultables. Le lot initial contient share1 à share6 ; followup contient share7 et share8 :
cat /home/labex/project/data/group-batches.json
L'outil de publication fourni envoie le lot choisi au broker réel, attend les rappels de livraison et échoue si la livraison est incomplète. Publiez les six événements initiaux seulement après la stabilisation de l'affectation :
/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
L'outil d'attente en lecture seule attend que les deux partitions aient un offset suivant validé de 3, un offset de fin de 3 et un retard de 0. Chaque partition a reçu trois enregistrements. Examinez les observations des deux clients :
cat /home/labex/project/output/reader-a.jsonl
cat /home/labex/project/output/reader-b.jsonl
Ensemble, leurs fichiers doivent contenir les six identifiants initiaux. Le rapport d'affectation actif prouve la stabilité de la répartition ; un fichier peut aussi contenir des observations antérieures à un changement d'affectation.
Démarrez maintenant un groupe distinct à lecture limitée, orders-observer, depuis le début des données conservées :
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
Il doit recevoir les six enregistrements indépendamment. Il ne retire aucune partition à orders-team, car il appartient à un autre groupe. Les deux groupes peuvent avoir rattrapé leur retard en même temps tout en conservant des positions validées distinctes.
Arrêter un membre et observer l'affectation du membre restant
Dans cette étape, vous allez laisser reader-a en cours d'exécution et arrêter uniquement reader-b. kill envoie un signal de terminaison au processus dont l'identifiant a été enregistré ; il n'arrête pas le broker. wait permet au shell de terminer la récupération de l'état de ce processus enfant. Un processus ayant reçu un signal peut renvoyer un code de sortie non nul ; le || true final évite donc que cet arrêt volontaire soit considéré comme une défaillance applicative :
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 doit posséder les deux partitions une fois le groupe stabilisé. L'outil d'attente vérifie le rapport réel avant la publication suivante. Avec deux partitions, ajouter plus de deux membres laisserait certains membres sans partition dans une affectation stable ; cet exercice ne nécessite pas de troisième client.
Publiez un événement du lot followup dans chaque partition et attendez que le membre restant rattrape son retard :
/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
Le fichier de reader-a doit maintenant contenir share7 et share8. reader-b est arrêté et ne peut donc pas traiter ces nouveaux enregistrements. Le groupe d'observation conserve sa position validée précédente ; relancez-le pour lire les deux nouveaux événements :
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
Vous devez obtenir share7 et share8, avec un offset suivant de 4 propre à chaque groupe dans les deux partitions. L'option de départ au début ne remplace pas l'offset 3 valide enregistré pour le groupe d'observation.
Un rééquilibrage peut entraîner une nouvelle livraison si un membre sortant a traité des enregistrements avant la validation de sa progression. Les observations d'événements peuvent donc contenir des répétitions ; une affectation stable attribue néanmoins chaque partition à un seul propriétaire à la fois. Ce lab vérifie les affectations réelles et le contenu complet des événements sans prétendre garantir une livraison exactement une fois de bout en bout.
Fermer le client restant et conserver les observations
Dans cette étape, vous allez terminer cette expérience de longue durée en arrêtant le dernier client. Ses positions finales ont déjà été observées à la fin des données ; aucun nouveau travail n'est donc en attente. Utilisez son propre identifiant de processus enregistré :
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
Le groupe ne doit plus avoir de membre actif. Les offsets enregistrés et les événements du broker restent présents ; la durée de vie des clients est distincte de la progression stockée. Les clients d'observation à lecture limitée se sont déjà arrêtés automatiquement.
Inspectez les rapports des membres conservés avant et après la reprise des partitions :
cat /home/labex/project/output/two-members.txt /home/labex/project/output/one-member.txt
Le premier rapport montre deux propriétaires sans chevauchement, et le second un seul propriétaire des deux partitions. Ce sont des rapports réels dont les identifiants de membres dépendent de l'exécution, et non des affectations fixes à mémoriser. Les fichiers des lecteurs et du groupe d'observation conservent les événements observés par les deux groupes. Vous avez fermé uniquement vos deux clients en console tout en préservant le topic et les positions validées pour les examiner.
Résumé
Vous avez observé deux consommateurs se partager des affectations de partitions réelles, un groupe distinct lire les mêmes enregistrements indépendamment et un membre restant reprendre les deux partitions avant l'arrivée de nouveau travail. Des vérifications d'état à durée limitée ont remplacé les suppositions sur les délais, et des commandes d'arrêt ciblées ont fermé les clients tout en préservant la progression et les événements. Vous allez ensuite remplacer l'outil de publication fourni par un producteur Python dont vous pourrez lire et modifier la sérialisation et la gestion de la livraison.
