消費グループと協調動作

consumer group は、パーティション所有権に対するコンシューマー間の合意形成であり、本質的には分散合意プロトコルの一種です。eager（強欲）リバランスと cooperative（協調的）リバランスの違いは、リバランス発生時にグループ全体が消費を停止するか継続するかを決定します。

概要

Kafka アーキテクチャでは、partition と ISR について解説しました。これらはストレージ・レプリケーション層における分散メカニズムです。しかし、コンシューマー側にも分散協調の問題があります。複数のコンシューマーが 1 つのトピックのパーティションを共有する場合、どのパーティションをどのコンシューマーが消費するのか？ 誰かがダウンした場合、どのように割り当てを再分配するのか？ 再分配中に消費は停止するのか？ これらの問題を解決するのが消費グループ（consumer group）⁠です。本質的には「一連のコンシューマーによるパーティション所有権の合意形成」であり、紛れもない分散合意プロトコルの一種です。

Consumer group の動作原理

Kafka の消費モデル：1 つの partition は、1 つの consumer group 内で 1 つの consumer によってのみ消費されます。 これが「同一 partition 内のメッセージ順序保証」の前提条件です。もし 2 つのコンシューマーが同時に同じ partition を読み取ると、それぞれが認識する offset 範囲が不連続になり、順序が乱れてしまいます。

重要：partition 数が並列処理の上限を決定します。6 つの partition → 最大 6 つの consumer。7 番目の consumer が参加するとアイドル状態になります（割り当てられる partition がないため）。⁠消費の並列度を拡張するには、consumer を追加するのではなく、partition を追加する必要があります。

リバランス（rebalance）: 所有権の合意形成

consumer がグループに参加または離脱すると、リバランスがトリガーされ、パーティションの所有権が再分配されます。これは consumer group メカニズムの中で最も重要かつ問題が発生しやすい部分です。

eager rebalance（旧プロトコル）: 停止して再分配

1. 新しい consumer C4 が参加 → group coordinator がすべての consumer に通知: 現在の割り当てを解放
2. すべての consumer が消費を停止し、partition 所有権を解放
3. 再分配: C1←P0,P1  C2←P2  C3←P3,P4  C4←P5
4. すべての consumer が最後にコミットした offset から消費を再開

このプロセスは stop-the-world と呼ばれます。ステップ 2-3 の間、consumer group 全体が消費を停止し、レイテンシが蓄積します。レイテンシに敏感なストリーム処理アプリケーションにとって、リバランスは每一次事故です。

cooperative rebalance（新プロトコル、KIP-429）: 漸進的

1. C4 が参加 → coordinator が新しい割当を計算し、「partition を解放する必要がある」consumer のみに通知
2. 影響を受けた consumer は、奪われた partition を解放し、影響を受けない partition の消費を継続
3. 新しい所有者（C4）が解放された partition を引き継ぐ
4. この間、影響を受けない consumer は停止しない

本質的な違い：eager は全員に解放させてから再分配しますが、cooperative は影響を受けた部分のみを変更し、他は停止しません。「すべて停止」から「一部停止しない」への変更により、リバランスによるビジネスへの影響を大幅に軽減します。

sticky assignment: 既存の割り当てを可能な限り維持

リバランス後の割り当ては「均等性」だけでなく「最小変更」も考慮されます。もしある consumer がすでに特定の partition を消費中で、解放する必要がない場合、⁠変更しません⁠。これを sticky と言います。無意味な割り当て変更を避け、consumer の状態遷移を減らします。

offset 管理: 消費進捗の「永続化ポインタ」

consumer がどこまで読んだかを記録する必要があります。これが offset です。管理方法は主に 2 つあります。

offset の保存場所⁠: Kafka の内部トピック __consumer_offsets（複数 partition、レプリケーションあり）。consumer group の offset コミットは、このトピックへの書き込みメッセージとして記録されます。つまり、Kafka は自前のストレージメカニズムを使って消費進捗を保存しています。

