Introduction
Kafka conserve les enregistrements indépendamment des consommateurs qui les ont lus. Un groupe de consommateurs doit aussi mémoriser sa progression pour qu'un processus ultérieur puisse reprendre la lecture au lieu de recommencer. Un offset validé est la prochaine position que ce groupe doit lire dans une partition, et non le numéro du dernier enregistrement lu.
Vous allez lire deux des quatre enregistrements préparés, inspecter la progression validée et le retard, reprendre avec le même groupe, puis comparer avec un nouveau groupe. Vous prévisualiserez ensuite une réinitialisation, l'appliquerez pendant que le groupe d'origine est inactif, puis relirez les enregistrements conservés. Le topic possède une seule partition pour faciliter la compréhension des positions.
Ce lab indépendant fournit le broker, quatre enregistrements et un outil qui interroge l'état réel du coordinateur jusqu'à l'apparition de l'offset attendu. Des connaissances élémentaires du terminal suffisent ; vous n'avez ni code Python à écrire ni service à configurer. L'outil ne modifie pas les offsets et ne fabrique aucun rapport. Utilisez le terminal.
Lire deux enregistrements et inspecter la prochaine position validée
Dans cette étape, vous allez relier une lecture limitée à la position réellement enregistrée pour le groupe. Inspectez les quatre valeurs préparées :
cd /home/labex/project
cat /home/labex/project/data/progress-events.txt
Leurs offsets dans ce nouveau topic à une partition vont de 0 à 3. Un nouveau groupe nommé orders-review n'a aucune position enregistrée ; --from-beginning sélectionne donc le début des données conservées pour cette première lecture.
--command-property transmet une valeur de configuration au client consommateur, sans modifier le format d'affichage des événements. Dans cet exercice, le client en console valide automatiquement la progression. max.poll.records=1 limite chaque interrogation à un seul enregistrement renvoyé, de sorte que l'arrêt après deux enregistrements affichés n'avance pas sa position dans un lot renvoyé plus grand. L'intervalle définit la périodicité des validations ; à sa fermeture, le client termine aussi sa gestion normale de la progression. Vous allez inspecter le résultat réel sans supposer un délai précis :
kafka-console-consumer.sh --bootstrap-server localhost:9092 --topic orders-progress --group orders-review --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/first-two.txt
cat /home/labex/project/output/first-two.txt
Vous devez obtenir les deux événements created, puis le client se termine. Exécutez l'outil fourni d'interrogation à durée limitée avec le nom du groupe et l'offset suivant attendu :
/home/labex/project/tools/wait-group-offset orders-review 2 > /home/labex/project/output/after-first.txt
cat /home/labex/project/output/after-first.txt
L'outil décrit le groupe réel à plusieurs reprises jusqu'à ce que son offset validé soit 2, ou échoue à l'expiration de son délai. Il ne valide ni ne réinitialise rien. Repérez CURRENT-OFFSET 2, LOG-END-OFFSET 4 et LAG 2 pour la partition 0. Un message indiquant que le groupe n'a aucun membre actif est normal après l'arrêt du client à lecture limitée ; sa progression enregistrée existe toujours.
L'offset 2 signifie que la prochaine lecture commence au troisième enregistrement stocké. L'offset de fin 4 est la position qui suit les quatre enregistrements existants ; le retard 4 − 2 = 2 mesure les enregistrements restants dans cet exemple simple à l'état stable. Les deux premiers enregistrements n'ont pas été supprimés.
Reprendre avec le même groupe et comparer avec un nouveau groupe
Dans cette étape, vous allez prévoir les enregistrements qu'un groupe reçoit à partir de sa position enregistrée. Relancez le même groupe en conservant volontairement l'option de lecture depuis le début :
kafka-console-consumer.sh --bootstrap-server localhost:9092 --topic orders-progress --group orders-review --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/resumed-two.txt
cat /home/labex/project/output/resumed-two.txt
Vous devez obtenir les deux événements paid, pas les événements created précédents. --from-beginning fournit une politique de départ lorsqu'aucun offset validé valide n'existe ; cette option ne remplace pas l'offset 2 valide enregistré pour ce groupe. Confirmez que la lecture est terminée à partir de l'état réel du coordinateur :
/home/labex/project/tools/wait-group-offset orders-review 4 > /home/labex/project/output/after-resume.txt
cat /home/labex/project/output/after-resume.txt
Vous devez observer un offset validé de 4, un offset de fin de 4 et un retard de 0. Anticipez maintenant le résultat pour un autre groupe nommé orders-independent. Il n'a aucune position enregistrée et est indépendant de orders-review ; une lecture depuis le début doit donc renvoyer les quatre enregistrements conservés :
kafka-console-consumer.sh --bootstrap-server localhost:9092 --topic orders-progress --group orders-independent --from-beginning --max-messages 4 --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/new-group-all.txt
cat /home/labex/project/output/new-group-all.txt
/home/labex/project/tools/wait-group-offset orders-independent 4 > /home/labex/project/output/new-group-position.txt
cat /home/labex/project/output/new-group-position.txt
Vous devez obtenir les quatre valeurs et un offset validé de 4 propre à ce nouveau groupe. Les deux groupes peuvent atteindre la fin tout en ayant suivi des historiques de lecture différents. La progression d'un groupe n'a pas déplacé la position de départ de l'autre, et la consommation par ces groupes n'a pas supprimé les enregistrements du topic.
Vous allez maintenant modifier volontairement la position enregistrée du groupe d'origine, plutôt que de considérer l'option de lecture depuis le début comme une commande de retour en arrière.
Prévisualiser la réinitialisation d'un groupe inactif et relire l'historique
Dans cette étape, vous allez réinitialiser uniquement orders-review. Son consommateur à lecture limitée s'est terminé ; le groupe n'a donc aucun membre actif. Avant la réinitialisation, arrêtez tout membre de ce même groupe qui aurait été laissé en cours d'exécution par erreur ; un consommateur actif pourrait continuer à modifier la progression.
--reset-offsets sélectionne l'opération de gestion de la progression, --topic en limite la portée et --to-earliest sélectionne le début des données conservées. Commencez par utiliser --dry-run pour prévisualiser le nouvel offset proposé sans l'appliquer :
kafka-consumer-groups.sh --bootstrap-server localhost:9092 --group orders-review --reset-offsets --to-earliest --topic orders-progress --dry-run > /home/labex/project/output/reset-preview.txt
cat /home/labex/project/output/reset-preview.txt
Le NEW-OFFSET proposé doit être 0. Confirmez que la prévisualisation n'a pas modifié la position actuellement enregistrée :
/home/labex/project/tools/wait-group-offset orders-review 4 > /home/labex/project/output/after-preview.txt
cat /home/labex/project/output/after-preview.txt
Appliquez maintenant cette réinitialisation ciblée avec --execute :
kafka-consumer-groups.sh --bootstrap-server localhost:9092 --group orders-review --reset-offsets --to-earliest --topic orders-progress --execute
/home/labex/project/tools/wait-group-offset orders-review 0 > /home/labex/project/output/after-reset.txt
cat /home/labex/project/output/after-reset.txt
L'offset suivant réellement validé doit maintenant être 0, avec un offset de fin de 4 et un retard de 4. Les enregistrements sont restés en place ; seule la position enregistrée de ce groupe a changé. Relisez les quatre enregistrements en utilisant cette position de début désormais validée, sans recourir à une option de politique de départ :
kafka-console-consumer.sh --bootstrap-server localhost:9092 --topic orders-progress --group orders-review --max-messages 4 --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/replayed-all.txt
cat /home/labex/project/output/replayed-all.txt
/home/labex/project/tools/wait-group-offset orders-review 4 > /home/labex/project/output/after-replay.txt
cat /home/labex/project/output/after-replay.txt
Vous devez retrouver les quatre valeurs d'origine et un offset validé de 4. orders-independent doit toujours conserver sa propre position de fin de lecture ; la réinitialisation n'a pas fait revenir ce groupe en arrière.
La relecture consiste à lire volontairement des enregistrements existants une nouvelle fois. Un arrêt brutal entre le traitement d'un enregistrement et la validation de la progression peut aussi provoquer une nouvelle livraison de cet enregistrement. Cet exercice en console illustre les positions de lecture, pas une transaction atomique combinant écriture de fichier et validation, ni une garantie de traitement exactement une fois de bout en bout. Les prochains labs Python rendront explicite l'ordre du traitement et de la validation.
Résumé
Vous avez inspecté les offsets suivants validés et le retard, repris à partir de la progression enregistrée malgré une politique de départ au début, et comparé avec un groupe indépendant. La prévisualisation puis l'exécution d'une réinitialisation de groupe inactif ont permis une relecture volontaire sans modifier les enregistrements réels du topic. Vous allez ensuite observer comment plusieurs membres d'un même groupe se partagent les partitions et reprennent une affectation après le départ d'un membre.
