Introduction
Un consommateur qui affiche un événement n'a pas nécessairement enregistré un résultat utile à transmettre. Cette application décode chaque valeur, conserve sa clé et sa position Kafka dans un fichier JSONL local, et ne valide la progression qu'après la réussite de l'écriture correspondante et le vidage de son tampon.
Vous allez transposer en Python le modèle des groupes et des offsets déjà étudié : subscribe choisit le topic, poll demande des enregistrements et une validation synchrone explicite enregistre l'offset suivant du message traité. La validation automatique et le stockage automatique des offsets sont désactivés pour qu'ils n'avancent pas la progression indépendamment de votre code. L'écriture du fichier et les validations Kafka restent des opérations distinctes ; elles ne forment pas une transaction atomique entre systèmes.
Ce lab indépendant prépare dans un nouveau topic les six événements de commande initiaux que vous connaissez depuis le lab sur le producteur. Aucun état d'une VM précédente n'est nécessaire. Vous devez connaître les fonctions Python, les boucles, les dictionnaires, JSON et l'exécution de scripts. La structure de l'analyseur d'arguments et du délai est fournie. Utilisez le terminal et l'environnement bigdata-python installé.
S'abonner, interroger le broker et décoder un enregistrement réel
Dans cette étape, vous allez examiner un enregistrement reçu sans valider la progression de cette prévisualisation. Commencez par lire les valeurs des événements préparés :
cd /home/labex/project
cat /home/labex/project/data/events.jsonl
Un Consumer est configuré avec une identité de groupe et une politique de départ earliest. subscribe enregistre le topic souhaité ; poll renvoie un message, un message portant une erreur ou None lorsqu'aucun enregistrement n'est disponible, notamment pendant l'affectation. La boucle continue à interroger le broker dans la limite d'un délai global.
La structure de contrôle fournie limite la durée de cette lecture. time.monotonic() mesure le temps écoulé sans dépendre de la date de l'horloge système. continue lance l'interrogation suivante lorsqu'aucun message n'est prêt ; break quitte la boucle après l'affichage d'un enregistrement. Le else de la boucle while ne s'exécute que si la boucle se termine sans ce break, ce qui signifie ici qu'aucun enregistrement n'est arrivé avant l'échéance.
finally ferme le consommateur, que la lecture réussisse ou lève une erreur. Considérez cette structure comme l'attente et le nettoyage autour des appels Kafka ; vous n'avez pas à concevoir de mécanisme de nouvelle tentative.
Cette prévisualisation utilise un groupe distinct, file-preview, avec les deux paramètres automatiques désactivés. La valeur décodée est un dictionnaire Python ; son enveloppe conserve aussi le topic, la partition, l'offset et une clé décodée. Aucun appel à commit n'est effectué :
cat > /home/labex/project/preview_consumer.py <<'PYTHON'
import json,time
from confluent_kafka import Consumer
consumer = Consumer({"bootstrap.servers":"localhost:9092", "group.id":"file-preview", "auto.offset.reset":"earliest", "enable.auto.commit":False, "enable.auto.offset.store":False, "allow.auto.create.topics":False})
deadline = time.monotonic() + 20
try:
consumer.subscribe(["orders-file"])
while time.monotonic() < deadline:
message = consumer.poll(1)
if message is None:
continue
if message.error() is not None:
raise RuntimeError(str(message.error()))
value = json.loads(message.value().decode("utf-8"))
record = {"topic":message.topic(), "partition":message.partition(), "offset":message.offset(), "key":message.key().decode("utf-8"), "value":value}
print(json.dumps(record, ensure_ascii=False))
break
else:
raise SystemExit("No preview record arrived before the deadline.")
finally:
consumer.close()
PYTHON
bigdata-python /home/labex/project/preview_consumer.py > /home/labex/project/output/one-event.json
cat /home/labex/project/output/one-event.json
Vous devez obtenir une enveloppe JSON dont la valeur contient l'un des événements py1 à py6 et dont la clé correspond au champ order_id de cet événement. Avec trois partitions, le premier événement observé n'est pas fixe. La partition et l'offset identifient son enregistrement réel sur le broker.
La fermeture de cette prévisualisation ne valide pas la progression, car les validations automatiques sont désactivées et le script n'a effectué aucune validation explicite. L'application suivante utilise son propre groupe pour livrer les données dans un fichier.
Écrire et vider le tampon de chaque enveloppe avant de valider son message
Dans cette étape, vous allez construire le consommateur vers fichier à durée limitée. Examinez attentivement son ordre de traitement : décoder et valider l'objet JSON et la clé, écrire une enveloppe JSONL, vider le tampon du fichier, valider ce message de manière synchrone, puis le compter comme terminé. Si une écriture, un décodage ou une validation échoue, l'application se termine au lieu d'ignorer l'erreur puis de prétendre que le lot a réussi.
L'instruction with ferme le fichier local à la sortie de son bloc. Path identifie le nom du fichier local. La sortie est ouverte en mode ajout pour qu'un redémarrage ne tronque pas silencieusement les données de livraison précédentes. La première exécution dans un environnement vierge commence sans fichier. max-messages est l'objectif de cette exécution ; le délai global provoque une sortie non nulle si elle reste incomplète. JSONL stocke un objet JSON complet par ligne, sans tableau englobant. flush() transfère au système d'exploitation les données du fichier mises en tampon par Python avant la demande de validation. Cela ne garantit pas leur survie à une panne de courant ou de disque, et ne transforme pas le fichier et la validation Kafka en transaction atomique. Cet exercice observe l'interruption d'un processus et la progression enregistrée dans la VM en cours d'exécution.
commit(message=..., asynchronous=False) valide l'offset suivant du message traité pour sa partition. Les erreurs renvoyées pour chaque partition sont vérifiées. enable.auto.offset.store=false empêche le stockage automatique indépendant de la progression, tandis que l'argument message explicite identifie précisément la position traitée à valider.
Enregistrez la structure complète fournie de l'analyseur et de la boucle avec cette séquence de traitement :
cat > /home/labex/project/consumer.py <<'PYTHON'
import argparse,json,sys,time
from pathlib import Path
from confluent_kafka import Consumer
parser = argparse.ArgumentParser(description="Save a finite JSON event delivery and commit only written records.")
parser.add_argument("--topic", required=True)
parser.add_argument("--group", required=True)
parser.add_argument("--output", required=True)
parser.add_argument("--max-messages", type=int, default=6)
parser.add_argument("--timeout-seconds", type=float, default=30)
parser.add_argument("--bootstrap-server", default="localhost:9092")
args = parser.parse_args()
if args.max_messages <= 0 or args.timeout_seconds <= 0:
parser.error("Message count and timeout must be positive.")
consumer = Consumer({
"bootstrap.servers": args.bootstrap_server,
"group.id": args.group,
"auto.offset.reset": "earliest",
"enable.auto.commit": False,
"enable.auto.offset.store": False,
"allow.auto.create.topics": False,
})
received = 0
deadline = time.monotonic() + args.timeout_seconds
try:
consumer.subscribe([args.topic])
with Path(args.output).open("a", encoding="utf-8") as sink:
while received < args.max_messages and time.monotonic() < deadline:
message = consumer.poll(1)
if message is None:
continue
position = f"topic={message.topic()} partition={message.partition()} offset={message.offset()}"
if message.error() is not None:
raise RuntimeError(f"Kafka error {position}: {message.error()}")
try:
key = message.key().decode("utf-8") if message.key() is not None else None
value = json.loads(message.value().decode("utf-8"))
if not isinstance(value, dict) or not isinstance(value.get("event_id"), str) or not isinstance(value.get("order_id"), str):
raise ValueError("Expected an object with string event_id and order_id.")
if key != value["order_id"]:
raise ValueError("Record key does not match order_id.")
except (UnicodeDecodeError, json.JSONDecodeError, ValueError) as error:
print(f"Decode failure {position}: {error}", file=sys.stderr)
raise SystemExit(1)
record = {"topic": message.topic(), "partition": message.partition(), "offset": message.offset(), "key": key, "value": value}
sink.write(json.dumps(record, ensure_ascii=False) + "\n")
sink.flush()
committed = consumer.commit(message=message, asynchronous=False)
if not committed or any(item.error is not None for item in committed):
raise RuntimeError(f"Commit failed {position}: {committed}")
received += 1
print(f"SAVED_AND_COMMITTED event_id={value['event_id']} {position} next_offset={message.offset()+1}")
if received < args.max_messages:
print(f"Only received {received} of {args.max_messages} requested records before the deadline.", file=sys.stderr)
raise SystemExit(2)
print("Saved and committed records:", received)
finally:
consumer.close()
PYTHON
bigdata-python /home/labex/project/consumer.py --topic orders-file --group file-writer --output /home/labex/project/output/received-events.jsonl --max-messages 6 --timeout-seconds 30 > /home/labex/project/output/consumer.log 2>&1
cat /home/labex/project/output/consumer.log
cat /home/labex/project/output/received-events.jsonl
Vous devez obtenir Saved and committed records: 6 et six enveloppes JSONL. Chaque enveloppe doit préserver la valeur complète, la clé correspondante et la position réelle de l'enregistrement. Un rappel ou un journal en console ne remplace pas le fichier enregistré ; l'étape suivante compare les données du fichier à la progression réellement validée.
Relier le fichier aux offsets suivants validés par partition
Dans cette étape, vous allez inspecter la progression du coordinateur avec l'outil en ligne de commande déjà connu :
kafka-consumer-groups.sh --bootstrap-server localhost:9092 --group file-writer --describe > /home/labex/project/output/file-writer-position.txt
cat /home/labex/project/output/file-writer-position.txt
Pour chaque partition contenant des enregistrements livrés, CURRENT-OFFSET doit atteindre le LOG-END-OFFSET de cette partition, avec un retard nul. Cette leçon ne fixe ni les affectations exactes ni la répartition entre partitions. Une partition sans enregistrement peut ne pas encore avoir de position validée ; aucun message n'y a été traité pour être validé.
Comparez l'offset d'une enveloppe enregistrée à l'offset suivant validé de sa partition. Si le dernier enregistrement traité d'une partition se trouve à l'offset n, la position suivante validée est n+1. Cette position suivante n'est pas un décompte global sur toutes les partitions.
Le consommateur est fermé ; un message indiquant l'absence de membres actifs est donc normal. Le fichier stocké et la progression enregistrée du groupe restent présents. Vous disposez d'observations distinctes de la sortie du traitement et de l'état du coordinateur, sans prétendre qu'ils ont été écrits en une seule transaction.
Redémarrer le même groupe et constater l'absence de nouvelle livraison
Dans cette étape, vous allez tester la position enregistrée au lieu de supposer que earliest fait toujours revenir en arrière. Les six enregistrements sont toujours conservés, mais file-writer possède des positions de fin validées valides. Anticipez un nouveau démarrage avec ce même groupe : il ne doit trouver aucun nouvel enregistrement, même si auto.offset.reset vaut earliest.
Utilisez un chemin de sortie distinct pour conserver le fichier original de six enregistrements. Demandez un enregistrement avec un court délai global. Le nombre demandé ne pouvant pas être atteint, cette application signale volontairement un lot incomplet par un code non nul. La commande suivante doit échouer.
bigdata-python /home/labex/project/consumer.py --topic orders-file --group file-writer --output /home/labex/project/output/no-new-events.jsonl --max-messages 1 --timeout-seconds 5 > /home/labex/project/output/same-group-restart.log 2>&1
Capturez immédiatement son code de sortie :
printf '%s\n' "$?" > /home/labex/project/output/same-group-exit-status.txt
cat /home/labex/project/output/same-group-exit-status.txt
cat /home/labex/project/output/same-group-restart.log
Vous devez obtenir un code non nul et Only received 0 of 1 requested records before the deadline. Le nouveau fichier doit être vide ; le fichier original received-events.jsonl contient toujours six enveloppes. Les offsets validés valides du même groupe n'ont pas été remplacés par sa politique de départ earliest.
Pour relire volontairement l'historique conservé, utilisez la procédure déjà étudiée d'un groupe distinct ou de réinitialisation d'un groupe inactif. Si un processus s'arrête après l'écriture mais avant la validation, la situation est différente : le même groupe peut recevoir à nouveau l'enregistrement. Le prochain lab rendra cette fenêtre de défaillance visible.
Résumé
Vous vous êtes abonné et avez interrogé Kafka avec Python, décodé des enregistrements JSON/UTF-8 et conservé les clés et les positions Kafka en JSONL. Le vidage réussi des tampons a précédé les validations synchrones explicites, et l'état réel du coordinateur correspondait au fichier terminé. Un redémarrage du même groupe n'a trouvé aucun nouvel enregistrement sans supprimer l'historique conservé du topic. Vous allez ensuite observer comment un échec de décodage et une interruption entre écriture et validation peuvent entraîner une nouvelle livraison des enregistrements.
