KafkaのConsumer GroupとRebalance: Consumerが停止する理由

1. 概要

Kafka からメッセージを読むサービスを持っており、何日間も安定して動作していました。ある日、Consumer ラグが急増してしまいました。ログは Revoking previously assigned partitions（割り当てられたパーティションを取り消し中）と CommitFailedException で埋まっています。犯人はたいてい、Consumer Group の Rebalance です。

この記事では、Consumer Group がパーティションをどのように割り当てるのか、Rebalance がいつ発生するのか、そしてなぜ Consumer が「フリーズ」するのかを説明します。読み終わった後、Spring Kafka を設定して Rebalance の回数を減らし、各停止時間を短縮する方法がわかるようになります。

2. Consumer Group のしくみ

Consumer Group は group.id を共有する Consumer の集合です。Kafka はトピックのパーティションを Group 内のメンバーに割り当てます。各パーティションは、ある時点で 1 つの Consumer にのみ割り当てられます。

orders というトピックに 6 つのパーティションがあると仮定します。order-service の 3 つのインスタンスを実行する場合：

order-service-1  -> orders-0, orders-1
order-service-2  -> orders-2, orders-3
order-service-3  -> orders-4, orders-5

7 つのインスタンスを実行する場合、割り当てるパーティションがないため、1 つのインスタンスはアイドル状態になります。

まず、依存関係を追加します：

<dependency>
    <groupId>org.springframework.kafka</groupId>
    <artifactId>spring-kafka</artifactId>
</dependency>

Spring Boot 3.x の簡単なリスナー例：

@Component
public class OrderListener {

    private static final Logger log = LoggerFactory.getLogger(OrderListener.class);

    @KafkaListener(topics = "orders", groupId = "order-service")
    public void onOrder(ConsumerRecord<String, String> record) {
        log.info("注文 {} をパーティション {} から受け取りました", record.key(), record.partition());
    }
}

Annotation @KafkaListener はリスナーコンテナを作成します。コンテナ内には poll() ループを実行するスレッドがあります。groupId = "order-service" を使用するすべてのインスタンスは同じ Group に参加します。1 つのブローカーが Group Coordinator として機能し、メンバーリストとパーティション割り当てを管理します。

3. Rebalance とは何か、いつ発生するのか

Rebalance は Group Coordinator がパーティションをメンバーに再割り当てするプロセスです。Rebalance を引き起こす状況は以下の通りです：

新しい Consumer が参加する：スケールアップ、新しいインスタンスのデプロイ。
Consumer が主動的に Group を離れる：シャットダウン、ローリング再起動。
Consumer が session.timeout.ms 内に Heartbeat を送信しない場合、死んでいると判断される。デフォルトは 45 秒です。
Consumer が max.poll.interval.ms 内に poll() を呼び出さない場合。デフォルトは 5 分です。
トピックのパーティション数が変更される、または Pattern ベースのサブスクリプションが新しいトピックにマッチします。

項目 3 と 4 の 2 つのタイムアウトメカニズムは異なり、よく混同されます：

設定	デフォルト	チェック内容
`heartbeat.interval.ms`	3000	Heartbeat 送信の周期（バックグラウンドスレッドから）
`session.timeout.ms`	45000	プロセスが生きているかどうか
`max.poll.interval.ms`	300000	メッセージ処理スレッドが詰まっていないかどうか

Heartbeat はバックグラウンドスレッドから送信されます。そのため、メッセージ処理スレッドが詰まっている場合でもプロセスは「生きている」とみなされます。逆に、max.poll.interval.ms は、メッセージ処理コードが遅すぎる場合を正確に検出します。

4. Consumer が停止する理由

4.1. Eager Rebalance：グループ全体の停止

従来のプロトコルと eager 系の assignor（RangeAssignor、RoundRobinAssignor、StickyAssignor）では、Rebalance は以下のように進行します：

すべての Consumer が保有するパーティションを 完全に取り消します。
すべての Consumer が JoinGroup リクエストを再送信し、互いに待ち合わせます。
Group のリーダーが新しい割り当てプランを計算し、Coordinator が結果を返送します。
Consumer は新しいパーティションを受け取り、処理を再開します。

