Kafka Cooperative Rebalance: Stop-the-world を避ける最新プロトコル

従来の Eager Rebalance は全コンシューマーが一斉にパーティションを手放すため、短時間でも“全停止”が発生しやすく、遅延やスループット低下の原因になっていました。

Cooperative Rebalance は割り当てを段階的・増分で移し替えることで Stop-the-world を避ける最新のプロトコルです。ここでは仕組み、導入手順、コールバック設計、運用の勘所を CCDAK 受験にも役立つ視点で解説します。

なぜ Stop-the-world が起きるのか（Eager Rebalance の限界）

Eager Rebalance では、メンバーの増減や設定変更をきっかけに、グループ内の全コンシューマーが一度すべてのパーティションを手放し、再割り当てを待ちます。短時間でも消費停止が全体に波及するため、レイテンシ尖りや処理スループットの落ち込みを招きます。

実務ではローリングデプロイ、スケールアウト、ネットワーク揺らぎ等が頻発し、再均衡の回数が増えがちです。Stop-the-world を抑えることは、そのまま SLO の安定やピーク時の耐性に直結します。

• 全メンバーが一斉に revoke → assign されるのが Eager の本質的ボトルネック
• 長時間処理とコミットのズレがあると重複処理や遅延が目立つ
• 再均衡のたびにキャッシュ温め直しや接続張り直しが発生しやすい

Eager Rebalance（従来）の典型設定例

props.put("partition.assignment.strategy", java.util.List.of(
  "org.apache.kafka.clients.consumer.RangeAssignor"
));
// デフォルトは RangeAssignor（Eager）で、全メンバーが一旦 revoke される

Cooperative Rebalance の仕組み（増分・段階的な再均衡）

Incremental Cooperative Rebalancing は、必要最小限のパーティションのみを段階的に移動します。既存メンバーの大半は割り当てを維持したまま、移動対象だけを revoke/assign するため、Stop-the-world を避けられます。

コンシューマー側は ConsumerRebalanceListener のコールバックで、revoke されたものだけを安全にコミット・解放し、assign されたものを消費開始します。グレースフルに revoke できなかった場合のために onPartitionsLost が呼ばれるケースを考慮します。

• 段階的に移行するため、全体停止が起きにくい
• onPartitionsRevoked は“本当に外れるパーティションだけ”が渡る
• onPartitionsLost は異常時の最終手段（未コミット処理の扱いに注意）
• Sticky かつ Cooperative により、パーティションの移動量を最小化

増分的な移動イメージ（Cooperative Rebalance）

Cooperative Sticky Assignor の有効化

props.put("partition.assignment.strategy", java.util.List.of(
  "org.apache.kafka.clients.consumer.CooperativeStickyAssignor"
));
// 可能なら全メンバーで同一戦略に統一すること

安全な導入手順（ローリング切替の勘所）

同一コンシューマーグループ内では、実効的に選ばれるアサイナーは“共通集合の先頭”になります。混在期間は予期せぬフォールバック（Eager 化）を招くため、切替は計画的に行います。

最も安全なのは、全メンバーを CooperativeStickyAssignor をサポートするバージョンへ先に上げ、その後に戦略を統一するか、新しい group.id へ移行する方法です。

• ローリング時は一時的に Eager 挙動へ戻る可能性がある点に注意
• group.instance.id（静的メンバーシップ）併用でリバランス頻度をさらに低減
• 互換性不明な混載は避け、観測しながら段階的に切替える

安全な切替（例: 新 group.id を用いた段階移行）

# 1) 先にアプリを CooperativeStickyAssignor 対応版へデプロイ
# 2) 新しい group.id で起動し、旧グループから段階的にトラフィック移行
props.put("group.id", "orders-consumer-v2");
props.put("partition.assignment.strategy", java.util.List.of(
  "org.apache.kafka.clients.consumer.CooperativeStickyAssignor"
));

コールバックとコミット設計（実装の落とし穴）

Cooperative では revoke されるパーティションだけを整然と閉じられます。onPartitionsRevoked 内では、対象パーティションの未処理を flush/commit し、外部リソースを解放します。onPartitionsAssigned では新規割り当ての初期化を軽量に行います。

onPartitionsLost は異常系（グレースフルに返却できなかった）で呼ばれます。ここでは“未処理が確実に再処理されうる”前提で、状態を破棄・再初期化する方が安全です。

• 同期コミットかつ対象パーティションのみコミット（不要な全体コミット回避）
• 長時間処理は max.poll.interval.ms を十分に確保
• 外部ストア（DB/キャッシュ）のハンドルはパーティション単位で管理すると楽

ConsumerRebalanceListener の骨子（Java）

consumer.subscribe(List.of("orders"), new ConsumerRebalanceListener() {
  @Override
  public void onPartitionsRevoked(Collection<TopicPartition> partitions) {
    // revoke 対象だけを同期コミット・クローズ
    commitOffsetsFor(partitions);
    closeResourcesFor(partitions);
  }
  @Override
  public void onPartitionsAssigned(Collection<TopicPartition> partitions) {
    // 必要最小の初期化。重い処理は避ける
    initStateFor(partitions);
  }
  @Override
  public void onPartitionsLost(Collection<TopicPartition> partitions) {
    // 異常系。未処理は再処理前提で状態を破棄
    dropStateFor(partitions);
  }
});

運用チューニングと監視ポイント

Stop-the-world を避けられても、再均衡そのものは発生します。タイムアウトやハートビートはアプリの処理時間と整合させます。特に max.poll.interval.ms は“1回の poll で処理できる最大時間”に見合う値が必要です。

監視では consumer lag、rebalancing の発生回数・時間、onPartitionsLost の有無に着目します。静的メンバーシップ（group.instance.id）や cooperative の採用で、グループの churn を最小化します。

• session.timeout.ms と heartbeat.interval.ms の整合（セッション切れを回避）
• max.poll.interval.ms はワークロード最長処理時間 + 余裕
• 長い GC や外部 API 待ちが続くと離脱扱いになり得る点に注意
• kafka-consumer-groups.sh で lag と安定性を定期点検

監視・点検コマンド例

$ kafka-consumer-groups.sh \
  --bootstrap-server broker:9092 \
  --group orders-consumer \
  --describe

# 再均衡の兆候はアプリログ（JoinGroup/SyncGroup）やメトリクスで確認

CCDAK 視点の要点と出題パターン

CCDAK では、Eager と Cooperative の違い、CooperativeStickyAssignor の利点、関連コールバック（特に onPartitionsLost）、静的メンバーシップ併用による安定化といった論点が頻出です。

設定の優先順位や混在時の挙動、ローリングデプロイ時の安全策、コミット設計の原則（重複許容・順序性）を、正確な概念として押さえておくと得点に直結します。

• キーワード: CooperativeStickyAssignor, Incremental Rebalancing, onPartitionsLost, group.instance.id
• Eager は一斉 revoke、Cooperative は最小限 revoke という対比を即答できること
• 導入手順の安全策（全メンバー更新 → 戦略統一 or 新 group.id 移行）

試験で問われやすい設定キー（抜粋）

# コンシューマー設定（代表例）
partition.assignment.strategy=org.apache.kafka.clients.consumer.CooperativeStickyAssignor
max.poll.interval.ms=900000
session.timeout.ms=45000
heartbeat.interval.ms=15000
group.instance.id=orders-c1   # 静的メンバーの例（ユニークに）

問題で確認

よくある質問