Introduction
Un studio a conservé son historique de votes tandis que son application d’affichage originale est suspendue. Vous devez produire un nouveau décompte sans revenir en arrière dans sa progression ni modifier le journal. Un sondage peut brièvement ne renvoyer aucun message alors qu’un nouveau consommateur n’a pas encore reçu le lot figé.
Vous construirez la boucle de rejeu native, capturerez séparément la position de fin de chaque partition et ne publierez qu’après lecture complète du lot fini. La fonction d’identité logique du laboratoire précédent est déjà fournie dans cette VM indépendante. Les bases de Python et Kafka utilisées précédemment sont nécessaires. Utilisez Terminal. L’estimation de 50–60 minutes de l’auteur inclut la lecture et les erreurs ordinaires ; il ne s’agit pas d’un essai chronométré auprès de débutants.
Il s’agit de reconstruire un lot figé. Aucun message supplémentaire n’est publié dans ces topics pendant l’exercice, vous n’avez pas de transaction à implémenter et aucune garantie de reprise fiable après plantage ou de traitement exactement une fois de bout en bout n’est annoncée.
Capturez la borne de chaque partition et lisez jusqu’à celle-ci
Dans cette étape, vous implémenterez une lecture native indépendante complète dans un programme préparé qui l’encadre. Commencez dans Terminal. Modifiez replay.py et, si nécessaire, reducer.py ou des auxiliaires Python sous helpers/ ; écrivez les fichiers dérivés sous output/. Préservez les outils/données fournis, les topics d’origine et les groupes des applications d’affichage suspendues :
cd /home/labex/project
bigdata-python -B vote_tool.py list
bigdata-python -B vote_tool.py describe main
La description en lecture seule présente les véritables ID de partition de studio-votes, le début retenu low, la fin figée actuelle et les répliques. Une partition est un journal ordonné, pas une VM distincte. La fin est la position suivante après le lot retenu, pas un enregistrement à compter. Prévoyez quelles partitions ont besoin d’enregistrements : une partition vide vérifie low == high et est déjà complète.
Un intervalle d’offsets n’est pas un nombre d’enregistrements. Kafka peut sauter des positions internes/de contrôle ou supprimées. Cette source comprend un véritable saut de position ; il n’y a pas de vote manquant à inventer. Inspectez les enveloppes réelles et comparez leurs positions effectives à la description :
bigdata-python -B vote_tool.py inspect main
Avant la reconstruction, lisez les positions enregistrées de l’application d’affichage suspendue avec l’outil des groupes de consommateurs utilisé dans le cours préalable :
kafka-consumer-groups.sh --bootstrap-server localhost:9092 --group studio-votes-display --describe
Gardez chaque CURRENT-OFFSET pour la comparaison ultérieure. L’absence de membres actifs est normale : l’application est suspendue mais a une progression enregistrée. Ici, LAG est une différence entre positions du journal, pas un nombre de votes logiques. Votre consommateur indépendant ne doit pas faire avancer la progression de ce groupe.
Lisez le programme préparé qui encadre le cœur
cat replay.py
Les imports, arguments(), la configuration du client, le signalement des erreurs et la fermeture dans finally sont déjà fournis. Le cœur lève volontairement NotImplementedError ; il ne lit pas les partitions et ne publie pas de réponse d’exemple. La fonction d’identité correcte du laboratoire précédent est préparée séparément dans reducer.py.
Le client utilise l’adresse bootstrap.servers sélectionnée et un group.id indépendant. allow.auto.create.topics vaut false. La validation automatique et le stockage automatique des offsets sont désactivés. assign lui donnera des positions de départ explicites sans attribution dynamique par un groupe ; cet exemple n’effectue aucun appel explicite de validation d’offsets. Une implémentation équivalente avec un groupe indépendant peut enregistrer sa propre progression, mais ne doit jamais rejoindre un groupe d’affichage original, y valider des offsets, le réinitialiser ou le supprimer.
Le vote_support.py préparé fournit arguments() pour le topic, le catalogue, la sortie, le broker et le groupe, ainsi que finish(...) pour l’écriture du résultat complet. Il ne découvre pas les partitions et ne récupère pas les votes. finish appelle la règle d’identité avant d’écrire ou remplacer le résultat. finally ferme le consommateur après réussite ou échec ; gardez cette structure extérieure.
Implémentez le cœur natif en courtes parties liées
Utilisez la méthode d’édition du laboratoire précédent :
nano replay.py
Gardez les imports, l’analyse des arguments, la configuration de Consumer et les sections try:, except et finally. Dans try:, supprimez uniquement les deux commentaires explicatifs et le raise NotImplementedError(...) du cœur incomplet. Utilisez les flèches pour vous placer et Retour arrière/Suppr pour modifier ; Ctrl+K peut couper toute la ligne courante : utilisez-le seulement lorsque vous êtes sur une ligne à remplacer.
Insérez les parties liées suivantes dans cet espace. Elles constituent un seul cœur : n’exécutez pas les blocs individuellement dans le shell et n’enregistrez/exécutez pas une boucle à moitié construite. Terminez toutes les parties, gardez le cadre fourni de gestion des erreurs et de fermeture, puis enregistrez et exécutez. Quatre espaces placent une instruction dans try: ; quatre de plus la placent dans une boucle ou une condition. L’indentation relative montrée fait partie du sens du Python.
Découvrez d’abord l’ensemble complet des véritables partitions. Appeler list_topics sans topic demandé évite de demander des métadonnées pour un nom inconnu. Vérifiez la disponibilité avant d’utiliser l’entrée sélectionnée. Trier les ID découverts donne à cet exemple une liste d’attribution stable ; cela ne crée pas d’ordre global des événements :
metadata = consumer.list_topics(timeout=10)
if args.topic not in metadata.topics or metadata.topics[args.topic].error:
raise ValueError("The selected topic is unavailable")
partitions = sorted(metadata.topics[args.topic].partitions)
Pour chaque partition découverte, demandez une seule fois get_watermark_offsets(..., timeout=10). TopicPartition identifie le topic et la partition pour cette demande. Gardez à la fois le début retenu low et la fin figée high, et initialisez la prochaine position de diagnostic à low. Ne rafraîchissez pas high dans la boucle de sondage : poursuivre une fin qui grandit ne serait plus un décompte fini. Cet exercice interdit d’ajouter à la source figée ou de la modifier :
bounds = {}
positions = {}
for number in partitions:
low, high = consumer.get_watermark_offsets(
TopicPartition(args.topic, number), timeout=10)
bounds[number] = [low, high]
positions[number] = low
Construisez pending uniquement à partir des partitions qui ont des positions retenues à lire. assign reçoit une liste d’offsets de départ explicites, y compris les low des partitions vides. Initialisez une nouvelle liste d’enregistrements physiques à chaque exécution. time.monotonic() mesure le temps écoulé indépendamment des corrections de l’horloge système ; l’échéance borne une lecture incomplète :
pending = {number for number in partitions
if bounds[number][0] < bounds[number][1]}
consumer.assign([TopicPartition(args.topic, number, bounds[number][0])
for number in partitions])
records = []
deadline = time.monotonic() + 30
Commencez maintenant la boucle. poll(0.2) peut renvoyer un message, un message porteur d’une erreur ou d’un événement, ou None à l’expiration de l’attente. Prévoyez pourquoi None ne peut pas signifier que c’est terminé : le client peut simplement ne pas encore avoir reçu un enregistrement figé. Continuez à attendre dans le délai global. Les événements EOF de partition sont activés dans la configuration ; _PARTITION_EOF est l’événement de fin attendu, tandis que les autres erreurs font échouer l’exécution. Cela permet une progression native au-delà d’une dernière position interne :
while pending:
if time.monotonic() >= deadline:
raise RuntimeError("Incomplete replay: frozen ends were not reached")
message = consumer.poll(0.2)
if message is not None:
if message.error():
if message.error().code() != KafkaError._PARTITION_EOF:
raise RuntimeError(str(message.error()))
Poursuivez le même bloc if message is not None avec le elif suivant, aligné avec if message.error(): à douze espaces. Seul un message normal dans une partition en attente, sous sa fin high fixe, devient une enveloppe. Conservez aussi les tentatives physiques ; summarize traitera ensuite l’identité logique. Décodez les octets UTF-8 de la clé et de la valeur ; json.loads transforme alors la valeur en dictionnaire :
elif (message.partition() in pending
and message.offset() < bounds[message.partition()][1]):
records.append({
"topic": message.topic(),
"partition": message.partition(),
"offset": message.offset(),
"key": message.key().decode("utf-8"),
"value": json.loads(message.value().decode("utf-8")),
})
Après avoir traité ce sondage, interrogez consumer.position(...) pour chaque partition restante. Il s’agit de la prochaine position du consommateur, distincte de la progression validée d’un groupe. Comparez chacune à son high capturé, pas à un nombre d’enregistrements supposé. Retirez uniquement les partitions terminées de pending. La ligne for position a huit espaces, dans while mais hors de la condition sur le message : même un sondage inactif peut donc être suivi d’une vérification des positions. Enfin, appelez finish à quatre espaces, après la fin de la boucle :
for position in consumer.position(
[TopicPartition(args.topic, number) for number in pending]):
positions[position.partition] = position.offset
if position.offset >= bounds[position.partition][1]:
pending.remove(position.partition)
finish(args, records, bounds, positions)
Chaque partition possède maintenant une borne fixe. Les partitions vides n’avaient besoin d’aucun vote ; les positions sautées n’exigeaient pas d’enregistrements inventés ; None n’a pas arrêté la lecture. Un instantané complet est publié uniquement une fois que toutes les prochaines positions en attente ont atteint leurs high.
Enregistrez avec Ctrl+O, Entrée pour confirmer replay.py, puis Ctrl+X. Inspectez le programme enregistré :
cat replay.py
Vérifiez que les six parties sont dans le try fourni, que le traitement des enveloppes reste dans la branche de message normal, que la vérification des positions reste dans la boucle et que finish est à l’extérieur. Exécutez immédiatement sur le lot natif sélectionné :
bigdata-python -B run_replay.py main
bigdata-python -B read_snapshot.py output/main-snapshot.json
Huit enregistrements physiques, cinq votes logiques et cedar=3, maple=2, willow=0 sont attendus. Le point d’entrée public vérifie d’abord la fonction d’identité sur des exemples jetables, puis enregistre les données du broker reçues par l’arbre natif réel du programme/des auxiliaires. Une comparaison métier complète distincte vérifie le résultat réel. Les bornes/positions déclarées dans l’instantané sont des diagnostics, pas une preuve indépendante de traitement natif.
Chaque appel réussi doit réellement écrire son résultat, directement ou via un fichier temporaire écrit puis renommé vers la sortie sélectionnée. Des octets identiques lors d’une répétition sont valides et les processus enfants auxiliaires peuvent écrire. Conserver un ancien fichier, ne mettre à jour que son horodatage ou écrire un marqueur de fin n’est pas une publication. Le finish fourni utilise la méthode enseignée de fichier temporaire/remplacement. Avec ce mécanisme préparé, une erreur de boucle ou de validation avant publication laisse l’instantané précédent disponible ; le point d’entrée n’annule pas des écritures arbitraires en place.
Pour une erreur de syntaxe, de nom ou d’indentation, utilisez la ligne signalée pour rouvrir nano replay.py, corrigez cette partie et préservez celles qui fonctionnent. Enregistrez et quittez, inspectez avec cat replay.py, puis relancez le point d’entrée public. Pour un dépassement du délai de rejeu incomplet, inspectez l’argument du topic sélectionné et la gestion de pending, d’EOF et des positions ; cela ne justifie pas de faire reculer le groupe original. Utilisez pwd et ls replay.py reducer.py pour retrouver un chemin mal saisi. Une erreur locale du programme ne nécessite ni de relancer la préparation de l’environnement ni de changer les enregistrements sources.
Reconstruisez depuis zéro et utilisez un autre lot
Dans cette étape, vous démontrerez qu’une nouvelle exécution reconstruit le même résultat fini, puis utiliserez un nouveau catalogue et une autre répartition des partitions. Restez dans Terminal, à /home/labex/project.
Le mécanisme d’écriture préparé trie les enveloppes physiques par partition/offset et les votes logiques par ID avant d’enregistrer le JSON. Un même instantané complet est ainsi stable, sans prétendre à un ordre global des événements. sha256sum calcule une empreinte du contenu. > enregistre cette empreinte dans votre répertoire de sortie ; -c compare ensuite le véritable fichier nommé à l’empreinte enregistrée :
sha256sum output/main-snapshot.json > output/main.sha256
bigdata-python -B run_replay.py main
sha256sum -c output/main.sha256
Vous devez voir output/main-snapshot.json: OK. À chaque appel, le programme repart de records, seen et de décomptes vides. Il remplace un instantané complet au lieu d’ajouter de nouveau les votes historiques aux anciens totaux. La consommation n’a ni supprimé ces enregistrements ni fait reculer l’application d’affichage suspendue.
Relisez les positions de l’application d’origine après cette répétition :
kafka-consumer-groups.sh --bootstrap-server localhost:9092 --group studio-votes-display --describe
Chaque CURRENT-OFFSET doit correspondre à la table de l’application suspendue relevée précédemment. Votre reconstruction indépendante n’a ni validé d’offsets dans ce groupe ni fait revenir celui-ci en arrière, bien qu’elle ait lu les enregistrements retenus. L’absence de membres actifs reste normale.
Affichez le lot check et examinez son propre catalogue de stands :
bigdata-python -B vote_tool.py list
cat data/check-catalog.json
Ce lot utilise d’autres noms de stands et trois partitions au lieu de quatre. Gardez les mêmes fichiers de programme. L’analyseur choisit son véritable topic et son catalogue, tandis que la découverte trouve l’ensemble complet de ses partitions réelles :
bigdata-python -B run_replay.py check
bigdata-python -B read_snapshot.py output/check-snapshot.json
Six enregistrements physiques, quatre votes logiques et amber=2, violet=2, moss=0 sont attendus. Coder en dur le premier catalogue ou s’arrêter au nombre d’enregistrements du premier topic ne décrirait pas ce lot.
Revenez au premier fichier réel :
bigdata-python -B read_snapshot.py output/main-snapshot.json
Il contient toujours le résultat du studio. Le lecteur ouvre directement le fichier demandé et ne complète rien depuis Kafka. S’il signale un fichier manquant, comparez le nom à vote_tool.py list et corrigez le chemin. Ne reconstruisez le lot correspondant que lorsque son résultat doit réellement être reconstruit.
Refusez un conflit en conservant l’instantané valide
Dans cette étape, vous observerez une erreur de données sans remplacer un résultat utile. Restez dans Terminal, à /home/labex/project. Le topic conflict répertorié contient volontairement un ID associé à des contenus valides de stand différents. Sa destination est l’instantané principal existant. Aucun gagnant n’est approuvé : cette exécution ne doit pas pouvoir publier un résultat comme valide.
L’empreinte de l’étape précédente décrit le fichier principal valide. Exécutez le même programme sur le lot conflictuel. La commande suivante doit échouer :
bigdata-python -B run_replay.py conflict
printf '%s\n' "$?"
Un code de sortie non nul et Conflicting event_id désignant le véritable ID sont attendus. La règle s’exécute avant que le mécanisme d’écriture préparé ne crée/remplace le résultat. Ici, le mécanisme .pending/remplacement publie un résultat complet ; il ne garantit ni reprise après plantage ni durabilité.
sha256sum -c output/main.sha256
bigdata-python -B read_snapshot.py output/main-snapshot.json
Le fichier principal valide doit toujours correspondre à l’empreinte, avec les décomptes originaux du studio. Le conflit échoué est enregistré séparément des exécutions réussies de main/check. Ne modifiez pas la source du conflit et ne choisissez pas l’événement arrivé en premier. Un parcours inverse ou séquentiel des partitions doit aussi refuser le même conflit.
Pour un simple nom de lot mal saisi, le point d’entrée affiche les noms disponibles avant de commencer le rejeu. Consultez vote_tool.py list, puis relancez avec le bon nom. Pour une erreur locale du programme, corrigez le fichier et relancez le lot valide concerné. Après toute modification du code source, relancez les deux lots valides pour que leurs résultats correspondent aux fichiers actuels, puis répétez l’autovérification du conflit.
Gardez suspendus les groupes des applications d’affichage originales : aucune adhésion, validation d’offsets, réinitialisation ou suppression n’est permise. Réinitialiser un groupe puis rétablir le même offset final ne constitue pas non plus un rejeu indépendant. Les vérifications fonctionnelles combinent réception native par l’arbre du programme et sorties réelles complètes ; elles ne prouvent ni la causalité du calcul ni la résistance à un administrateur qui falsifie l’environnement.
Résumé
Vous avez capturé de véritables bornes par partition et lu les journaux finis complets malgré des sondages temporairement inactifs, des partitions vides et des positions sautées. Le même programme natif a reconstruit deux catalogues sans cumuler d’anciens totaux, et un ID conflictuel a provoqué un échec tout en préservant l’instantané principal utile et la progression originale du consommateur.
