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

コンシューマーグループは、パーティション所有権に対するコンシューマー間の合意形成であり、本質的には分散合意プロトコルの一種です。イグリーvs協調的なリバランスは、リバランス時にグループ全体が消費を停止するか継続するかを決定します。

概要

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

コンシューマーグループの動作原理

Kafkaの消費モデル：1つのパーティションは、1つのコンシューマーグループ内では1つのコンシューマーのみが消費できます。 これが「パーティション内のメッセージ順序が保証される」ための前提条件です。もし2つのコンシューマーが同時に同じパーティションを読み取ると、それぞれが認識するオフセットの範囲が不連続になり、順序が乱れてしまいます。

重要なのは、パーティション数が並列処理の上限を決定することです。パーティションが6つある場合、最大で6つのコンシューマーが利用可能です。7番目のコンシューマーが参加しても、割り当てられるパーティションがないためアイドル状態になります。⁠消費の並列度を拡張するには、コンシューマーを追加するのではなく、パーティションを追加する必要があります。

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

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

イグリーリバランス（旧プロトコル）: 停止して再分配

1. 新しいコンシューマー C4 が参加 → グループコーディネーターが全コンシューマーに通知: 現在の割り当てを解放
2. 全コンシューマーが消費を停止し、パーティションの所有権を解放
3. 再分配: C1←P0,P1, C2←P2, C3←P3,P4, C4←P5
4. 全コンシューマーが最後にコミットしたオフセットから消費を再開

このプロセスは stop-the-world（ワールドストップ） と呼ばれます。ステップ2-3の間、⁠コンシューマーグループ全体が消費を停止し、レイテンシーが蓄積します。レイテンシーに敏感なストリーム処理アプリケーションにとって、リバランスは毎回の事故です。

協調的リバランス（新プロトコル、KIP-429）: 漸進的

1. C4 が参加 → コーディネーターが新しい割り当てを計算し、「パーティションを解放する必要がある」コンシューマーのみ通知
2. 影響を受けたコンシューマーは、奪われたパーティションを解放し、影響を受けないパーティションの消費を継続
3. 新しい所有者（C4）が解放されたパーティションを引き受け
4. この間、影響を受けないコンシューマーは停止しない

本質的な違い: イグリーでは全員が解放してから再分配を行うのに対し、協調的では影響を受けたもののみを変更し、他は停止しません。「すべて停止」から「一部は停止しない」への変化により、リバランスによるビジネスへの影響を大幅に軽減します。

スティッキー割り当て: 既存の割り当てを維持する

リバランス後の割り当ては「均等性」だけでなく「最小変更」も考慮されます。もしあるコンシューマーが既に特定のパーティションを消費中で、解放する必要がない場合、その割り当ては変更されません⁠。これをスティッキー（sticky）⁠と呼び、不要な割り当て変更を避け、コンシューマーの状態遷移を減らします。

オフセット管理: 消費進捗の「永続化ポインター」

コンシューマーがどこまで読んだかを記録する必要があります。これがオフセットです。管理方法には2つの種類があります。

オフセットの保存場所⁠: Kafkaの内部トピック __consumer_offsets（マルチパーティション、レプリケーションあり）です。コンシューマーグループのオフセットコミットは、このトピックへの書き込みメッセージとして記録されます。つまり、Kafkaは自前のストレージメカニズムを使用して消費進捗を保存しています。

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

メッセージセマンティクスでは、プロデューサー側の exactly-once（冪等プロデューサー + トランザクション）について解説しました。消費側で exactly-once を実現するには、⁠オフセットのコミットと処理結果の出力を同じトランザクションに含めることが核心です。

コンシューマートランザクション:
  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 に設定する必要があります。これにより、コミット済みトランザクションのメッセージのみを読み取り、アボートされたトランザクションや未完成のdanglingトランザクションを無視します。

この一連の組み合わせ（冪等プロデューサー + トランザクションプロデューサー + read_committed コンシューマー + オフセットと出力が同じトランザクション内）により、Kafkaの exactly-once セマンティクスが構成されます。これは理論上の「完璧な僅か1回の処理」ではなく、工学的に「僅か1回の処理と同等の効果を持ち、重複も欠落も発生しない」ことを意味します。

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

同じ章で解説されている別のアプローチ: RabbitMQはログではなくキューを使用します。メッセージは確認されると削除されます（不変ログへの追加ではないため）、リプレイ/巻き戻しはできず、パーティション内のグローバル順序も保証されません。コンシューマーグループの協調も異なります。RabbitMQのコンシューマーはパーティション所有権を合意形成するのではなく、ブローカーがラウンドロビンまたは一貫性ハッシュ交換によってメッセージを配信します。これはよりシンプルですが、Kafkaのような「同じキーは同じパーティションへ」という順序保証はありません。

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

よくある問題と解決策

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

参考

• 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