Einführung
Kafka bewahrt Datensätze unabhängig davon auf, wer sie gelesen hat. Eine Consumer-Gruppe muss sich außerdem ihren Fortschritt merken, damit ein späterer Prozess fortfahren kann, statt von vorn zu beginnen. Ein festgeschriebener Offset ist die nächste Position, die diese Gruppe in einer Partition lesen soll, nicht die Nummer des letzten Datensatzes.
Sie lesen zwei von vier vorbereiteten Datensätzen, untersuchen den festgeschriebenen Fortschritt und den Rückstand, setzen mit derselben Gruppe fort und vergleichen eine neue Gruppe. Anschließend prüfen Sie die Vorschau eines Zurücksetzens, führen es bei inaktiver ursprünglicher Gruppe aus und lesen die gespeicherten Datensätze erneut. Das Topic hat eine Partition, damit die Positionen leicht nachvollziehbar bleiben.
Dieses unabhängige Lab stellt den Broker, vier Datensätze und ein Hilfsprogramm bereit, das den tatsächlichen Koordinatorzustand abfragt, bis ein erwarteter Offset erscheint. Grundlegende Terminalkenntnisse reichen aus; Python-Code oder die Einrichtung eines Dienstes sind nicht erforderlich. Das Hilfsprogramm verändert keine Offsets und erzeugt keinen künstlichen Bericht. Verwenden Sie das Terminal.
Lesen Sie zwei Datensätze und prüfen Sie die nächste festgeschriebene Position
In diesem Schritt verbinden Sie einen begrenzten Lesevorgang mit der tatsächlich gespeicherten Position der Gruppe. Untersuchen Sie die vier vorbereiteten Werte:
cd /home/labex/project
cat /home/labex/project/data/progress-events.txt
Ihre Offsets in diesem neuen Topic mit einer Partition reichen von 0 bis 3. Eine neue Gruppe namens orders-review hat keine gespeicherte Position. Deshalb wählt --from-beginning für diesen ersten Lesevorgang den gespeicherten Anfang aus.
--command-property übergibt einen Konfigurationswert an den Consumer-Client, statt das Anzeigeformat der Ereignisse zu ändern. Der Konsolenclient schreibt in dieser Übung den Fortschritt automatisch fest. max.poll.records=1 begrenzt jeden Poll-Aufruf auf einen zurückgegebenen Datensatz. So führt das Beenden nach zwei angezeigten Datensätzen nicht dazu, dass die Position durch einen größeren zurückgegebenen Stapel voranschreitet. Das Intervall legt das regelmäßige Festschreiben fest; beim Schließen schließt der Client außerdem seine normale Fortschrittsverwaltung ab. Sie untersuchen das tatsächliche Ergebnis, statt eine bestimmte Verzögerung anzunehmen:
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
Erwarten Sie die beiden created-Ereignisse; danach beendet sich der Client. Führen Sie das bereitgestellte Hilfsprogramm für begrenzte wiederholte Abfragen mit dem Gruppennamen und dem erwarteten nächsten Offset aus:
/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
Das Hilfsprogramm fragt die tatsächliche Gruppe wiederholt ab, bis ihr festgeschriebener Offset 2 beträgt, oder meldet nach Ablauf seiner Frist einen Fehler. Es schreibt nichts fest und setzt nichts zurück. Suchen Sie für Partition 0 nach CURRENT-OFFSET 2, LOG-END-OFFSET 4 und LAG 2. Eine Meldung, dass die Gruppe keine aktiven Mitglieder hat, ist nach dem Ende des begrenzten Clients normal; ihr gespeicherter Fortschritt besteht weiterhin.
Offset 2 bedeutet, dass der nächste Lesevorgang mit dem dritten gespeicherten Datensatz beginnt. Der Endoffset 4 ist die nächste Position nach den vier vorhandenen Datensätzen. Der Rückstand 4 − 2 = 2 misst in diesem einfachen Beispiel bei unverändertem Datenbestand die verbleibenden Datensätze. Die ersten beiden Datensätze wurden nicht gelöscht.
Setzen Sie dieselbe Gruppe fort und vergleichen Sie eine neue Gruppe
In diesem Schritt sagen Sie voraus, welche Datensätze eine Gruppe ab ihrer gespeicherten Position empfängt. Führen Sie dieselbe Gruppe erneut aus und behalten Sie dabei bewusst die Option zum Lesen ab dem Anfang bei:
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
Erwarten Sie die beiden paid-Ereignisse, nicht die früheren created-Ereignisse. --from-beginning gibt eine Startrichtlinie vor, wenn kein gültiger festgeschriebener Offset vorhanden ist. Es überschreibt nicht den gültigen gespeicherten Offset 2 dieser Gruppe. Bestätigen Sie den Abschluss anhand des tatsächlichen Koordinatorzustands:
/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
Erwarten Sie den festgeschriebenen Offset 4, den Endoffset 4 und den Rückstand 0. Überlegen Sie nun, was eine andere Gruppe namens orders-independent lesen wird. Sie hat keine gespeicherte Position und ist von orders-review unabhängig. Ein Lesevorgang ab dem Anfang sollte daher alle vier gespeicherten Datensätze zurückgeben:
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
Erwarten Sie alle vier Werte und den eigenen festgeschriebenen Offset 4 der neuen Gruppe. Beide Gruppen können das Ende erreichen, obwohl sie unterschiedliche Leseverläufe durchlaufen haben. Der Fortschritt der einen Gruppe hat die Startposition der anderen nicht verschoben, und das Konsumieren durch beide Gruppen hat die Topicdatensätze nicht entfernt.
Als Nächstes ändern Sie die gespeicherte Position der ursprünglichen Gruppe gezielt, statt die Option zum Lesen ab dem Anfang als Rückspulbefehl zu behandeln.
Prüfen Sie das Zurücksetzen einer inaktiven Gruppe vorab und lesen Sie den Verlauf erneut
In diesem Schritt setzen Sie nur orders-review zurück. Ihr begrenzter Consumer wurde beendet, sodass die Gruppe kein aktives Mitglied hat. Beenden Sie vor dem Zurücksetzen alle versehentlich noch laufenden Mitglieder derselben Gruppe. Ein aktiver Consumer könnte den Fortschritt weiter verändern.
--reset-offsets wählt den Vorgang zur Fortschrittsverwaltung aus, --topic begrenzt seinen Umfang und --to-earliest wählt den gespeicherten Anfang. Verwenden Sie zunächst --dry-run, um den vorgeschlagenen neuen Offset anzuzeigen, ohne ihn anzuwenden:
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
Erwarten Sie einen vorgeschlagenen NEW-OFFSET von 0. Prüfen Sie, dass die Vorschau die aktuell gespeicherte Position nicht verändert hat:
/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
Wenden Sie nun dieses begrenzte Zurücksetzen mit --execute an:
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
Der tatsächlich festgeschriebene nächste Offset sollte nun 0 betragen, bei Endoffset 4 und Rückstand 4. Die Datensätze blieben an ihrem Platz; nur die gespeicherte Position dieser Gruppe hat sich geändert. Lesen Sie alle vier erneut über den nun gültigen festgeschriebenen Anfang, ohne sich auf eine Option zur Startvorgabe zu stützen:
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
Erwarten Sie die ursprünglichen vier Werte und erneut den festgeschriebenen Offset 4. orders-independent sollte weiterhin seine eigene abgeschlossene Position haben; das Zurücksetzen hat diese Gruppe nicht zurückgespult.
Beim erneuten Lesen werden vorhandene Datensätze absichtlich noch einmal gelesen. Auch ein Absturz zwischen der Verarbeitung eines Datensatzes und dem Festschreiben des Fortschritts kann eine erneute Zustellung verursachen. Diese Konsolenübung veranschaulicht Lesepositionen, keine atomare Transaktion aus Dateischreiben und Festschreiben und keine durchgängige Exactly-once-Garantie. Spätere Python-Labs machen die Reihenfolge von Verarbeitung und Festschreiben explizit.
Zusammenfassung
Sie haben festgeschriebene nächste Offsets und Rückstände untersucht, trotz einer Startvorgabe ab dem gespeicherten Fortschritt fortgesetzt und eine unabhängige Gruppe verglichen. Die Vorschau und das ausgeführte Zurücksetzen einer inaktiven Gruppe ermöglichten ein gezieltes erneutes Lesen, während die tatsächlichen Topicdatensätze unverändert blieben. Als Nächstes beobachten Sie, wie mehrere Mitglieder einer Gruppe Partitionen untereinander aufteilen und nach dem Ausscheiden eines Mitglieds eine Zuweisung übernehmen.
