はじめに
グループは、トピックのパーティションを複数のコンシューマーで分担できます。安定したグループ内では、各パーティションにアクティブな所有者が1つあります。一方、別のグループは、保持された同じレコードを独立して読み取れます。これにより、すべてのアプリケーションが単一の読み取り位置を奪い合うことなく、処理を分担できます。
1つのグループで長時間動作するコンソールクライアントを2つ起動し、送信前に実際の割り当てを確認して、別のグループの観察結果と比較します。メンバーが1つ離脱した後、残ったメンバーが両方のパーティションを所有するまで待ち、小さな追加バッチを送ります。用意された送信ツールは両方のパーティションを明示的に対象にするため、実験中はそれぞれに処理対象があります。実際にブローカーへ配信されたことも確認します。
この独立した演習には、パーティションが2つのトピック、内容を確認できるイベントのサンプルデータ、時間制限付きの読み取り専用待機ツールが用意されています。基本的なターミナル操作ができれば十分で、サンプルデータ用ツールの Python コードを編集する必要はありません。コンソールクライアントは、最後のステップで停止するまでバックグラウンドで動作します。プロセスの変数を使い続けられるよう、同じ Terminal のシェルを維持してください。
メンバーを2つ起動し、所有関係の安定を待つ
このステップでは、グループに処理対象を渡す前に割り当てを観察します。用意されたトピックには2つのパーティションがあり、まだレコードはありません。メタデータを確認します。
cd /home/labex/project
kafka-topics.sh --bootstrap-server localhost:9092 --describe --topic orders-shared
次のクライアントは、どちらも orders-team グループを使います。--command-property は各設定値をコンシューマークライアントに渡します。client.id によって、レポート内で reader-a と reader-b を識別できます。これまで使った自動コミット設定により、ポーリングを続けながら読み取りの進行状況が保存されます。
末尾の & はクライアントをシェルのバックグラウンドで起動するため、同じターミナルで確認コマンドを実行できます。$! は直前に起動したバックグラウンドプロセスの ID です。直後の代入でそれぞれを保存し、後で対象を限定して終了処理に使います。
シェルには、レコード用の標準出力と、診断用の標準エラー出力という2つの出力ストリームがあります。> は前者を JSONL ファイルに書き、2> は後者をログに書きます。consumer_a_pid=$! という代入は = の前後に空白を入れずにシェル変数に保存し、後で "$consumer_a_pid" としてそのプロセス ID を取り出します。通常のレコードと診断情報は別ファイルに書かれるので、プロンプトへの入力を妨げません。これらのクライアントは、終盤で対象を指定した kill コマンドを実行するまで動作し続けます。
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=$!
実際に2メンバーへの割り当てが整うのを待ち、レポートを保存します。
/home/labex/project/tools/wait-assignment 2 > /home/labex/project/output/two-members.txt
cat /home/labex/project/output/two-members.txt
読み取り専用の補助ツールは、コンシューマーグループのメンバーレポートを繰り返し取得します。識別名を付けた2つのクライアントに、それぞれ空ではなく重複のない割り当てがあり、合わせてパーティション0と1を所有するまで待ちます。期限があり、割り当てを作り上げたり進行状況を変更したりはしません。
レポートの CLIENT-ID は reader-a または reader-b を識別し、CURRENT-ASSIGNMENT は現在の実際の所有関係を示します。たとえば orders-shared:0 は、そのトピックのパーティション0を意味します。目標の割り当てはまだ適用待ちの場合があるため、現在の割り当ての列を確認してください。各メンバーが1つずつ所有するはずですが、reader-a が特定の番号を受け取る必要はありません。リバランスは、メンバーの参加や離脱に応じてパーティションの所有関係を再配分することです。新しい割り当てに落ち着くまで時間が必要で、プロセスを起動しただけでは所有関係が確定したとは限りません。安定した割り当てが処理の分担を示す証拠です。プロセス ID が2つあるだけでは、どのパーティションを所有しているかは分かりません。
処理対象を送信し、独立したグループと比較する
このステップでは、割り当て済みの両パーティションにレコードを送り、別のグループが同じ履歴を読む様子を観察します。内容を確認できるバッチを調べます。initial バッチには share1 から share6、followup には share7 と share8 が含まれます。
cat /home/labex/project/data/group-batches.json
用意された送信ツールは、選択したサンプルデータを実際のブローカーに送り、配信コールバックを待ち、配信が完了しなければ失敗します。割り当てが安定してから、最初の6件のイベントを送ります。
/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
読み取り専用の追い付き確認ツールは、両パーティションのコミット済みの次のオフセットが3、末尾オフセットが3、ラグが0になるまで待ちます。各パーティションには3件ずつ届いています。両クライアントの観察結果を確認します。
cat /home/labex/project/output/reader-a.jsonl
cat /home/labex/project/output/reader-b.jsonl
両方のファイルを合わせると、最初の6つの ID がすべて含まれるはずです。動作中の割り当てレポートが、安定した所有関係の証拠です。ファイルには、割り当てが変更される前の観察結果も含まれる場合があります。
次に、別のグループ orders-observer を、保持された先頭から件数を限定して起動します。
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
このグループは、6件すべてを独立して受け取るはずです。別のグループに属するため、orders-team からパーティションを奪うことはありません。別々のコミット済み位置を保持しながら、両グループが同時に末尾まで追い付けます。
一方のメンバーを停止して残った割り当てを観察する
このステップでは、reader-a を動かしたまま reader-b だけを停止します。kill は保存したプロセスに終了シグナルを送り、ブローカーは停止しません。wait は、シェルがその子プロセスの終了を回収するのを待ちます。シグナルを受けたプロセスは0以外の終了状態を返すことがあるため、末尾の || true により、この意図的な終了処理がアプリケーションの失敗として扱われないようにします。
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
グループが安定すると、reader-a が両パーティションを所有するはずです。待機ツールは、次の送信前に実際のレポートを確認します。パーティションが2つの場合、メンバーを3つ以上に増やすと、安定した割り当てではパーティションを持たないメンバーが生じます。この演習で3つ目のクライアントは必要ありません。
各パーティションに追加イベントを1件ずつ送り、残ったメンバーが追い付くまで待ちます。
/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 には、share7 と share8 の両方が含まれるはずです。reader-b は停止しているため、新しいレコードを処理できません。観察用グループは以前のコミット済み位置を保持しています。2件の新しいイベントを読むため再開します。
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
share7 と share8 が読み取れ、両グループそれぞれで、両パーティションの次のオフセットが4になるはずです。先頭から読むオプションは、観察用グループの有効な保存済みオフセット3を上書きしません。
離脱するメンバーが、進行状況をコミットする前にレコードを処理していた場合、リバランスによって再配信が起こることがあります。そのため、観察結果に重複が含まれる場合があります。それでも、安定した割り当てでは、同時点で各パーティションを所有するメンバーは1つです。この演習では、実際の割り当てと完全なイベントの内容を確認し、エンドツーエンドの exactly-once 配信は保証しません。
残ったクライアントを終了して確認結果を保持する
このステップでは、残った1つのクライアントを停止して、長時間動作させる実験を終えます。最終位置はすでに末尾に到達したことを確認しているため、未処理の新しいデータはありません。そのクライアント用に保存した 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
グループには、アクティブなメンバーがいないはずです。保存済みのオフセットとブローカーのイベントは残ります。クライアントの稼働期間と保存された進行状況は別のものです。件数を制限した観察用クライアントは、すでに自動終了しています。
引き継ぎ前後に保存したメンバーレポートを確認します。
cat /home/labex/project/output/two-members.txt /home/labex/project/output/one-member.txt
最初のレポートは重複なく分担する2つの所有者を示し、次のレポートは両パーティションを持つ1つの所有者を示します。これらは実行ごとに異なるメンバー識別子を含む実際のレポートであり、固定の割り当てとして暗記するものではありません。読み取り側と観察用のファイルには、両グループのイベントを確認できる結果が残っています。自分の2つのコンソールクライアントだけを終了し、確認用にトピックとコミット済み位置を保持できました。
まとめ
実際のパーティション割り当てを2つのコンシューマーが分担し、別のグループが同じレコードを独立して読み取り、新しい処理対象が到着する前に、残った1つのメンバーが両パーティションを引き継ぐ様子を観察しました。時間を推測して待つ代わりに、期限付きで状態を確認し、対象を限定した終了処理で進行状況とイベントを保持したままクライアントを閉じました。次は、用意された送信ツールを、シリアライズと配信処理を読んで変更できる Python プロデューサーに置き換えます。
