コンシューマーグループと協調動作

25 分で読了 #分散システム #メッセージとストリーム

このページの目次

コンシューマーグループは2つの問題を解決します。⁠どのパーティションをどのメンバーが処理するか、そしてメンバーが入れ替わった場合にどこから再開するかです。前者は所有権の調整、後者は進捗の永続化です。この2つが連携して初めて、複数のプロセスが論理的な消費タスクを分担し、再開できるようになります。

ここでは Kafka の通常の consumer group について議論し、share group は含みません。partition と offset については Kafka アーキテクチャを、重複処理や外部副作用の保証についてはメッセージセマンティクスを参照してください。

グループ化はタスクの共有を意味するが、読み取り位置の共有を意味しない

トピックに6つのパーティションがあり、コンシューマーグループに3つのメンバーがいると仮定します。

同じグループ内では、各パーティションは1つのメンバーに割り当てられます。異なるグループはそれぞれ独立して同じログを読み取ります。G1 と G2 の進捗は相互に独立しており、一方のグループが先に読み終えたからといって、もう一方のグループが読み取りを開始できるわけではありません。

6つのパーティションは最大で6つのパーティション所有権を持つメンバーに割り当てられます。7番目のメンバーはアイドル状態になる可能性があります。パーティションを増やすことでシャーディングの並列度を高められますが、キーのマッピングと状態の分布も変化します。パーティション内ではワーカースレッドを使用して並列実行できますが、順序の完了、状態の競合、および offset のコミットはアプリケーションの責任となります。

コーディネーター、メンバーシップ、および割り当て

グループコーディネーターは、メンバー、割り当て、および進捗に関連するプロトコルを管理します。メンバーはハートビートなどのインタラクションを通じてグループへの参加を証明し、コーディネーターはメンバーの離脱、タイムアウト、またはサブスクリプションの変更後に新しい割り当てを手配します。

ここでは2つのレイヤーを区別する必要があります。1つはブローカー側でグループ状態をどのように確実に保存するか、もう1つはコンシューマーがパーティションの分担をどのように取得するかです。コンシューマーは互いに Raft を実行して割り当てのコンセンサスに達するわけではありません。消費グループを総称して「コンセンサスアルゴリズム」と呼ぶと、実際のプロトコルの役割が隠蔽されてしまいます。

メンバーシップは、古いインスタンスの分離という役割も果たします。あるプロセスが通信不能になっても実行され続けている可能性があります。新しいメンバーが引き継いだからといって、古いプロセスが外部データベースへの書き込みを停止したと仮定することはできません。Kafka は自身のグループプロトコルとコミットを制約できますが、外部副作用にはトランザクション、冪等性、またはビジネスロジックによるフェンシングが必要です。

リバランス：安全な引継ぎは均等な割り当てよりも重要

P1 が C1 から C4 へ移行する例を挙げます。安全な引継ぎでは、少なくとも以下の状態を処理する必要があります。

C1 がコミットする前にクラッシュした場合、C4 が直近のデータを再実行するのは正常な復旧動作です。C1 がすでに所有権を失っている場合、古いタスクの完了後の offset をまだコミット権限があるものとして扱うことはできません。同様に、外部ストレージへの無限の書き込みを許可することもできません。

リバランスのコストには、サービス停止時間、キャッシュのウォーミングアップ、状態の復旧、および再実行が含まれます。単に割り当て表を再計算するだけではありません。

Revoked と Lost では同じ後処理を実行できない

通常の取り消し時には、アプリケーションは配信を停止し、進行中のタスクを有限の時間待ってから、完了済みの水位のコミットを試みます。待機はグループプロトコルを無限にブロックしてはいけません。コミットに失敗した場合、完了していない位置をコミットするよりも、再実行を許可する方が処理の約束を保ちやすくなります。

パーティションがすでに失われている場合、新しいメンバーが作業を開始している可能性があります。古いメンバーはローカルのタスク権限を取り消し、状態をクリーンアップする必要があります。この時点で onPartitionsLost を、まだパーティションを所有している最終的なコミットの機会として扱うことはできません。2つのコールバックの違いについては ConsumerRebalanceListener を参照してください。

ローカルの権限チェックは、そのプロセスの配信とコールバックのみを保護します。古いタスクがすでにデータベースへのリクエストを送信している場合、cancelled フラグを設定してもそのリクエストを取り消すことはできません。外部ストレージでは、実際の書き込み時に権限の世代をアトミックにチェックするか、安定したイベント ID による重複排除を行う必要があります。リクエスト送信前に所有権を1回だけクエリする場合、クエリから書き込みまでの間に引継ぎが発生する可能性があります。

