Introduction
Le producteur en console a permis de comprendre le modèle des événements et des clés. Une application doit faire une distinction supplémentaire : appeler une API d'envoi peut placer un enregistrement dans une file locale du client avant que le broker le confirme. Un programme qui se termine à ce stade peut annoncer une réussite sans avoir livré ses enregistrements.
Vous allez utiliser le client Python confluent-kafka installé avec le broker Apache Kafka local préparé. Ce paquet est une bibliothèque cliente ; aucun compte Confluent Cloud n'est nécessaire. Vous sérialiserez des dictionnaires en octets JSON/UTF-8, enverrez des clés de commande et examinerez les rappels de livraison ainsi que l'état final de la file. Une adresse locale inaccessible permettra de tester un échec dans un délai limité.
Ce lab indépendant prépare un topic à trois partitions et deux fichiers de données d'entrée. La suite du cours suppose que vous connaissez les fonctions Python, les boucles, les dictionnaires, JSON et l'exécution de scripts ; le parcours Python peut vous aider à acquérir ces prérequis. L'installation de Kafka et la configuration des services sont déjà préparées. Utilisez le terminal.
Transformer un dictionnaire en clé et en valeur à transmettre
Dans cette étape, vous allez examiner la conversion avant de contacter Kafka. Commencez par l'entrée de six événements :
cd /home/labex/project
cat /home/labex/project/data/events.jsonl
Chaque ligne est un objet JSON complet. json.loads convertit ce texte en dictionnaire Python. json.dumps convertit le dictionnaire en texte JSON, puis encode("utf-8") convertit ce texte en octets adaptés à l'API cliente. La clé est encodée séparément à partir de order_id ; ce n'est pas event_id.
Enregistrez et exécutez ce petit programme de prévisualisation. Le délimiteur entre guillemets du document intégré préserve littéralement le code Python :
cat > /home/labex/project/preview_event.py <<'PYTHON'
import json
from pathlib import Path
event = json.loads(Path("/home/labex/project/data/events.jsonl").read_text().splitlines()[0])
value_text = json.dumps(event, ensure_ascii=False)
value_bytes = value_text.encode("utf-8")
key_bytes = event["order_id"].encode("utf-8")
print("Event dict:", event)
print("JSON text:", value_text)
print("UTF-8 value:", value_bytes)
print("UTF-8 key:", key_bytes)
PYTHON
bigdata-python /home/labex/project/preview_event.py > /home/labex/project/output/encoding-preview.txt
cat /home/labex/project/output/encoding-preview.txt
Repérez le dictionnaire, le texte JSON et les représentations en octets. L'affichage d'un objet bytes Python commence par b ; cette notation d'affichage ne fait pas partie de la valeur transmise. La première clé doit encoder order-4001. JSON conserve les champs nommés de l'événement, tandis qu'UTF-8 définit l'encodage des caractères.
Aucun événement n'a encore été envoyé. L'étape suivante fournit ces octets à un producteur et vérifie leur confirmation par le broker.
Publier et confirmer chaque livraison réelle
Dans cette étape, vous allez exécuter un producteur complet dont la gestion des arguments et le rappel de livraison sont fournis. Producer configure le client ; bootstrap.servers indique l'adresse initiale. acks=all demande un accusé de réception des réplicas actuellement synchronisés. Cet environnement à un broker possède un réplica ; ce paramètre ne démontre donc pas la tolérance aux pannes entre plusieurs brokers. Le paramètre de routage murmur2_random fourni maintient la compatibilité du routage avec clés avec les exemples précédents en console Java, tant que le nombre de partitions reste fixe.
Repérez la distinction essentielle dans le programme : produce place un enregistrement dans la file locale. poll exécute les rappels disponibles, et flush attend les envois en file tout en exécutant aussi les rappels.
Un rappel est une fonction que vous fournissez à la bibliothèque cliente pour qu'elle l'appelle lorsqu'un résultat de livraison est disponible. Passer callback=delivery fournit la fonction elle-même ; cela ne l'appelle pas immédiatement. poll(0) exécute les rappels disponibles sans attendre, tandis que flush attend aussi les envois en cours. Le rappel enregistre soit le topic, la partition et l'offset confirmés par le broker, soit un échec. Après flush, le programme vérifie les enregistrements restants, les échecs et le nombre de confirmations avant d'annoncer la réussite. Un rappel d'échec est enregistré et entraîne un code de sortie non nul ; il n'est pas simplement affiché puis ignoré.
La section argparse fournie associe les options de ligne de commande aux attributs de args : par exemple, --input devient args.input, et --delivery-timeout-ms devient args.delivery_timeout_ms. Path(args.input).open(...) ouvre le fichier local choisi ; la boucle lit un événement JSON par ligne non vide. Vous n'avez pas besoin de concevoir un analyseur d'arguments pour utiliser ce client.
Enregistrez le programme complet. La structure de l'analyseur et du rappel est fournie, tandis que la sérialisation et l'appel à produce utilisent les octets que vous venez d'examiner :
cat > /home/labex/project/producer.py <<'PYTHON'
import argparse,json,sys
from pathlib import Path
from confluent_kafka import Producer
parser = argparse.ArgumentParser(description="Publish a finite JSONL event file and confirm deliveries.")
parser.add_argument("--input", required=True)
parser.add_argument("--topic", required=True)
parser.add_argument("--bootstrap-server", default="localhost:9092")
parser.add_argument("--delivery-timeout-ms", type=int, default=15000)
args = parser.parse_args()
if args.delivery_timeout_ms <= 0:
parser.error("The delivery timeout must be positive.")
producer = Producer({
"bootstrap.servers": args.bootstrap_server,
"acks": "all",
"message.timeout.ms": args.delivery_timeout_ms,
"partitioner": "murmur2_random",
})
delivered = []
failures = []
def delivery(error, message):
event_id = json.loads(message.value().decode("utf-8"))["event_id"]
if error is not None:
failures.append(str(error))
print(f"DELIVERY_FAILED event_id={event_id} error={error}", file=sys.stderr)
else:
delivered.append(event_id)
print(f"DELIVERED event_id={event_id} topic={message.topic()} partition={message.partition()} offset={message.offset()}")
sent = 0
with Path(args.input).open(encoding="utf-8") as source:
for line in source:
if not line.strip():
continue
event = json.loads(line)
key = event["order_id"].encode("utf-8")
value = json.dumps(event, ensure_ascii=False).encode("utf-8")
producer.produce(args.topic, key=key, value=value, callback=delivery)
sent += 1
producer.poll(0)
remaining = producer.flush(args.delivery_timeout_ms / 1000 + 5)
if remaining or failures or len(delivered) != sent:
print(f"Delivery incomplete: queued={remaining}, failures={len(failures)}, confirmed={len(delivered)}, requested={sent}", file=sys.stderr)
raise SystemExit(1)
print("Delivered records:", len(delivered))
PYTHON
Exécutez le producteur et enregistrez les deux flux de sortie. > redirige la sortie standard vers le journal ; 2>&1 envoie la sortie d'erreur standard vers la même destination, afin de conserver aussi les échecs de rappel et les diagnostics de connexion.
bigdata-python /home/labex/project/producer.py --input /home/labex/project/data/events.jsonl --topic orders-python > /home/labex/project/output/producer-deliveries.log 2>&1
cat /home/labex/project/output/producer-deliveries.log
Vous devez obtenir six lignes de rappel DELIVERED et Delivered records: 6. La partition et l'offset réels de chaque rappel peuvent varier. Le simple retour réussi de produce ne suffirait pas à établir ces confirmations.
Inspectez indépendamment le broker avec le consommateur en console déjà étudié :
kafka-console-consumer.sh --bootstrap-server localhost:9092 --topic orders-python --from-beginning --max-messages 6 --timeout-ms 30000 --formatter-property print.key=true --formatter-property print.partition=true --formatter-property print.offset=true > /home/labex/project/output/python-first-observed.txt
cat /home/labex/project/output/python-first-observed.txt
Repérez py1 à py6 avec des clés correspondant à leurs valeurs order_id. Cette lecture indépendante relie le rapport de livraison de l'application aux enregistrements réels du broker ; aucun numéro de partition fixe ni ordre global d'affichage n'est exigé.
Modifier un champ d'un nouvel événement et réutiliser le producteur
Dans cette étape, vous allez réutiliser la même application pour un nouvel événement. Le fichier next-event préparé possède un nouvel event_id, py7, qui n'a donc pas encore été publié. Inspectez son champ channel :
cat /home/labex/project/data/next-event.jsonl
Dans un fichier de travail distinct, remplacez uniquement la valeur online du champ channel de cet événement non publié par store. L'événement fait toujours référence à order-4001 et suit ses événements précédents avec sequence 3 :
cat > /home/labex/project/data/changed-event.jsonl <<'EVENT'
{"event_id":"py7","order_id":"order-4001","sequence":3,"event_type":"shipped","channel":"store"}
EVENT
bigdata-python /home/labex/project/producer.py --input /home/labex/project/data/changed-event.jsonl --topic orders-python > /home/labex/project/output/changed-delivery.log 2>&1
cat /home/labex/project/output/changed-delivery.log
kafka-console-consumer.sh --bootstrap-server localhost:9092 --topic orders-python --from-beginning --max-messages 7 --timeout-ms 30000 --formatter-property print.key=true --formatter-property print.partition=true --formatter-property print.offset=true > /home/labex/project/output/python-final-observed.txt
cat /home/labex/project/output/python-final-observed.txt
Vous devez obtenir un nouveau rappel de réussite et sept événements conservés dans la lecture indépendante. py7 doit avoir channel store, tandis que py1 à py6 gardent leur contenu d'origine. Les événements liés à order-4001 suivent leur séquence dans la même partition de routage ; l'ordre d'affichage entre partitions peut varier.
L'implémentation du producteur n'a pas changé. Le chemin d'entrée fourni lui permet de sérialiser et de livrer les nouvelles données au lieu d'intégrer une réponse d'exercice fixe.
Observer un échec à durée limitée sur une adresse inaccessible
Dans cette étape, vous allez vérifier que la mise en file n'est pas confondue avec la livraison. Le même client peut créer une file locale même si localhost:1 n'héberge aucun broker Kafka. L'option de délai de livraison fournie limite l'achèvement des messages à 3000 millisecondes, et flush ajoute une petite marge limitée. Le broker réel reste à localhost:9092.
Envoyez le fichier d'un événement à cette adresse volontairement erronée. La commande suivante doit échouer.
bigdata-python /home/labex/project/producer.py --input /home/labex/project/data/changed-event.jsonl --topic orders-python --bootstrap-server localhost:1 --delivery-timeout-ms 3000 > /home/labex/project/output/unreachable-producer.log 2>&1
Capturez immédiatement le code de sortie avant qu'une autre commande ne le remplace. $? est le code de sortie de la commande précédente : zéro pour une réussite, une valeur non nulle pour un échec. printf l'enregistre sous forme d'une ligne de texte :
printf '%s\n' "$?" > /home/labex/project/output/unreachable-exit-status.txt
cat /home/labex/project/output/unreachable-exit-status.txt
cat /home/labex/project/output/unreachable-producer.log
Vous devez observer un code non nul, des diagnostics de connexion, un rappel DELIVERY_FAILED et Delivery incomplete au lieu d'un décompte final de réussite. Le programme a vérifié les échecs des rappels même lorsque la file s'est entièrement vidée après l'échec des messages.
L'adresse défaillante ne doit pas avoir ajouté un second py7 au topic réel. Les sept événements livrés avec succès restent disponibles. Corriger une adresse et retenter un événement métier est une décision distincte de l'hypothèse selon laquelle un enregistrement en file a été confirmé ; l'échec volontaire de ce lab n'a rien envoyé au broker réel. Vous allez ensuite décoder le JSON reçu, l'enregistrer et valider la progression de lecture dans un consommateur Python.
Résumé
Vous avez sérialisé des dictionnaires et des clés de commande en octets UTF-8, publié avec un véritable client Python et rapproché les rappels de réussite des positions réelles sur le broker. Un nouvel événement modifié a réutilisé le même programme, tandis qu'une adresse inaccessible a produit un échec à durée limitée avec code non nul au lieu d'une fausse réussite. Le prochain consommateur Python rendra explicites le décodage, la sortie vers un fichier et l'écriture avant validation.