ステップ 1～3 の実行中、Group 内の誰もメッセージを処理しません。新しいインスタンスを 1 つ起動するだけで、グループ全体が停止します。これを「stop-the-world」現象と呼びます。

4.2. 処理が遅すぎる：`max.poll.interval.ms` オーバー

実務で最も一般的な原因は、コード処理が遅いことです。以下の例は、外部サービスを呼び出すのに 1 秒かかります：

@KafkaListener(topics = "orders", groupId = "order-service")
public void onOrder(ConsumerRecord<String, String> record) throws InterruptedException {
    paymentService.charge(record.value()); // 約 1 秒かかると仮定
}

デフォルトは max.poll.records = 500 です。1 回の poll() は 500 レコードを返す可能性があり、処理時間は約 500 秒になります。この数は max.poll.interval.ms の 300 秒を超えています。この場合、Consumer は自動的に Group を離れ、次のようなログが表示されます：

[Consumer clientId=consumer-order-service-1, groupId=order-service] consumer poll timeout has expired.
This means the time between subsequent calls to poll() was longer than the configured
max.poll.interval.ms, which typically implies that the poll loop is spending too much time
processing messages. You can address this either by increasing max.poll.interval.ms or by
reducing the maximum size of batches returned in poll() with max.poll.records.

その後、オフセットのコミットに失敗します：

org.apache.kafka.clients.consumer.CommitFailedException: Offset commit cannot be completed
since the consumer is not part of an active group for auto partition assignment; it is likely
that the consumer was kicked out of the group.

その結果、連鎖的に問題が発生します：

この Consumer のパーティションは他の誰かに割り当てられます。
コミットされていないオフセットにより、パーティションを受け取った側がこれらのメッセージを 再処理 します。
古い Consumer が再度参加し、さらに別の Rebalance をトリガーします。
次のバッチも遅い場合、ループが繰り返されます。ラグは増え続けます。

5. 修正方法

5.1. バッチサイズと poll 時間の調整

覚えておくべきルール：max.poll.records × 1 レコードあたりの処理時間 は max.poll.interval.ms より十分に小さい必要があります。

spring:
  kafka:
    consumer:
      group-id: order-service
      max-poll-records: 50
      properties:
        max.poll.interval.ms: 120000

50 レコード、1 レコードあたり 1 秒の場合、1 バッチは約 50 秒かかります。この数は 120 秒の制限内に安全に収まります。

注意： max.poll.interval.ms を数時間に増やして「消火」しないでください。この値は Kafka が真に詰まっている Consumer を検出する最大時間でもあります。さらに、Rebalance は最も遅い Consumer が poll() を呼び出すまで待つ必要があるため、大きな値は Rebalance を長くします。

各メッセージの処理時間が本来長い場合、例えば遅い API を呼び出す場合は、外部呼び出しにタイムアウトを設定する必要があります。もう 1 つの方法は、メッセージを内部キューにプッシュして、poll スレッドを長く保持しないようにすることです。

5.2. Cooperative Rebalancing

CooperativeStickyAssignor は incremental rebalance を使用します。Consumer は実際に移動する必要があるパーティションのみを取り消し、その他のパーティションは通常通り処理を続けます。

spring:
  kafka:
    consumer:
      properties:
        partition.assignment.strategy: org.apache.kafka.clients.consumer.CooperativeStickyAssignor

セクション 2 の 3 インスタンスグループに order-service-4 を追加する場合、いくつかのパーティションのみが新しいインスタンスに転送されます：

order-service-1  -> orders-0, orders-1        (変わらず、停止なし)
order-service-2  -> orders-2                  (orders-3 を失う)
order-service-3  -> orders-4                  (orders-5 を失う)
order-service-4  -> orders-3, orders-5

注意： 実行中の Group で eager から cooperative に直接変更することはできません。2 回のローリングデプロイが必要です。1 回目は両方を設定します：CooperativeStickyAssignor,RangeAssignor。2 回目は RangeAssignor を削除します。メンバーが共通の assignor を持たない場合、Group はエラーを報告します。

5.3. Static Membership

ローリング再起動のたびに、1 つのインスタンスが Group を離れて戻ります。その結果、各 Pod ごとに 2 回の Rebalance が発生します。Static Membership はこの問題を解決します。group.instance.id を使用して各インスタンスに固定識別子を割り当てます：

