Einführung
Der erfolgreiche Consumer schrieb eine Datei und anschließend den Lesefortschritt fest. Dies sind zwei getrennte Vorgänge. Dieses Lab macht zwei Fehlerfälle sichtbar: Ein fehlerhafter JSON-Datensatz muss die Verarbeitung stoppen, bevor der Fortschritt über ihn hinausgeht. Ein Prozess, der nach dem Schreiben, aber vor dem Commit beendet wird, kann denselben Datensatz erneut empfangen.
Sie führen zunächst eine normale Zustellung durch, untersuchen dann die genaue Position eines fehlerhaften Datensatzes und verwenden schließlich einen bereitgestellten Übungsschalter, um zwischen flush und commit zu stoppen. Ein Neustart derselben Gruppe zeigt die erneute Zustellung in der Datei anhand derselben Ereignis-ID und Kafka-Position. Sie beobachten dieses Verhalten, ohne eine Duplikatbereinigung oder ein System für Wiederholungsversuche zu implementieren.
Diese unabhängige Umgebung bereitet drei Topics mit jeweils einer Partition und den vollständigen Consumer aus dem vorherigen Lab vor. Der zusätzliche Schalter zum kontrollierten Beenden ist bereits implementiert. Sie benötigen die vorherigen Konzepte zu Python-JSON, Dateien und manuellen Commits. Eine Brokerneukonfiguration oder neue Wiederherstellungsarchitektur ist nicht erforderlich. Verwenden Sie das Terminal.
Prüfen Sie den erfolgreichen Verarbeitungsablauf
In diesem Schritt bestätigen Sie, dass die Anwendung mit gültigen Eingaben funktioniert, bevor Sie einen Fehler untersuchen. Jedes Topic hat eine Partition, sodass die Datensatzpositionen direkt nachvollziehbar sind. orders-normal enthält zwei gültige Datensätze. Der Consumer verwendet dieselbe explizite Abfolge aus Schreiben, Leeren des Puffers, synchronem Nachrichten-Commit und Schließen wie zuvor.
Untersuchen Sie die bereitgestellte Anwendung einschließlich ihres Schalters zum kontrollierten Beenden, ohne sie zu ändern:
cd /home/labex/project
cat /home/labex/project/consumer.py
Der Schalter ist standardmäßig deaktiviert. Führen Sie den normalen Ablauf mit zwei Datensätzen, einer eigenen Gruppe und einer eigenen Datei aus:
bigdata-python /home/labex/project/consumer.py --topic orders-normal --group normal-reader --output /home/labex/project/output/normal-delivery.jsonl --max-messages 2 --timeout-seconds 20 > /home/labex/project/output/normal.log 2>&1
cat /home/labex/project/output/normal.log
cat /home/labex/project/output/normal-delivery.jsonl
Erwarten Sie normal1 und normal2 an den Offsets 0 und 1 und nach beiden Schreibvorgängen den nächsten festgeschriebenen Offset 2. max-messages bezeichnet die in diesem Aufruf angeforderte Anzahl. Die Datei im Anhängemodus kann in späteren Schritten Beobachtungen mehrerer Aufrufe aufbewahren.
Dieser erfolgreiche Ausgangstest grenzt ein anwendungsweites Verbindungsproblem von einem datensatzspezifischen Fehler ab. Für die absichtlichen Fehler verwenden Sie unterschiedliche Topics, Gruppen und Dateien, damit die Nachweise eindeutig bleiben.
Stoppen Sie bei fehlerhaftem JSON, ohne darüber hinaus festzuschreiben
In diesem Schritt untersuchen Sie einen gültigen Anfangsdatensatz, einen ungültigen JSON-Datensatz und einen gültigen späteren Datensatz. Lesen Sie die lokalen Nachweise:
cat /home/labex/project/data/bad-events.txt
In Partition 0 von orders-bad liegt der gültige Anfangsdatensatz an Offset 0, der fehlerhafte Text an Offset 1 und das spätere gültige Ereignis an Offset 2. Die Richtlinie lautet, sofort abzubrechen: die fehlerhafte Position melden und beenden. Fangen Sie den Fehler nicht ab, um anschließend weitere Datensätze dieser Partition festzuschreiben.
Fordern Sie die Zustellung der drei Datensätze an. Der nächste Befehl soll fehlschlagen.
bigdata-python /home/labex/project/consumer.py --topic orders-bad --group bad-reader --output /home/labex/project/output/bad-prefix.jsonl --max-messages 3 --timeout-seconds 20 > /home/labex/project/output/bad-first.log 2>&1
printf '%s\n' "$?" > /home/labex/project/output/bad-first-exit.txt
cat /home/labex/project/output/bad-first-exit.txt
cat /home/labex/project/output/bad-first.log
cat /home/labex/project/output/bad-prefix.jsonl
Erwarten Sie einen Status ungleich null und Decode failure mit der Angabe topic=orders-bad partition=0 offset=1. Die Datei sollte nur bad-prefix an Offset 0 enthalten. Dessen erfolgreicher Schreibvorgang wurde festgeschrieben, sodass der nächste Offset der Gruppe 1 ist. Der fehlerhafte Datensatz und der spätere Datensatz bleiben unverarbeitet.
Starten Sie dieselbe Gruppe neu und hängen Sie an dieselbe Datei an. Der nächste Befehl soll fehlschlagen.
bigdata-python /home/labex/project/consumer.py --topic orders-bad --group bad-reader --output /home/labex/project/output/bad-prefix.jsonl --max-messages 3 --timeout-seconds 20 > /home/labex/project/output/bad-retry.log 2>&1
printf '%s\n' "$?" > /home/labex/project/output/bad-retry-exit.txt
cat /home/labex/project/output/bad-retry-exit.txt
cat /home/labex/project/output/bad-retry.log
kafka-consumer-groups.sh --bootstrap-server localhost:9092 --group bad-reader --describe
Derselbe Offset 1 sollte erneut zum Fehler führen. Der gültige Anfangsdatensatz sollte nicht ein zweites Mal geschrieben werden, da seine nächste Position 1 bereits festgeschrieben wurde. Das spätere gültige Ereignis darf nicht in der Datei erscheinen. So wird die Richtlinie sichtbar: Eine verantwortliche Person muss entscheiden, wie die fehlerhaften Daten behandelt werden. Diese minimale Anwendung überspringt sie weder stillschweigend noch schreibt sie darüber hinaus fest.
Beenden Sie nach erfolgreichem Schreiben, aber vor dem Commit
In diesem Schritt erzeugen Sie mit dem vorbereiteten Topic orders-replay ein anderes Fehlerfenster. Es enthält nur repeat1 an Offset 0. Der Übungsschalter --exit-before-commit beendet den Prozess mit Status 73 unmittelbar nach dem Leeren des Dateipuffers und vor dem expliziten Commit. Er protokolliert die tatsächliche Position und schließt den Consumer weiterhin bei deaktivierten automatischen Commits.
Verwenden Sie eine neue Gruppe und eine separate Datei. Der nächste Befehl soll fehlschlagen.
bigdata-python /home/labex/project/consumer.py --topic orders-replay --group replay-reader --output /home/labex/project/output/redelivery.jsonl --max-messages 1 --timeout-seconds 20 --exit-before-commit > /home/labex/project/output/before-commit-exit.log 2>&1
printf '%s\n' "$?" > /home/labex/project/output/before-commit-exit-status.txt
cat /home/labex/project/output/before-commit-exit-status.txt
cat /home/labex/project/output/before-commit-exit.log
cat /home/labex/project/output/redelivery.jsonl
Erwarten Sie Status 73 und eine gespeicherte Hülle für repeat1 in Partition 0 an Offset 0. Das Protokoll enthält CONTROLLED_EXIT_AFTER_WRITE_BEFORE_COMMIT. Dies ist eine bereitgestellte Fehlerdemonstration, keine Änderung, die Sie implementieren müssen.
Die Datei enthält nun den Datensatz, während die Gruppe dafür keine festgeschriebene Position hat. Beim erneuten Start derselben Gruppe gelten daher die früher erläuterten Regeln zur Startrichtlinie. Genau diese Trennung beobachtet der nächste Schritt.
Starten Sie neu und identifizieren Sie den erneut zugestellten Kafka-Datensatz
In diesem Schritt starten Sie replay-reader ohne den Übungsschalter neu. Die fehlende festgeschriebene Position und die Richtlinie earliest lassen die Gruppe den gespeicherten Datensatz erneut lesen. Behalten Sie den Anhängemodus und dieselbe Ausgabedatei bei, damit beide Zustellungen sichtbar bleiben:
bigdata-python /home/labex/project/consumer.py --topic orders-replay --group replay-reader --output /home/labex/project/output/redelivery.jsonl --max-messages 1 --timeout-seconds 20 > /home/labex/project/output/after-restart.log 2>&1
cat /home/labex/project/output/after-restart.log
cat /home/labex/project/output/redelivery.jsonl
kafka-consumer-groups.sh --bootstrap-server localhost:9092 --group replay-reader --describe
Erwarten Sie zwei Hüllen mit derselben event_id, demselben Schlüssel, Wert, Topic, derselben Partition und demselben Offset. Es gibt weiterhin nur einen ursprünglichen Brokerdatensatz. Der zweite Aufruf schreibt erfolgreich den nächsten Offset 1 fest, während die Datei Nachweise beider Zustellungen aufbewahrt.
Unterscheiden Sie häufige Situationen anhand der folgenden Nachweise:
| Beobachtung | Erste nützliche Nachweise |
|---|---|
| Keine neuen Ereignisse | Gültiges Topic und Gruppenfortschritt am Ende; eine begrenzte Abfrage kann ohne Datensätze enden |
| Falscher Topicname | Vergleichen Sie den übergebenen Namen mit der Topicliste oder -beschreibung; dieser Client deaktiviert die versehentliche Topicerstellung |
| Keine gültige festgeschriebene Position | Untersuchen Sie den Gruppenfortschritt und die konfigurierte Startrichtlinie earliest; sie kann gespeicherte Verläufe lesen |
| Dekodierungsfehler | Lesen Sie die gemeldeten Angaben zu Topic, Partition und Offset sowie den Wertnachweis; die Gruppe stoppt vor dieser fehlerhaften Position |
Das Schreiben einer Datei und das Festschreiben einer Kafka-Position bilden keine gemeinsame Transaktion. Eine Unterbrechung dazwischen kann zu erneuter Zustellung führen, wie diese tatsächliche Datensatzposition zeigt. Diese Übung erhält beide Zustellungen und implementiert weder eine Duplikatbereinigung noch eine Dead-Letter-Queue oder einen transaktionalen Entwurf. Die Abschlusschallenge wendet den erfolgreichen Ablauf mit einem expliziten Wiederholungsverfahren für einen sauberen Lauf an.
Zusammenfassung
Sie haben eine normale Zustellung durchgeführt, bei fehlerhaftem JSON gestoppt, ohne darüber hinaus festzuschreiben, und beobachtet, wie ein geschriebener, aber nicht festgeschriebener Datensatz nach einem Neustart erneut zugestellt wurde. Übereinstimmende Ereignisinhalte und Kafka-Positionen erklärten die doppelten Dateibeobachtungen, ohne ein zweites Brokerereignis zu erfinden. Die Abschlusschallenge verbindet nun erfolgreiche Veröffentlichung, begrenztes Konsumieren, Dateiausgabe und das Schreiben vor dem Commit anhand neuer Daten.
