不正な Kafka イベントを扱い、再配信を観察する

はじめに

正常に動作したコンシューマーは、ファイルを書き込んでから読み取りの進行状況をコミットしました。これらは別々の操作です。この演習では、2つの失敗ケースを観察します。不正な JSON レコードでは、その位置を越えて進行状況を進める前に処理を止める必要があります。また、書き込み後・コミット前にプロセスが終了すると、同じレコードを再び受け取る場合があります。

まず正常な配信を確認し、次に不正なレコードの正確な位置を調べ、最後に用意された学習用スイッチでフラッシュとコミットの間で停止します。同じグループを再起動すると、同じイベント ID と Kafka 上の位置によって、ファイル内で再配信を確認できます。重複排除や再試行の枠組みを実装せずに、この動作を観察します。

この独立した環境には、パーティションが1つのトピックが3つと、前の演習で使った完全なコンシューマーが用意されています。追加の意図的な終了スイッチも実装済みです。これまでの Python の JSON、ファイル、手動コミットの概念を使います。ブローカーの再設定や新しい復旧アーキテクチャは不要です。Terminal を使用してください。

正常な処理経路を確認する

このステップでは、不具合を診断する前に、有効な入力でアプリケーションが動くことを確認します。位置を分かりやすくするため、各トピックのパーティションは1つです。orders-normal には有効なレコードが2件あります。コンシューマーは以前と同じ、明示的な書き込み、フラッシュ、メッセージの同期コミット、終了という順序を使います。

意図的な終了スイッチを含め、用意されたアプリケーションを変更せずに確認します。

cd /home/labex/project

cat /home/labex/project/consumer.py

スイッチはデフォルトでは無効です。専用のグループとファイルで、通常の2件の処理を実行します。

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

normal1 と normal2 がオフセット0と1にあり、両方の書き込み後に次のコミット済みオフセットが2になるはずです。max-messages は今回の実行で要求する件数です。追記専用のファイルには、後のステップで複数回の実行の観察結果を残せます。

正常な基準を作ることで、アプリケーション全体の接続問題と、特定のレコードの失敗を区別できます。意図的な失敗では、確認結果を分かりやすくするため、別々のトピック、グループ、ファイルを使います。

不正な JSON で停止し、その先をコミットしない

このステップでは、有効な先頭レコード、不正な JSON レコード、その後の有効なレコードを調べます。ローカルに用意された内容を読みます。

cat /home/labex/project/data/bad-events.txt

orders-bad のパーティション0では、有効な先頭レコードがオフセット0、不正なテキストがオフセット1、その後の有効なイベントがオフセット2にあります。方針は、失敗を検出したら直ちに停止することです。不正な位置を報告して終了します。エラーを捕捉して続行し、同じパーティションの後のレコードまでコミットしないでください。

3件の配信を要求します。次のコマンドは失敗することを想定しています。

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

0以外の状態と、topic=orders-bad partition=0 offset=1 を示す Decode failure があるはずです。ファイルにはオフセット0の bad-prefix だけが含まれるはずです。その書き込みは成功してコミットされたため、グループの次のオフセットは1です。不正なレコードとその後のレコードは未処理のまま残っています。

同じグループを再起動し、同じファイルに追記します。次のコマンドも失敗することを想定しています。

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

同じオフセット1で再び失敗するはずです。有効な先頭レコードは、次の位置1がすでにコミットされているため、2回目は書かれないはずです。その後の有効なイベントもファイルに含まれてはいけません。これにより、方針が明確になります。不正なデータへの対処方法は運用担当者が選ぶ必要があり、この最小限のアプリケーションは、黙って飛ばしたり、その先までコミットしたりしません。

書き込み成功後、コミット前に終了する

このステップでは、用意された orders-replay トピックを使い、別の失敗が起こりうる区間を作ります。このトピックにはオフセット0の repeat1 だけがあります。学習用の --exit-before-commit スイッチは、ファイルのフラッシュ直後、明示的なコミットの直前に、状態73で終了します。実際の位置を記録し、自動コミットを無効にしたままコンシューマーを閉じます。

新しいグループと別のファイルを使います。次のコマンドは失敗することを想定しています。

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

終了状態は73で、repeat1 のエンベロープが1件、パーティション0、オフセット0として保存されるはずです。ログには CONTROLLED_EXIT_AFTER_WRITE_BEFORE_COMMIT が示されます。これは用意された障害の実演であり、自分で変更を実装する必要はありません。

ファイルにはレコードがありますが、グループにはそのコミット済み位置がありません。そのため、同じグループが再起動すると、これまで学んだ開始方針の規則が適用されます。次のステップでは、この分離による動作を観察します。

再起動して同じ Kafka レコードの再配信を確認する

このステップでは、学習用スイッチを付けずに replay-reader を再起動します。コミット済み位置がなく earliest 方針があるため、保持されたレコードを再び読み取れます。両方の配信を確認できるよう、追記モードと同じ出力ファイルを維持します。

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

event_id、キー、値、トピック、パーティション、オフセットが同じエンベロープが2件あるはずです。元のブローカーレコードは依然として1件だけです。2回目の実行は次のオフセット1を正常にコミットし、ファイルには両方の配信の記録が残ります。

よくある状況を区別するには、次の情報を使います。

ファイルへの書き込みと Kafka の位置のコミットは、1つのトランザクションではありません。実際のレコード位置で確認したとおり、間で中断すると再配信が起こる場合があります。この演習は両方の配信を残し、重複排除、デッドレターキュー、トランザクションを使う設計は実装しません。最後のチャレンジでは、初期状態からやり直す手順を明示して、正常な処理を適用します。

まとめ

正常な配信を確認し、不正な JSON を越えてコミットせずに停止して、フラッシュ済みでも未コミットのレコードが再起動後に再配信されることを観察しました。一致するイベントの内容と Kafka 上の位置によって、ブローカーに2件目のイベントがあると考えずに、ファイル内の重複した観察結果を説明できました。最後のチャレンジでは、新しいデータで、正常な送信、制限付きの受信、ファイル出力、書き込み後のコミットを組み合わせます。