Kafka Consumer Group の仕組みを実務目線で整理する：並列処理とグループ調整の勘所

Consumer Group は Kafka のスケールアウトと耐障害性の中核です。並列処理の上限はパーティションで決まり、グループ調整（再バランス）が可用性とレイテンシを左右します。

本稿は CCDAK の出題観点を踏まえつつ、運用で効く設定と挙動の境界条件を、公式ドキュメント準拠の安定した概念に絞って整理します。

1. 並列処理の基本：パーティションと Consumer Group の関係

Consumer Group は同一 group.id の複数コンシューマで構成され、同一トピックの各パーティションはグループ内のいずれか1コンシューマにのみ割り当てられます。したがって、最大の並列度は「グループ全体で割り当てられるパーティション数」によって上限が決まります。

順序保証はパーティション単位です。グループで水平スケールしても、単一パーティション内の順序は保持されますが、パーティションをまたぐ全体順序は保証されません。必要な並列度は事前にパーティション設計で確保します。コンシューマはスレッドセーフではないため、基本は1スレッド=1インスタンスで運用し、内部での並列化が必要ならワーカーキュー方式などでメッセージ処理を分離します。

• 最大並列度 ≒ min(コンシューマ数, 合計パーティション数)
• 1パーティションは同一グループ内の1コンシューマに専有割当
• 順序保証はパーティション単位。グループ全体の順序は保証されない
• コンシューマは基本的にスレッド非安全。1スレッド1インスタンスが原則

Consumer Group とパーティション割り当て（概念図）

2. グループ調整（再バランス）の仕組みと発火条件

Consumer Group はブローカー上の Group Coordinator により管理され、メンバーは heartbeats を送信して存続を示します。セッション切れや構成変更が起きると再バランスが走り、パーティション割当が見直されます。再バランス中は一時的にフェッチ/処理が停止しうるため、頻度と時間の最適化がスループットとレイテンシの鍵です。

主な発火要因は、メンバーの参加/離脱、サブスクライブトピック集合の変更、トピックのパーティション数変更、セッションタイムアウト/最大ポール間隔超過などです。設定は session.timeout.ms、heartbeat.interval.ms、max.poll.interval.ms の整合と、割当戦略の選択が中心です。

• 再バランス発火: 参加/離脱、subscribe 変更、パーティション数変更、セッション切れ、max.poll.interval.ms 超過
• 監視優先: 再バランス回数、平均/最大所要時間、ハートビート遅延
• 安定化の要: Cooperative 戦略の採用、静的メンバーシップ、適切なタイムアウト

3. 割当戦略と安定化チューニング

割当戦略は partition.assignment.strategy にリスト形式で指定します。RangeAssignor、RoundRobinAssignor、StickyAssignor、CooperativeStickyAssignor などが代表例です。実運用では CooperativeStickyAssignor による増分再バランスが広く推奨されますが、デフォルトはクライアントやディストリビューションにより異なるため、明示設定が安全です（バージョン差異を避ける目的）。

静的メンバーシップ（group.instance.id の設定）は、同一メンバーが再起動しても同一 ID と見なされるため、ローリングリスタート時の再バランスを大幅に抑制できます。タイムアウト群は相互関係を理解して設定し、max.poll.interval.ms 超過による不要な離脱を防ぎます。

• partition.assignment.strategy: CooperativeStickyAssignor を先頭に明示（推奨）
• group.instance.id の付与で静的メンバー化（ローリング時の安定化）
• session.timeout.ms と heartbeat.interval.ms は比率を維持（例: heartbeat は session の 1/3 程度）
• max.poll.interval.ms は処理時間の上限に見合うだけ十分に長くする

4. オフセット管理と配信セマンティクス

コンシューマのオフセットはデフォルトで内部トピック __consumer_offsets にコミットされます。enable.auto.commit を使うと定期的に自動コミットされますが、処理の成否と無関係に進むため、少なくとも重要処理では手動コミット（commitSync/commitAsync）を検討します。

配信セマンティクスはコミット順序に依存します。処理の後でコミットすれば少なくとも一度（at-least-once）。処理の前にコミットすれば高々一度（at-most-once）ですがロスの可能性。厳密な厳密一度（exactly-once）は Kafka のトランザクションと組み合わせたプロデューサ/コンシューマ連携が必要で、設計単位が広がります。