spring:
  kafka:
    consumer:
      properties:
        group.instance.id: ${HOSTNAME}
        session.timeout.ms: 60000

group.instance.id を持つ Consumer はシャットダウン時に LeaveGroup を送信しません。session.timeout.ms 内に同じ id で戻った場合、Coordinator は以前のパーティションを返す際に Rebalance を実施しません。

注意： HOSTNAME は再起動を通じて安定している必要があります。例えば、Kubernetes の StatefulSet のポッド名。通常の Deployment では、ポッド名は毎回変わるため Static Membership は効果がありません。同じ id で同時に 2 つのインスタンスが実行されると、互いを Group から追い出します（FencedInstanceIdException）。concurrency > 1 の場合、Spring Kafka は各 Consumer サブスレッドに -0、-1 などのサフィックスを自動的に追加します。

このアプローチのトレードオフ：インスタンスが真に死亡した場合、そのパーティションは session.timeout.ms が経過するまで処理されません。

5.4. 新しい Consumer Group プロトコル

Kafka 4.0 以降、新しい Rebalance プロトコル（KIP-848）は本番環境に対応しています。group.protocol=consumer を使用して有効にします。パーティション割り当てはブローカーに完全に移行し、常に incremental 方式で実行されます。Consumer は共通の JoinGroup 「バリア」で互いに待つ必要がありません。

このプロトコルでは、クライアント側の partition.assignment.strategy は使用されなくなります。代わりに、ブローカーが assignor を選択します（クライアント側の設定は group.remote.assignor）。クラスターとクライアントが十分に新しい場合、これは検討する価値があります。有効にする前に、Spring Boot が取得している kafka-clients のバージョンを確認する必要があります。

6. パーティション取り消し時の処理

どれだけ最適化しても、Rebalance は発生します。重要なことは、パーティションを失う前に処理済みのオフセットをコミットし、重複処理を減らすことです。

Spring Kafka はパーティションが取り消される前に、待機中のオフセットを自動的にコミットします。KafkaConsumer を直接使用する場合は、ConsumerRebalanceListener を使用して手動で行う必要があります：

public class CommitOnRevokeListener implements ConsumerRebalanceListener {

    private final Consumer<String, String> consumer;
    private final Map<TopicPartition, OffsetAndMetadata> pending = new ConcurrentHashMap<>();

    public CommitOnRevokeListener(Consumer<String, String> consumer) {
        this.consumer = consumer;
    }

    public void markProcessed(ConsumerRecord<String, String> record) {
        pending.put(new TopicPartition(record.topic(), record.partition()),
                new OffsetAndMetadata(record.offset() + 1));
    }

    @Override
    public void onPartitionsRevoked(Collection<TopicPartition> partitions) {
        Map<TopicPartition, OffsetAndMetadata> toCommit = new HashMap<>();
        for (TopicPartition tp : partitions) {
            OffsetAndMetadata offset = pending.remove(tp);
            if (offset != null) {
                toCommit.put(tp, offset);
            }
        }
        if (!toCommit.isEmpty()) {
            consumer.commitSync(toCommit);
        }
    }

    @Override
    public void onPartitionsAssigned(Collection<TopicPartition> partitions) {
        // 新しいパーティションのローカル状態をここで再度ロードすることができます
    }
}

各部分の説明：

markProcessed() は、レコード処理後に読み込む次のオフセット（つまり offset + 1）を記録します。
onPartitionsRevoked() は poll スレッド上で実行され、パーティションが取られる直前に実行されます。取り消されるパーティションのオフセットのみをコミットします。これは cooperative rebalancing にとって重要です。この場合、パーティションの一部のみが取り消されるためです。
commitSync() はブローカーが確認するまでブロックし、別の Consumer がパーティションを受け取る前にオフセットが保存されることを保証します。

kafka-clients に含まれる MockConsumer を使用してこの動作をテストできます：

class CommitOnRevokeListenerTest {