Eager、Cooperative、および New Consumer プロトコル

「漸進的な引継ぎ」と「どの世代のグループプロトコルを使用するか」は2つの次元であり、単純に同一視することはできません。Kafka 4.1 のドキュメントによると：

メカニズム	割り当てと引継ぎの特徴	理解のポイント
Classic + eager 割り当て	まず広範囲に取り消しを行い、その後新しい割り当てを完了する	一時停止範囲が大きい
Classic + cooperative 割り当て	移行が必要なパーティションを段階的に取り消す	移行しないパーティションは継続可能。複数のラウンドの調整が必要になる場合がある
Consumer プロトコル	サーバー側での割り当て、新しいハートビートプロトコルによる漸進的な調整	`group.protocol=consumer` で選択。Classic とは設定が異なる

Classic における cooperative sticky assignor と KIP-848 で導入された Consumer プロトコルは、同じ機能ではありません。プロトコルのサポート、移行の制約、およびサーバー側のアサイナーについては Consumer Rebalance Protocol を参照してください。

Sticky の目的は不要な移動を減らすことです。均衡（均衡化）の目的は負荷を分散することです。パーティション数が均等であっても、作業量が均等であるとは限りません。1つのホットパーティションが他の10のパーティションよりも忙しい可能性があります。アサイナーを変更しただけでは、ホットパーティションは自動的に分割されません。

Offset は復旧の約束であり、フェッチカウントではない

コミット値は通常、⁠次に読み取るべき位置を表します。offset 100 を処理し、101 をコミットします。アプリケーションが実際に維持するのは、「それ以前のすべてのレコードが完了した」という連続した水位です。

103 が完了していても、104 をコミットすることはできません。この時点でクラッシュすると、101 がスキップされてしまいます。101 の成功を待ってから、水位は完了済みの 102、103 を越え、一度に 104 まで進めることができます。

単にすべてのワーカースレッドが完了した offset の最大値を取得してコミットすると、穴を越えてしまいます。パーティションごとに未完了の集合と連続した水位を維持することで、並列実行を信頼性の高い復旧進捗に変換できます。パーティション間で独立して進めることができ、1つの遅いパーティションがすべてのパーティションのコミットをブロックする必要はありません。

連続するのは配信済みタスクの接頭辞であり、すべての整数 offset ではない

ある poll で1つのパーティションに対して [100, 103, 107] が返された場合、スキップされた位置は圧縮またはトランザクションフィルタリングに起因する可能性があります。101、102 の完了通知を待つと永遠にブロックされます。完了済みの最大値を直接取得すると、実際の未完了タスクを越えてしまいます。正しい基準は読み取り順に登録されたタスクキューです。

以下のモデルは1バッチずつ受け入れ、バッチが完全に完了してから次のバッチを受け入れます。これにより、復旧条件を明確に記述できます。next_position はこのパーティションの今回の読み取り後の位置であり、配信済みタスクがすべて完了して初めてコミットできます。token はアプリケーションローカルに一意なバッチの権限であり、現在のパーティションの今回の割り当てに関連付けられます。これは Kafka プロトコルの generation フィールドではありません。

class BatchProgress:
    def __init__(self, offsets, next_position, token):
        if offsets != sorted(set(offsets)):
            raise ValueError("offset は厳密に増分である必要があります")
        if next_position < 0 or (offsets and next_position <= offsets[-1]):
            raise ValueError("読み取り終了位置は最後のレコードの後に存在する必要があります")
        self.offsets = list(offsets)
        self.next_position = next_position
        self.token = token
        self.active = True
        self.done = set()
        self.front = 0

    def completed(self, offset, token):
        if not self.active or token != self.token:
            return False  # 古い割り当て、古いバッチ、または取り消し後のコールバック
        if offset not in self.offsets:
            raise ValueError("このバッチは該 offset を配信していません")
        self.done.add(offset)
        while (self.front < len(self.offsets)
               and self.offsets[self.front] in self.done):
            self.front += 1
        return True

    def committable(self):
        if not self.active:
            return None
        if self.front == len(self.offsets):
            return self.next_position
        return self.offsets[self.front]  # 最も早くに未完了となった位置

    def revoke(self):
        self.active = False

