はじめに
Kafka は、誰が読んだかとは独立してレコードを保持します。コンシューマーグループには、後で起動したプロセスが最初からではなく続きから処理できるよう、進行状況を記憶する仕組みも必要です。コミット済みオフセットは、そのグループがパーティション内で次に読む位置を表します。最後に読んだレコードの番号ではありません。
用意された4件のうち2件を読み取り、コミットされた進行状況とラグを調べ、同じグループで再開して、新しいグループと比較します。その後、元のグループが非アクティブな間にリセットをプレビューして実行し、保持されたレコードを再び読み取ります。位置を考えやすいよう、トピックのパーティションは1つにしています。
この独立した演習には、ブローカー、4件のレコード、期待するオフセットが現れるまで実際のコーディネーターの状態をポーリングする補助ツールが用意されています。基本的なターミナル操作ができれば十分で、Python のコーディングやサービスの設定は不要です。補助ツールはオフセットの変更や架空のレポートの作成を行いません。Terminal を使用してください。
2件を読み取ってコミット済みの次の位置を調べる
このステップでは、件数を制限した読み取りと、グループが実際に保存した位置を結び付けます。用意された4つの値を確認します。
cd /home/labex/project
cat /home/labex/project/data/progress-events.txt
この新しい単一パーティションのトピックでは、各値のオフセットは0から3です。orders-review という新しいグループには保存済みの位置がないため、--from-beginning によって初回は保持された先頭から読み取ります。
--command-property は、イベントの表示形式を変えるのではなく、コンシューマークライアントに設定値を渡します。この演習では、コンソールクライアントが進行状況を自動コミットします。max.poll.records=1 は、各ポーリングが返すレコードを1件に制限します。これにより、2件を表示して停止するときに、返された大きなバッチの先まで位置が進むのを防ぎます。間隔の設定は定期的なコミット動作を決め、終了するクライアントも通常の進行状況の処理を完了します。特定の遅延を仮定せず、実際の結果を確認します。
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
2件の created イベントを取得すると、クライアントは終了するはずです。用意された時間制限付きのポーリング補助ツールを、グループ名と期待する次のオフセットを指定して実行します。
/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
補助ツールは、コミット済みオフセットが2になるまで実際のグループ情報を繰り返し取得し、期限内に確認できなければ失敗します。コミットやリセットは行いません。パーティション0について CURRENT-OFFSET 2、LOG-END-OFFSET 4、LAG 2 を探してください。件数を制限したクライアントが終了した後に、グループにはアクティブなメンバーがいないというメッセージが出るのは正常です。保存済みの進行状況は残っています。
オフセット2は、次の読み取りが保存された3件目のレコードから始まることを意味します。末尾オフセット4は、既存の4件の次の位置です。この単純で状態が落ち着いた例では、ラグ 4 − 2 = 2 が残りのレコード数を表します。最初の2件は削除されていません。
同じグループを再開して新しいグループと比較する
このステップでは、保存済みの位置からグループがどのレコードを受け取るかを予想します。先頭から読むオプションを意図的に残して、同じグループをもう一度実行します。
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
以前の created イベントではなく、2件の paid イベントになるはずです。--from-beginning は有効なコミット済みオフセットがない場合の開始方針を指定します。このグループの有効な保存済みオフセット2は上書きしません。実際のコーディネーターの状態で完了を確認します。
/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
コミット済みオフセット4、末尾オフセット4、ラグ0になるはずです。次に、orders-independent という別のグループについて予想します。このグループには保存済みの位置がなく、orders-review から独立しているため、先頭から読み取ると保持された4件すべてを取得できるはずです。
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
4つの値すべてと、新しいグループ自身のコミット済みオフセット4を確認できるはずです。異なる読み取り履歴をたどっても、両方のグループは末尾に到達できます。一方のグループの進行状況は、他方の開始位置を動かしません。また、どちらが消費しても、トピックのレコードは削除されません。
次は、先頭から読むオプションを巻き戻しコマンドとして扱わず、元のグループの保存済み位置を意図的に変更します。
非アクティブなグループのリセットをプレビューして履歴を再読する
このステップでは、orders-review だけをリセットします。件数を制限したコンシューマーは終了済みなので、グループにアクティブなメンバーはいません。同じグループのメンバーを誤って動作させたままにしている場合は、リセット前に停止してください。アクティブなコンシューマーがいると、進行状況の変更が続く可能性があります。
--reset-offsets は進行状況を管理する操作を選択し、--topic は対象を限定し、--to-earliest は保持された先頭を指定します。まず --dry-run を使い、適用せずに新しいオフセットの案を確認します。
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
提案される NEW-OFFSET は0のはずです。プレビューによって現在の保存済み位置が変わっていないことを確認します。
/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
次に --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
実際のコミット済みの次のオフセットは0になり、末尾オフセットは4、ラグは4のはずです。レコードはそのままで、このグループの保存済み位置だけが変わりました。開始位置のデフォルト方針に頼らず、今は有効なコミット済み位置となった先頭から、4件すべてを再び読み取ります。
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
元の4つの値が読み取れ、コミット済みオフセットは再び4になるはずです。orders-independent は、自身の完了した位置を維持しているはずです。このリセットはそのグループを巻き戻していません。
再読み取りは、既存のレコードを意図的にもう一度読む操作です。レコードの処理と進行状況のコミットの間にクラッシュした場合にも、同じレコードが再配信されることがあります。このコンソール演習が示すのは読み取り位置であり、ファイル書き込みとコミットを不可分に実行するトランザクションや、エンドツーエンドでの exactly-once の保証ではありません。後の Python 演習では、処理とコミットの順序を明示します。
まとめ
コミット済みの次のオフセットとラグを調べ、先頭から読む初期方針があっても保存済みの進行状況から再開することを確認し、独立したグループと比較しました。グループが非アクティブな間にリセットをプレビューして実行し、実際のトピックレコードを変えずに意図的な再読み取りを行いました。次は、同じグループの複数メンバーがパーティションを分担し、メンバーが抜けた後に割り当てを引き継ぐ様子を観察します。