    @Test
    void shouldCommitOnlyRevokedPartitions() {
        MockConsumer<String, String> consumer = new MockConsumer<>(OffsetResetStrategy.EARLIEST);
        TopicPartition p0 = new TopicPartition("orders", 0);
        TopicPartition p1 = new TopicPartition("orders", 1);
        consumer.assign(List.of(p0, p1));

        CommitOnRevokeListener listener = new CommitOnRevokeListener(consumer);
        listener.markProcessed(new ConsumerRecord<>("orders", 0, 41L, "order-1", "{}"));
        listener.markProcessed(new ConsumerRecord<>("orders", 1, 7L, "order-2", "{}"));

        listener.onPartitionsRevoked(List.of(p0));

        Map<TopicPartition, OffsetAndMetadata> committed = consumer.committed(Set.of(p0, p1));
        assertThat(committed.get(p0).offset()).isEqualTo(42L);
        assertThat(committed.get(p1)).isNull();
    }
}

テストは orders-0 のみが取り消されるシナリオをシミュレートします。期待される結果は、orders-0 にオフセット 42 がコミットされることです。orders-1 はまだ保有されているため、オフセットはまだコミットされていません。

注意： onPartitionsRevoked() 内のコードは高速である必要があります。このコールバックは Rebalance 中に実行されます。遅い場合、グループ全体がそれを待たなければなりません。

7. よくあるエラー

非同期処理だが poll に従ってコミットしている。 レコードを ExecutorService にプッシュして、コンテナが poll() 直後にオフセットをコミットします。Rebalance が発生すると、処理中のメッセージが失われる可能性があります。処理完了したレコードのオフセットのみをコミットしてください。
session.timeout.ms と max.poll.interval.ms を混同している。 Session タイムアウトを増やしても、原因が遅い処理である場合は役に立ちません。
concurrency をパーティション数より大きく設定している。 余った Consumer は何もすることがなく、Group を大きくするだけで Rebalance を長くします。
コンテナの GC 一時停止が長い、または CPU が制限されている。 Heartbeat スレッドが間に合わず、コードが遅くなくても Consumer が Group から追い出されます。JVM メトリクスと CPU 制限を確認してください。
処理が冪等性を持つように設計されていない。 Rebalance は常に一部のメッセージの重複処理を引き起こす可能性があります。トランザクションを使用していない、または orderId が既に処理されたかどうかを確認するなどの重複排除メカニズムがない場合は、「exactly once」を想定しないでください。

8. まとめ

Rebalance は Consumer Group が自動スケール・フォールトトレランスする仕組みとして必要です。その代償として Consumer は一時停止する必要があります。Consumer が継続的に「停止」する最も一般的な原因は、poll() ループが max.poll.interval.ms を超えることです。この場合、Consumer は Group から追い出され、コミットが失敗し、Rebalance が繰り返されます。

これを最小限に抑えるには、以下の方法があります：

max.poll.records を使用して各バッチを小さく保つ。
CooperativeStickyAssignor または新しい Consumer Group プロトコルに切り替える。
安定した識別子を持つデプロイメント向けに Static Membership を使用する。
onPartitionsRevoked() で適切なタイミングでオフセットをコミットする。

最後に、常にメッセージ処理が冪等性を持つように設計してください。Rebalance を完全に排除することはできませんが、発生回数を減らし、害を最小限にすることはできます。

KafkaのConsumer GroupとRebalance: Consumerが停止する理由

1. 概要

2. Consumer Group のしくみ

3. Rebalance とは何か、いつ発生するのか

4. Consumer が停止する理由

4.1. Eager Rebalance：グループ全体の停止

4.2. 処理が遅すぎる：max.poll.interval.ms オーバー

5. 修正方法

5.1. バッチサイズと poll 時間の調整

5.2. Cooperative Rebalancing

5.3. Static Membership

5.4. 新しい Consumer Group プロトコル

6. パーティション取り消し時の処理

7. よくあるエラー

8. まとめ

同じトピックの記事

Kafka の Consumer group と Rebalance (Spring Boot)

Change Data CaptureとDebeziumとKafka

KafkaでのSagaパターンによる分散トランザクション：ツアー予約の例

KafkaとのOutboxパターン：データベース書き込みでイベントを失わない

Kafkaの監視: consumer lagと監視対象のメトリクス

4.2. 処理が遅すぎる：`max.poll.interval.ms` オーバー