old = BatchProgress([100, 103, 107], 110, token="assignment-A/batch-1")
assert old.committable() == 100
assert old.completed(107, old.token)
assert old.committable() == 100
assert old.completed(100, old.token)
assert old.committable() == 103
assert old.completed(100, old.token)  # 重複する完了通知は2回進めない
assert old.committable() == 103
old.revoke()
assert not old.completed(103, old.token)
assert old.committable() is None

new = BatchProgress([103, 107], 110, token="assignment-B/batch-1")
assert not new.completed(103, old.token)
assert new.committable() == 103
assert new.completed(107, new.token)
assert new.completed(103, new.token)
assert new.committable() == 110
assert BatchProgress([], 110, token="empty").committable() == 110

ここでは、100 が完了した後に 103 をコミットするのは安全です。この順序の読み取りでは、101、102 には配信すべきビジネスレコードがないことが確定しており、103 はまだ完了していないためです。すべて完了した後に 110 をコミットすると、読み取り位置を越えた制御レコードも跨ぐことができます。読み取りなしに推測で offset を跨ぐことはできません。実際のクライアントは、パーティションごとに読み取り後の next offset をキャプチャし、対応するバッチにバインドする必要があります。後続の poll の位置を使って古いバッチの境界を置き換えないようにします。位置の定義については KafkaConsumer.position を参照してください。

例では、103 の復旧位置が正常にコミットされた後に、新しいメンバーが 103 から引継ぎを行うと仮定しています。107 は古いメンバーによって実行済みですが、再実行される可能性があります。進捗追跡は漏れを防ぎますが、重複効果を消除するものではありません。このモデルは Kafka のコミットインターフェースを呼び出していません。committable() は候補値であり、コミットの成否は、そのパーティションを所有するコンシューマスレッドによって処理されます。ワックスレッドは完了キューを通じて結果を返却し、同じ KafkaConsumer に対して直接並列呼び出しを行うことはありません。

マルチバッチパイプラインでは、このモデルをパーティションごとの順序付きキューに拡張できますが、各バッチの読み取り境界、割り当てID、および未完了タスクを保持する必要があります。各 poll で直前に前のバッチの状態を上書きしてはいけません。

107 が完了したのに、なぜ直接 108 をコミットできないのか？

進捗バーは、最大の完了番号ではなく、配信済みタスクの順序に従う必要があります。以下では、まず 107 を完了し、次に 100 を完了します。コミット可能な位置は、まだ完了していない 103 で停止します。すべて完了してから、既知の next position 110 にジャンプします。候補水位と永続水位は別々に表示され、クラッシュ時は後者を使用して復旧するため、「完了済み」であっても再実行が必要になる場合があります。

図を準備しています

完了 107、なぜ直接 108 をコミットできないのか？

コミット可能な水位は、最も早くに未完了となった配信済みタスクで停止します。すべて完了してから、このバッチの next position をコミットできます。数字のギャップはタスクの欠落を意味しません。

自動コミットと手動コミット

自動コミットはクライアントメカニズムに基づいて定期的に消費位置を保存しますが、アプリケーションのスレッドプール内のビジネスタスクが完了したかどうかは認識しません。同期フェッチと処理の完了、およびフェッチ直後の非同期配信は、異なる使用パターンです。

手動コミットは制御を提供しますが、重複を自動的に消除しません。ビジネスの完了からコミットの成功までの間には、クラッシュのウィンドウが存在します。非同期コミットでは、リクエストの完了順序と失敗コールバックも処理する必要があり、失敗後に古い offset を盲目的に再実行して、更新された進捗を上書きしてはいけません。

メッセージの処理ができない場合、ビジネスロジックのポリシーを明確にする必要があります。そのパーティションをブロックして修復を待つのか、それとも失敗を永続的に記録して専用チャネルに転送するのかです。単純にスキップしてコミットし直すと、実質的に「すべてのメッセージが正常に処理された」という約束を変更することになります。具体的な API とスレッドの制限については KafkaConsumer Javadoc を参照してください。

ハートビート、処理タイムアウト、およびバッキングプレッシャー

「プロセスが生きていること」と「ビジネスが進んでいること」は同じではありません。ハートビートはメンバーの存続を判断するのに役立ちます。poll 間隔などの制約は、消費プロトコルにタイムリーに参加できないアプリケーションを検出するためにも使用されます。タイムアウトを大きくしすぎると、誤検知の遅延だけでなく、実際の故障後の引継ぎも遅らせることになります。

