Introduction
Le consommateur qui réussissait écrivait un fichier, puis validait la progression de lecture. Ce sont deux opérations distinctes. Ce lab rend visibles deux cas de défaillance : un enregistrement JSON mal formé doit arrêter le traitement avant que la progression ne le dépasse, et un processus qui se termine après l'écriture mais avant la validation peut recevoir à nouveau le même enregistrement.
Vous allez d'abord établir une livraison normale, puis inspecter la position exacte d'un enregistrement invalide et enfin utiliser une option pédagogique fournie pour vous arrêter entre le vidage du tampon et la validation. Le redémarrage du même groupe révélera la nouvelle livraison dans le fichier grâce au même identifiant d'événement et à la même position Kafka. Vous observerez ce comportement sans implémenter de déduplication ni de mécanisme de nouvelle tentative.
Cet environnement indépendant prépare trois topics à une partition et le consommateur complet présenté dans le lab précédent. L'option supplémentaire de sortie contrôlée est déjà implémentée. Vous devez connaître les notions précédentes de JSON en Python, de fichiers et de validation manuelle ; aucune reconfiguration du broker ni nouvelle architecture de récupération n'est nécessaire. Utilisez le terminal.
Établir le déroulement d'un traitement réussi
Dans cette étape, vous allez confirmer que l'application fonctionne avec une entrée valide avant de diagnostiquer un défaut. Chaque topic possède une partition pour que les positions des enregistrements soient directes. orders-normal contient deux enregistrements valides. Le consommateur utilise la même séquence explicite qu'auparavant : écriture, vidage du tampon, validation synchrone du message et fermeture.
Inspectez l'application fournie, y compris son option de sortie contrôlée, sans la modifier :
cd /home/labex/project
cat /home/labex/project/consumer.py
L'option est désactivée par défaut. Exécutez le traitement normal de deux enregistrements avec son propre groupe et son propre fichier :
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
Vous devez obtenir normal1 et normal2 aux offsets 0 et 1, avec un offset suivant validé de 2 après les deux écritures. max-messages est le nombre demandé pour cette exécution ; le fichier ouvert en ajout peut conserver les observations de plusieurs exécutions dans les étapes suivantes.
Cette référence réussie distingue un problème de connexion affectant toute l'application d'un échec propre à un enregistrement. Vous utiliserez des topics, groupes et fichiers différents pour les échecs volontaires afin de conserver des observations claires.
S'arrêter sur un JSON mal formé sans valider au-delà
Dans cette étape, vous allez examiner un premier enregistrement valide, un enregistrement JSON invalide et un enregistrement valide ultérieur. Lisez les données locales :
cat /home/labex/project/data/bad-events.txt
Dans la partition 0 de orders-bad, l'enregistrement initial valide est à l'offset 0, le texte mal formé à l'offset 1 et l'événement valide ultérieur à l'offset 2. La politique est l'arrêt immédiat : signaler la position invalide et quitter. N'interceptez pas l'erreur pour poursuivre et valider ensuite des enregistrements ultérieurs de cette partition.
Demandez la livraison des trois enregistrements. La commande suivante doit échouer.
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
Vous devez obtenir un code non nul et Decode failure indiquant topic=orders-bad partition=0 offset=1. Le fichier ne doit contenir que bad-prefix à l'offset 0. Son écriture réussie a été validée : l'offset suivant du groupe est donc 1 ; l'enregistrement invalide et le suivant restent non traités.
Redémarrez le même groupe en ajoutant les données au même fichier. La commande suivante doit échouer.
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
Le même offset 1 doit à nouveau provoquer un échec. L'enregistrement initial valide ne doit pas être écrit une seconde fois, car sa position suivante 1 a déjà été validée. L'événement valide ultérieur ne doit pas apparaître dans le fichier. La politique devient ainsi visible : un opérateur doit choisir comment traiter les données invalides, tandis que cette application minimale ne les ignore pas silencieusement et ne valide pas au-delà.
Quitter après une écriture réussie mais avant sa validation
Dans cette étape, vous allez créer une autre fenêtre de défaillance à l'aide du topic orders-replay préparé. Il contient seulement repeat1 à l'offset 0. L'option pédagogique --exit-before-commit provoque une sortie avec le code 73 immédiatement après le vidage du tampon du fichier et avant la validation explicite. Elle indique la position réelle et ferme tout de même le consommateur, avec les validations automatiques désactivées.
Utilisez un nouveau groupe et un fichier distinct. La commande suivante doit échouer.
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
Vous devez obtenir le code 73 et une enveloppe enregistrée pour repeat1 à la partition 0, offset 0. Le journal indique CONTROLLED_EXIT_AFTER_WRITE_BEFORE_COMMIT. Il s'agit d'une démonstration de défaillance fournie, pas d'une modification à implémenter.
Le fichier contient maintenant l'enregistrement, alors que le groupe n'a aucune position validée pour celui-ci. Les règles de politique de départ étudiées précédemment s'appliquent donc lorsque ce même groupe redémarre. Cette séparation est précisément la fenêtre que vous allez observer à l'étape suivante.
Redémarrer et identifier l'enregistrement Kafka livré une nouvelle fois
Dans cette étape, vous allez redémarrer replay-reader sans l'option pédagogique. L'absence de position validée et la politique earliest lui permettent de relire l'enregistrement conservé. Gardez le mode ajout et le même fichier de sortie afin que les deux livraisons restent 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
Vous devez obtenir deux enveloppes avec les mêmes event_id, clé, valeur, topic, partition et offset. Il n'existe toujours qu'un seul enregistrement original sur le broker. La seconde exécution valide avec succès l'offset suivant 1, tandis que le fichier conserve les traces des deux livraisons.
Utilisez les observations suivantes pour distinguer des situations courantes :
| Observation | Premiers éléments utiles à examiner |
|---|---|
| Aucun nouvel événement | Topic valide et progression du groupe à la fin ; une interrogation à durée limitée peut ne renvoyer aucun enregistrement |
| Nom de topic erroné | Comparez le nom fourni à la liste ou à la description des topics ; ce client désactive la création accidentelle de topics |
| Aucune position validée valide | Inspectez la progression du groupe et la politique de départ earliest configurée ; elle peut lire l'historique conservé |
| Échec de décodage | Lisez le topic, la partition et l'offset signalés ainsi que les données de valeur ; le groupe s'arrête avant cette position invalide |
Écrire un fichier et valider une position Kafka ne constituent pas une transaction unique. Une interruption entre ces opérations peut provoquer une nouvelle livraison, comme le montre cette position réelle d'enregistrement. Cet exercice conserve les deux livraisons sans implémenter de déduplication, de file d'événements rejetés ni de conception transactionnelle. Le défi final applique le déroulement réussi avec une procédure explicite de nouvelle tentative à partir d'un état propre.
Résumé
Vous avez établi une livraison normale, arrêté le traitement d'un JSON mal formé sans valider au-delà et observé un enregistrement écrit, dont le tampon avait été vidé mais qui n'avait pas été validé, livré à nouveau après redémarrage. La correspondance du contenu de l'événement et de sa position Kafka explique les observations en double dans le fichier sans inventer un second événement sur le broker. Le défi final combine maintenant publication réussie, consommation à durée limitée, sortie vers un fichier et écriture avant validation sur de nouvelles données.