• auto-commit は便利だが再処理/重複の制御が難しい
• at-least-once: 処理完了後にオフセットコミット（一般的で安全側）
• at-most-once: フェッチ直後にコミット（ロス容認のユースケース限定）
• トランザクション利用で read-process-write の EOS を実現可能（設計コスト増）

5. スケールとスループット最適化のパターン

並列度を上げるには、原則としてパーティション数を増やすか、グループ内のコンシューマ数を増やします。後者は前者の上限に縛られるため、計画段階で所要のパーティション数を見積もることが重要です。過剰なパーティションはメタデータ/レプリケーション負荷を増やすため、根拠あるスケール設計が必要です。

フェッチ/処理の最適化として、max.poll.records によるバッチ処理、fetch.min.bytes と fetch.max.wait.ms による帯域効率化、pause/resume による一時的なバックプレッシャ制御などが有効です。処理スレッドは内部キューで水平分散し、オフセットコミットは処理完了基準で管理します。

• スケール設計はパーティション数が主軸。後からの増加はキーベースの局所性に影響しうる
• max.poll.records でバッチ処理、スループットとレイテンシのトレードオフを調整
• pause/resume で一時的に取り込み削減。過負荷時のスロットルに有効
• client.rack を使うと近傍レプリカ優先フェッチが可能（ネットワーク効率）

6. 運用の要点：監視と安全なローリングリスタート

運用では、再バランス指標（回数・所要時間）、lag、フェッチレイテンシ、ハートビート遅延を可視化します。障害対応やデプロイ時は、静的メンバーシップと Cooperative 戦略を併用し、リリース単位の影響を最小化します。

ローリングリスタート時は 1 台ずつ停止・復帰し、メンバーの復帰を待ってから次に進みます。静的メンバーが維持されていれば、割当の再計算は局所化され、中断は最小限で済みます。CLI の kafka-consumer-groups --describe や --reset-offsets は状態把握・修正に有用です。

• 監視: group lag、rebalance 回数/時間、heartbeat、fetch 指標
• 手順: 静的メンバー + Cooperative で段階的移行、1 台ずつローリング
• ツール: kafka-consumer-groups --describe / --reset-offsets の正しい使い方

Cooperative + 静的メンバーの Java Consumer（要点抜粋）

Properties props = new Properties();
props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, "broker1:9092,broker2:9092");
props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());
props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());
props.put(ConsumerConfig.GROUP_ID_CONFIG, "orders-consumer-g");
// Cooperative Sticky を明示（順序は優先度）
props.put(ConsumerConfig.PARTITION_ASSIGNMENT_STRATEGY_CONFIG,
         "org.apache.kafka.clients.consumer.CooperativeStickyAssignor");
// 静的メンバーシップ（各インスタンスで一意な ID を設定）
props.put(ConsumerConfig.GROUP_INSTANCE_ID_CONFIG, System.getenv("POD_NAME"));
// タイムアウトは処理時間とネットワーク特性に合わせて整合
props.put(ConsumerConfig.SESSION_TIMEOUT_MS_CONFIG, "15000");
props.put(ConsumerConfig.HEARTBEAT_INTERVAL_MS_CONFIG, "3000");
props.put(ConsumerConfig.MAX_POLL_INTERVAL_MS_CONFIG, "600000");
props.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, "false");

KafkaConsumer<String, String> consumer = new KafkaConsumer<>(props);

CountDownLatch latch = new CountDownLatch(1);

ConsumerRebalanceListener listener = new ConsumerRebalanceListener() {
  @Override
  public void onPartitionsRevoked(Collection<TopicPartition> partitions) {
    // 処理済みのオフセットを確実にコミット（at-least-once）
    try { consumer.commitSync(); } catch (Exception e) { /* ログ */ }
  }
  @Override
  public void onPartitionsAssigned(Collection<TopicPartition> partitions) {
    // 必要に応じてシークや初期化
  }
};

consumer.subscribe(Arrays.asList("orders"), listener);

try {
  while (true) {
    ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(500));
    for (ConsumerRecord<String, String> r : records) {
      process(r); // アプリ固有処理（例外は握りつぶさず、DLQ 等へ）
    }
    consumer.commitAsync(); // 通常は非同期、終了/リボーク前は同期
  }
} finally {
  try { consumer.commitSync(); } catch (Exception ignore) {}
  consumer.close();
}

問題で確認

よくある質問