入力レートが秒間 1000 件、処理能力が秒間 800 件であると仮定します。バックログの増加は秒間約 200 件になります。10分間継続すると、約 12万件になります。クライアントのバッファを大きくするだけでは、この差は解消されず、バックログをブローカーからプロセスメモリへ移動させるだけです。

バッキングプレッシャーは、フェッチと処理能力を一致させる必要があります。進行中のタスク数を制限し、必要に応じてパーティションのフェッチを一時停止し、同時にグループプロトコルへの正常な参加を継続します。バッチサイズ、ワーカースレッド数、および poll 頻度は共同で設計する必要があります。順序を保持する必要があるパーティションでは、パーティション間で並列に、パーティション内で直列に実行できます。

一時停止の閾値には、すでに取得されたバッチの分を残す

ローカルで 1200 個のタスクを保持でき、一度に最大 200 個を受け入れることができると仮定します。キューが 1200 に達してから一時停止すると、現在返却済みの1バッチによって 1400 に達する可能性があります。進行中とキュー内の合計数が 1000 に達した時点で一時停止し、400 未満になったら再開するようにします。2つの閾値間の間隔は、タスクが完了するたびに頻繁に一時停止と再開を繰り返すのを防ぎます。

この予算は、配信前に容量をチェックし、最大 200 件の待機バッチが1つだけ存在することを前提としています。複数のプリフェッチバッチがある場合は、すべて予算に含める必要があります。件数ベースのレートリミティングは、バイトベースのメモリ予算と等価ではありません。200件の大きなメッセージと 200件の小さなメッセージでは、メモリコストが異なります。

Kafka の max.poll.records は、1回の poll で返されるレコード数を制限しますが、下位層の fetch で既にキャッシュされているデータは制限しません。したがって、タスクの上限はバイトの予算とともに設計する必要があり、一時停止中でもクライアントの要件に従って poll を呼び出し、調整に参加し続ける必要があります。ワークキューで消費スレッドをブロックしてメンバー資格を失わせることはできません。Consumer Configs：max.poll.records と max.poll.interval.ms を参照してください。

Lag はどのように解釈すべきか

一般的な lag は、あるログの境界とコミット済みの位置との差です。この数は傾向を観察するのに適していますが、常に処理待ちのビジネスレコード数と等しいとは限らず、レイテンシに直接換算できるわけでもありません。レコードのサイズ、圧縮による穴、パーティションのトラフィックはそれぞれ異なる可能性があります。

少なくとも同時に観察する必要があります：

各パーティションの lag と最も古い未処理イベントの年齢。合計値が単一パーティションのホットスポットを隠さないようにする。
フェッチ位置とコミット位置の距離。バックログがブローカー側にあるのか、アプリケーション内部にあるのかを判断する。
処理スループット、レイテンシ分布、進行中のタスク数。下流のボトルネックを特定する。
リバランスの頻度とパーティションの復旧時間。安定した低速消費と繰り返される引継ぎを区別する。

これらの量はそれぞれ、「残りどれだけか」「どれくらい待つか」「どこで詰まっているか」を記述します。指標とアラートのサービス目標と組み合わせることができます。

キューコンシューマーとの比較

RabbitMQ の従来のキューやアービトレーションキューは、メッセージの配信と確認に重点を置いていますが、Kafka の通常のコンシューマーグループは、ログパーティションの分担と再生に重点を置いています。ただし、RabbitMQ にも Streams があり、製品全体を「確認したら即削除、永遠に再生できない」と概括することはできません。RabbitMQ Streams を参照してください。

選択の際は、具体的なモデルを比較する必要があります。イベントごとに独立して確認するのか、それとも連続した位置でコミットするのか、履歴を保持するのか、完了後に解放するのか、複数のグループが独立して再生するのか、複数のワーカーがタスクを競合させるのか。Kafka はパーティション内の順序を提供しますが、トピック全体としてのグローバルな順序は提供しません。

参照

キーワード：consumer group、coordinator、rebalance、cooperative、offset、連続水位、バッキングプレッシャー、lag、fencing

滞留の解消とコミットの正しさを分けて検証する

コミット境界が正しくても滞留は増えます。容量と過負荷からの回復で到着仕事量、余力、再試行から追いつき時間を考え、計算機システムの制御で拡張遅延と振動を扱います。コンシューマーの追加が有効なのは、パーティション、外部ストレージ、処理能力の制約内です。永続コミットと重複排除の規則は別途必要です。

マイリーディング

シリーズで学びを始める

気になる記事をここに

続きはここから