exactly-once 消費: 消費側トランザクションの統合

メッセージセマンティクスでは、producer 側の exactly-once（冪等 producer + トランザクション）について解説しました。消費側で exactly-once を実現するにはどうすればよいでしょうか？ 核心は、offset のコミットと処理結果の出力を同じトランザクションに含めることです。

消費者トランザクション:
  1. topic-A からメッセージを読み取る (offset=N)
  2. 処理 → 結果を topic-B に書き込む
  3. __consumer_offsets に対して offset(N+1) をコミット
  ステップ 2 と 3 は同じ Kafka トランザクション内で行われる:
    - トランザクション成功: 結果が topic-B に書き込まれ、offset が N+1 に進む → 消費が「完了した」とみなされる
    - トランザクション失敗/消費者クラッシュ: 結果が書き込まれず、offset が N にロールバック → メッセージが再消費される

重要な要件：消費側の isolation.level を read_committed に設定する必要があります。コミット済みトランザクションのメッセージのみを読み取り、abort されたトランザクションや未完成の dangling トランザクションを無視します。

この一連の組み合わせ（冪等 producer + トランザクション producer + read_committed consumer + offset と出力の同一トランザクション）により、Kafka の exactly-once セマンティクスが構成されます。これは理論上の「完璧な僅一度」ではなく、工学的に「僅一度と同等の効果を持ち、重複や消失が発生しない」ことを意味します。

RabbitMQ との比較: キューモデル vs ログモデル

同じ章で取り上げられた別のアプローチ：RabbitMQ はログではなくキューを使用します。メッセージは確認されると削除されます（不変ログへの追加ではないため）、リプレイや巻き戻し、partition 内のグローバル順序保証はありません。consumer group の協調方法も異なります。RabbitMQ の consumer は partition 所有権を合意形成するのではなく、broker が round-robin または consistent-hash exchange によってメッセージを配布します。これはよりシンプルですが、Kafka の「同一キーは同一 partition へという順序保証」のような機能はありません。

Kafka のログモデルが選ばれる理由は、分散ストリーム処理にリプレイ（replay）⁠とグローバル順序が必要だからです。これらはキューモデルでは実現できません。Kafka の partition は本質的に、シャーディングされ、不変で、リプレイ可能なログであり、consumer group はこのログの上で「誰がどのシャーディングを読むか」を合意形成する層です。

よくある問題と解決策

• リバランスストーム⁠: consumer の頻繁な参加/離脱（例：ハートビートタイムアウト閾値が短すぎる、処理時間が長すぎる → ダウンとみなされる）。毎回のリバランスでグループ全体が停止する。→ max.poll.interval.ms を十分に大きく設定する（最長処理時間より長く）、cooperative rebalance を使用する、リバランス頻度を監視する。
• partition 数 < consumer 数⁠: 余分な consumer がアイドル状態になる → partition を追加するか、アイドル状態を受け入れる（consumer をホットスタンバイとして使用する）。
• 消費遅延（lag）: consumer の処理速度が producer の書き込み速度に追いつかない → lag を監視する（consumer offset vs partition end offset）、consumer または partition を追加する。
• 自動コミット + 非同期処理⁠: consumer がメッセージをスレッドプールに投げるとすぐに offset をコミットする → 非同期スレッドで処理に失敗した場合、メッセージはすでに「消費済み」とみなされる → 消失する。→ 同期処理を行うか、非同期処理完了後に手動コミットを行う。

参考

• Kafka ドキュメント⁠: Consumer Group Protocol、KIP-429 (Incremental Cooperative Rebalancing)

Keywords: consumer group, rebalance, eager, cooperative (incremental), sticky assignment, stop-the-world, group coordinator, offset management, __consumer_offsets, exactly-once consumption, isolation.level, read_committed, transactional consumer, 消費遅延, consumer lag, rebalance storm, Kafka vs RabbitMQ, log vs queue