Einführung
Eine Gruppe kann die Partitionen eines Topics auf mehrere Consumer verteilen. Innerhalb einer stabilen Gruppe hat jede Partition genau einen aktiven Besitzer, während eine andere Gruppe dieselben gespeicherten Datensätze unabhängig lesen kann. Dies ermöglicht eine Arbeitsaufteilung, ohne dass alle Anwendungen um eine einzige globale Leseposition konkurrieren müssen.
Sie starten zwei dauerhaft laufende Konsolenclients in einer Gruppe, untersuchen vor der Veröffentlichung ihre tatsächlichen Zuweisungen und vergleichen die Beobachtungen einer separaten Gruppe. Nachdem ein Mitglied ausgeschieden ist, warten Sie darauf, dass das verbleibende Mitglied beide Partitionen besitzt, bevor Sie einen weiteren kleinen Stapel senden. Das vorbereitete Veröffentlichungswerkzeug adressiert beide Partitionen explizit, sodass jede während des Experiments Arbeit erhält. Seine tatsächliche Zustellung an den Broker wird geprüft.
Dieses unabhängige Lab stellt ein Topic mit zwei Partitionen, einsehbare Ereignis-Testdaten und begrenzte, ausschließlich lesende Wartewerkzeuge bereit. Grundlegende Terminalkenntnisse reichen aus; für die Testdatenwerkzeuge müssen Sie keinen Python-Code bearbeiten. Die Konsolenclients laufen im Hintergrund, bis Sie sie in den letzten Schritten beenden. Behalten Sie dieselbe Terminal-Shell bei, damit ihre Prozessvariablen verfügbar bleiben.
Starten Sie zwei Mitglieder und warten Sie auf eine stabile Zuweisung
In diesem Schritt beobachten Sie die Zuweisung, bevor Sie der Gruppe Arbeit geben. Das vorbereitete Topic hat zwei Partitionen und keine Datensätze. Untersuchen Sie seine Metadaten:
cd /home/labex/project
kafka-topics.sh --bootstrap-server localhost:9092 --describe --topic orders-shared
Die folgenden Clients verwenden dieselbe Gruppe orders-team. --command-property übergibt jeden Konfigurationswert an den Consumer-Client. client.id kennzeichnet die Clients in Berichten als reader-a und reader-b. Mit den vertrauten Einstellungen zum automatischen Festschreiben kann ihr Lesefortschritt gespeichert werden, während sie weiter Daten abfragen.
Das abschließende & startet einen Client im Hintergrund der Shell, sodass Sie im selben Terminal Untersuchungsbefehle ausführen können. $! ist die Prozess-ID des neu gestarteten Hintergrundprozesses. Die jeweils unmittelbar folgende Zuweisung speichert sie für das spätere gezielte Beenden.
Die Shell führt zwei Ausgabeströme: die Standardausgabe für Datensätze und die Standardfehlerausgabe für Diagnosen. > schreibt den ersten Strom in die JSONL-Datei; 2> schreibt den Fehlerstrom in das Protokoll. Die Zuweisung consumer_a_pid=$! speichert eine Shellvariable ohne Leerzeichen um =. Mit "$consumer_a_pid" rufen Sie später genau diese Prozess-ID ab. Reguläre Datensätze und Diagnosen gelangen in getrennte Dateien, damit sie Ihre Eingabeaufforderung nicht unterbrechen. Diese Clients laufen bis zu den gezielten kill-Befehlen gegen Ende weiter:
kafka-console-consumer.sh --bootstrap-server localhost:9092 --topic orders-shared --group orders-team --from-beginning --command-property client.id=reader-a --command-property enable.auto.commit=true --command-property auto.commit.interval.ms=1000 --command-property max.poll.records=1 > /home/labex/project/output/reader-a.jsonl 2> /home/labex/project/output/reader-a.log &
consumer_a_pid=$!
kafka-console-consumer.sh --bootstrap-server localhost:9092 --topic orders-shared --group orders-team --from-beginning --command-property client.id=reader-b --command-property enable.auto.commit=true --command-property auto.commit.interval.ms=1000 --command-property max.poll.records=1 > /home/labex/project/output/reader-b.jsonl 2> /home/labex/project/output/reader-b.log &
consumer_b_pid=$!
Warten Sie auf die tatsächliche Zuweisung an zwei Mitglieder und speichern Sie den Bericht:
/home/labex/project/tools/wait-assignment 2 > /home/labex/project/output/two-members.txt
cat /home/labex/project/output/two-members.txt
Das ausschließlich lesende Hilfsprogramm fragt den Mitgliederbericht der Consumer-Gruppe wiederholt ab, bis zwei gekennzeichnete Clients nicht leere, überschneidungsfreie Zuweisungen besitzen, deren Vereinigung die Partitionen 0 und 1 ergibt. Es hat eine Frist und erzeugt weder künstliche Zuweisungen noch verändert es den Fortschritt.
Im Bericht identifiziert CLIENT-ID reader-a oder reader-b, und CURRENT-ASSIGNMENT listet den tatsächlichen aktuellen Besitz auf. Beispielsweise bedeutet orders-shared:0 die Partition 0 dieses Topics. Eine Zielzuweisung kann noch ausstehen; untersuchen Sie deshalb die Spalte für die aktuelle Zuweisung. Erwarten Sie eine Partition je Mitglied, ohne vorauszusetzen, dass reader-a eine bestimmte Partitionsnummer erhält. Ein Rebalance verteilt den Partitionsbesitz neu, wenn Gruppenmitglieder hinzukommen oder ausscheiden. Die Gruppe benötigt Zeit, um ihre neue Zuweisung zu erreichen. Der Start eines Prozesses bedeutet nicht, dass sich der Besitz bereits stabilisiert hat. Eine stabile Zuweisung belegt die Arbeitsaufteilung; zwei Prozess-IDs allein beweisen nicht, welche Partitionen ihnen gehören.
Veröffentlichen Sie Arbeit und vergleichen Sie eine unabhängige Gruppe
In diesem Schritt geben Sie beiden zugewiesenen Partitionen Datensätze und beobachten eine zweite Gruppe beim Lesen desselben Verlaufs. Untersuchen Sie die einsehbaren Stapel. Der Stapel initial enthält share1 bis share6; followup enthält share7 und share8:
cat /home/labex/project/data/group-batches.json
Das bereitgestellte Veröffentlichungswerkzeug sendet die ausgewählten Testdaten an den tatsächlichen Broker, wartet auf Zustellungsrückrufe und meldet bei unvollständiger Zustellung einen Fehler. Veröffentlichen Sie die sechs anfänglichen Ereignisse erst nach der stabilen Zuweisung:
/home/labex/project/tools/publish-shared-batch initial
/home/labex/project/tools/wait-caught-up orders-team 3 > /home/labex/project/output/team-initial-position.txt
cat /home/labex/project/output/team-initial-position.txt
Das ausschließlich lesende Hilfsprogramm wartet, bis beide Partitionen den festgeschriebenen nächsten Offset 3, den Endoffset 3 und den Rückstand 0 haben. Jede Partition hat drei Datensätze empfangen. Untersuchen Sie, was die beiden Clients beobachtet haben:
cat /home/labex/project/output/reader-a.jsonl
cat /home/labex/project/output/reader-b.jsonl
Zusammen sollten ihre Dateien alle sechs anfänglichen IDs enthalten. Der aktuelle Zuweisungsbericht belegt den stabilen Besitz; eine Datei kann auch frühere Beobachtungen aus der Zeit vor einer Zuweisungsänderung enthalten.
Starten Sie nun eine separate begrenzte Gruppe orders-observer ab dem gespeicherten Anfang:
kafka-console-consumer.sh --bootstrap-server localhost:9092 --topic orders-shared --group orders-observer --from-beginning --max-messages 6 --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/observer-initial.jsonl
cat /home/labex/project/output/observer-initial.jsonl
/home/labex/project/tools/wait-caught-up orders-observer 3 > /home/labex/project/output/observer-initial-position.txt
cat /home/labex/project/output/observer-initial-position.txt
Sie sollte alle sechs Datensätze unabhängig empfangen. Sie entzieht orders-team keine Partitionen, da sie zu einer anderen Gruppe gehört. Beide Gruppen können gleichzeitig auf dem neuesten Stand sein und dabei getrennte festgeschriebene Positionen behalten.
Beenden Sie ein Mitglied und beobachten Sie die verbleibende Zuweisung
In diesem Schritt lassen Sie reader-a weiterlaufen und beenden nur reader-b. kill sendet dem gespeicherten Prozess ein Beendigungssignal; der Broker wird nicht beendet. Mit wait kann die Shell den Abschluss dieses Kindprozesses entgegennehmen. Ein durch ein Signal beendeter Prozess kann einen Exitstatus ungleich null melden. Deshalb verhindert das abschließende || true, dass dieser beabsichtigte Aufräumbefehl als Anwendungsfehler behandelt wird:
kill "$consumer_b_pid"
wait "$consumer_b_pid" || true
/home/labex/project/tools/wait-assignment 1 > /home/labex/project/output/one-member.txt
cat /home/labex/project/output/one-member.txt
Erwarten Sie, dass reader-a nach der Stabilisierung der Gruppe beide Partitionen besitzt. Das Wartewerkzeug prüft den tatsächlichen Bericht vor der nächsten Veröffentlichung. Bei zwei Partitionen würden durch das Hinzufügen von mehr als zwei Mitgliedern einige Mitglieder in einer stabilen Zuweisung ohne Partition bleiben. Diese Übung erfordert keinen dritten Client.
Veröffentlichen Sie ein Folgeereignis je Partition und warten Sie darauf, dass das verbleibende Mitglied aufholt:
/home/labex/project/tools/publish-shared-batch followup
/home/labex/project/tools/wait-caught-up orders-team 4 > /home/labex/project/output/team-followup-position.txt
cat /home/labex/project/output/team-followup-position.txt
cat /home/labex/project/output/reader-a.jsonl
reader-a sollte nun sowohl share7 als auch share8 enthalten. reader-b wurde beendet und kann diese neuen Datensätze deshalb nicht verarbeiten. Die Beobachtergruppe hat weiterhin ihre frühere festgeschriebene Position. Setzen Sie sie für die beiden neuen Ereignisse fort:
kafka-console-consumer.sh --bootstrap-server localhost:9092 --topic orders-shared --group orders-observer --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/observer-followup.jsonl
cat /home/labex/project/output/observer-followup.jsonl
/home/labex/project/tools/wait-caught-up orders-observer 4 > /home/labex/project/output/observer-followup-position.txt
cat /home/labex/project/output/observer-followup-position.txt
Erwarten Sie share7 und share8 sowie den eigenen nächsten Offset 4 jeder Gruppe in beiden Partitionen. Die Option zum Lesen ab dem Anfang überschreibt nicht den gültigen gespeicherten Offset 3 der Beobachtergruppe.
Ein Rebalance kann eine erneute Zustellung verursachen, wenn ein ausscheidendes Mitglied Datensätze verarbeitet hat, bevor sein Fortschritt festgeschrieben wurde. Ereignisbeobachtungen können deshalb Wiederholungen enthalten. Eine stabile Zuweisung gibt einer Partition dennoch jeweils nur einen Besitzer. Dieses Lab prüft tatsächliche Zuweisungen und vollständige Ereignisinhalte, ohne eine durchgängige Exactly-once-Zustellung zu beanspruchen.
Beenden Sie den verbleibenden Client und bewahren Sie die Nachweise auf
In diesem Schritt schließen Sie das dauerhaft laufende Experiment ab, indem Sie den einen verbleibenden Client beenden. Seine endgültigen Positionen wurden bereits am Ende beobachtet, sodass keine neue Arbeit aussteht. Verwenden Sie seine eigene gespeicherte PID:
kill "$consumer_a_pid"
wait "$consumer_a_pid" || true
/home/labex/project/tools/wait-assignment 0 > /home/labex/project/output/no-members.txt
cat /home/labex/project/output/no-members.txt
Die Gruppe sollte nun keine aktiven Mitglieder haben. Gespeicherte Offsets und Brokerereignisse bleiben erhalten; die Lebensdauer eines Clients und der gespeicherte Fortschritt sind voneinander getrennt. Die begrenzten Beobachterclients wurden bereits automatisch beendet.
Untersuchen Sie die aufbewahrten Mitgliederberichte von vor und nach der Übernahme:
cat /home/labex/project/output/two-members.txt /home/labex/project/output/one-member.txt
Der erste Bericht zeigt zwei Besitzer ohne Überschneidung, der zweite einen Besitzer beider Partitionen. Dies sind tatsächliche Berichte mit laufabhängigen Mitgliederkennungen, keine feststehenden Zuweisungen zum Auswendiglernen. Die Reader- und Beobachterdateien bewahren Ereignisnachweise für beide Gruppen auf. Sie haben nur Ihre beiden Konsolenclients beendet und dabei das Topic sowie die festgeschriebenen Positionen zur Überprüfung erhalten.
Zusammenfassung
Sie haben beobachtet, wie zwei Consumer tatsächliche Partitionszuweisungen aufteilen, eine separate Gruppe dieselben Datensätze unabhängig liest und ein verbleibendes Mitglied beide Partitionen übernimmt, bevor neue Arbeit eintrifft. Begrenzte Zustandsprüfungen ersetzten Vermutungen über Wartezeiten. Gezieltes Beenden schloss die Clients und erhielt dabei Fortschritt und Ereignisse. Als Nächstes ersetzen Sie das vorbereitete Veröffentlichungswerkzeug durch einen Python-Producer, dessen Serialisierung und Zustellungsbehandlung Sie lesen und ändern können.
