JA
FrançaisDeutschTiếng ViệtEspañolEnglishPortuguês日本語简体中文한국어
ホーム / アーキテクチャ・設計 / Kafka

Kafka の Consumer group と Rebalance (Spring Boot)

Cris The Dev · 2026/10/13 · 9 分で読めます · 217

1. 概要

複数の instance で同じ service を実行する場合、どの message をどの instance で処理するか把握する必要があります。Kafka はこれを consumer group と rebalance メカニズムで解決します。

このアーティクルでは、Kafka が consumer group 内で partition を配分する方法と、rebalance を引き起こす状況について説明します。その後、Spring Boot を設定して rebalance の頻度を減らし、データの喪失や重複を避ける方法を紹介します。記事の最後には、Embedded Kafka を使ったテストがあり、partition の配分方法を検証できます。

2. Consumer group はどのように機能するか

consumer group は同じ group.id を持つ複数の consumer の集合です。Kafka は以下を保証します：

  • Topic の各 partition は、特定の時刻において group 内のちょうど 1 つの consumer に割り当てられます。
  • 1 つの consumer は複数の partition を受け取ることができます。
  • 異なるグループは独立して読み取ります。各グループは独自の offset を持ちます。

例として、orders topic が 4 つの partition を持つとします：

Group "billing" with 2 consumers:
  consumer-A -> orders-0, orders-1
  consumer-B -> orders-2, orders-3

Group "billing" with 5 consumers:
  consumer-A -> orders-0
  consumer-B -> orders-1
  consumer-C -> orders-2
  consumer-D -> orders-3
  consumer-E -> (idle, no partition assigned)

重要な結果として、partition の数が group の最大並列処理度を制限します。partition 数を超える consumer を追加しても、処理速度は向上しません。

調整は broker によって行われ、group coordinator と呼ばれます。Consumer は定期的に heartbeat を coordinator に送信して、自分がまだ生存していることを報告します。

2.1. 依存関係

Spring for Apache Kafka を追加します。Spring Boot 3.x はバージョン管理を行います：

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

2.2. シンプルなリスナー

@Component
public class OrderListener {

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

    @KafkaListener(topics = "orders", groupId = "billing", concurrency = "2")
    public void onOrder(ConsumerRecord<String, String> record) {
        log.info("partition={} offset={} order={}",
                record.partition(), record.offset(), record.value());
    }
}
  • groupId = "billing" は consumer を billing group に配置します。
  • concurrency = "2" は同じ instance 内に 2 つの consumer (2 つのスレッド) を作成します。2 つの instance で service を実行する場合、group は 4 つの consumer を持ち、各 consumer は 1 つの partition を受け取ります。

起動時のログは partition 配分の結果を示します：

billing: partitions assigned: [orders-0, orders-1]
billing: partitions assigned: [orders-2, orders-3]

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

Rebalance は group が consumer 間で partition を再配分するプロセスです。以下の場合に発生します：

  1. 新しい consumer が参加する場合。例えば、pod を scale するか、新しい instance を deploy する。
  2. Consumer が通常の方法で終了する場合。例えば、close() を呼び出して shutdown する。
  3. Consumer が dead と見なされる場合：coordinator が session.timeout.ms 内に heartbeat を受け取らない。
  4. Consumer の処理が遅すぎる場合：poll() 呼び出し間の間隔が max.poll.interval.ms を超える。Consumer は自動的に group を離れます。
  5. Topic が変更される場合：partition 数を増やす、または subscribe したパターンに一致する新しい topic が追加される。

問題は、古い eager プロトコルでは、すべての consumer がすべての partition を返してから再配分されることです。この間、group は処理を停止します。これは通常、“stop-the-world” と呼ばれています。大きな group や rolling deploy では、rebalance が繰り返し発生する可能性があります。

4. 重要なタイムアウトパラメータ

spring:
  kafka:
    bootstrap-servers: localhost:9092
    consumer:
      group-id: billing
      enable-auto-commit: false
      max-poll-records: 100
      properties:
        session.timeout.ms: 45000
        heartbeat.interval.ms: 3000
        max.poll.interval.ms: 300000
パラメータ 意味
session.timeout.ms heartbeat なしで consumer が削除されるまでの最大時間
heartbeat.interval.ms heartbeat 送信のサイクル。通常 ≤ session.timeout.ms の 1/3
max.poll.interval.ms 2 つの poll() 呼び出し間の最大間隔
max-poll-records 各 poll() の最大レコード数

Heartbeat は別の background thread から送信されます。したがって、メインスレッドが忙しくても、consumer は session.timeout.ms に従って “生存” します。しかし、max.poll.interval.ms は処理スレッドのみを測定します。batch 処理が長時間かかると、consumer は group から削除されます。

典型的なエラーは次のようにログに表示されます：

Member consumer-billing-1 sending LeaveGroup request to coordinator due to
consumer poll timeout has expired. This means the time between subsequent
calls to poll() was longer than the configured max.poll.interval.ms ...
CommitFailedException: Commit cannot be completed since the group has already rebalanced

満たすべき式：

max-poll-records × 1 record の処理時間  <  max.poll.interval.ms

注意： max.poll.interval.ms を数時間に設定することで「修正」しないでください。実際に hung した consumer は、partition を長時間保持したままになり、誰も処理しません。代わりに max-poll-records を減らすか、重い処理を consume スレッドの外に移してください。

5. Rebalance の影響を減らす

5.1. Cooperative rebalancing

CooperativeStickyAssignor を使用すると、実際に転送が必要な partition のみが回収されます。その他の consumer は rebalance 中も自分の partition の処理を続けます。

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

例：group は A (orders-0..3) を持ち、その後 B が参加します。

Eager:       A が 4 つの partition を返す -> 再配分：A {0,1}, B {2,3}
Cooperative: A は {2,3} だけを返す、{0,1} の処理を続ける -> B が {2,3} を受け取る

注意： eager から cooperative に running group を 1 度の deploy で変更しないでください。2 回の rolling deploy が必要です。最初のデプロイでは両方の assignor を宣言します（例：CooperativeStickyAssignor,RangeAssignor）。次のデプロイで RangeAssignor を削除します。

Kafka 4.0 以降、新しい rebalance プロトコル (KIP-848) が GA になりました。group.protocol=consumer で有効にできます。この場合、partition の配分は broker が実行し、rebalance はデフォルトでインクリメンタルになり、クライアント側の partition.assignment.strategy と session.timeout.ms パラメータは使用されなくなります。cluster がこれをサポートしている場合、これが最初に検討するオプションです。

5.2. Static membership

pod を再起動するたびに、consumer は新しい member.id を取得し、これが 2 回の rebalance をトリガーします（離脱してから再参加）。Static membership は consumer に固定の識別子を割り当てます：

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

Consumer が session.timeout.ms 内に戻ってきた場合、coordinator は rebalance なしで以前の partition を返します。これは Kubernetes StatefulSet に非常に適しており、HOSTNAME が安定しています（billing-0, billing-1…）。

注意： group.instance.id は group 内でユニークである必要があります。concurrency > 1 の場合、Spring Kafka は各 consumer に自動的にサフィックス -n を追加します。2 つの instance が同じ ID を使用する場合、後から入ってくる instance は前の instance を「キック」します（FencedInstanceIdException）。ランダムな pod 名を持つ Deployment では static membership を使用しないでください。利点がないからです。

6. Rebalance 時の offset 処理

rebalance の最大のリスクは重複処理です。Consumer A がレコードの処理を終了しましたが、commit していません。Partition が B に転送され、B は最後に commit された offset から読み取ります。

Spring Kafka はデフォルトで batch ごとに commit します（AckMode.BATCH）。partition が回収される前に、保留中の offset を自動的に commit します。介入が必要な場合、例えば buffer をフラッシュするか log に書き込む場合は、ConsumerAwareRebalanceListener を使用します：

@Component
public class OrderRebalanceListener implements ConsumerAwareRebalanceListener {

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

    @Override
    public void onPartitionsRevokedBeforeCommit(Consumer<?, ?> consumer,
                                                Collection<TopicPartition> partitions) {
        log.info("partition {} を失おうとしています。メモリに保持されているデータをフラッシュしています", partitions);
    }

    @Override
    public void onPartitionsRevokedAfterCommit(Consumer<?, ?> consumer,
                                               Collection<TopicPartition> partitions) {
        log.info("commit を完了し、partition {} を返しました", partitions);
    }

    @Override
    public void onPartitionsAssigned(Consumer<?, ?> consumer,
                                     Collection<TopicPartition> partitions) {
        log.info("partition {} を受け取りました", partitions);
    }

    @Override
    public void onPartitionsLost(Consumer<?, ?> consumer,
                                 Collection<TopicPartition> partitions) {
        log.warn("commit する前に partition {} を失いました", partitions);
    }
}
  • onPartitionsRevokedBeforeCommit は container が offset を commit する前に実行されます。これは未完了の作業を完了する場所です。
  • onPartitionsRevokedAfterCommit は offset が commit された後に実行されます。
  • onPartitionsAssigned は新しい partition が割り当てられたときに実行されます。
  • onPartitionsLost は consumer が group から削除されたときに呼ばれます。例えば、max.poll.interval.ms を超えた場合。この時点では partition は別の consumer に属している可能性があるため、commit しないでください。

リスナーを container factory に登録します：

@Configuration
public class KafkaConfig {

    @Bean
    ConcurrentKafkaListenerContainerFactory<Object, Object> kafkaListenerContainerFactory(
            ConcurrentKafkaListenerContainerFactoryConfigurer configurer,
            ConsumerFactory<Object, Object> consumerFactory,
            OrderRebalanceListener rebalanceListener) {

        var factory = new ConcurrentKafkaListenerContainerFactory<Object, Object>();
        configurer.configure(factory, consumerFactory);
        factory.getContainerProperties().setConsumerRebalanceListener(rebalanceListener);
        return factory;
    }
}

configurer.configure(...) は application.yml から全体の設定を保持します。次の行は rebalance listener を追加するだけです。

注意： どんなに設定が良くても、Kafka は通常の consumer に対してat-least-once のみを保証します。処理を idempotent にしてください。例えば、PaymentService を呼び出す前に orderId が既に支払済みかチェックしてください。rebalance が重複を引き起こさないと仮定しないでください。

7. Partition の配分をテストする

次のテストは @EmbeddedKafka を使用して、2 つの consumer が同じ group で 4 つの partition を均等に共有することを確認します：

@EmbeddedKafka(partitions = 4, topics = "orders")
class ConsumerGroupAssignmentTest {

    @Test
    void twoConsumersInSameGroupShareAllPartitions(EmbeddedKafkaBroker broker) {
        Map<String, Object> props = KafkaTestUtils.consumerProps("billing", "false", broker);
        props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);
        props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);

        try (var consumerA = new KafkaConsumer<String, String>(props);
             var consumerB = new KafkaConsumer<String, String>(props)) {

            consumerA.subscribe(List.of("orders"));
            consumerB.subscribe(List.of("orders"));

            long deadline = System.currentTimeMillis() + 30_000;
            while (System.currentTimeMillis() < deadline
                    && !(consumerA.assignment().size() == 2 && consumerB.assignment().size() == 2)) {
                consumerA.poll(Duration.ofMillis(200));
                consumerB.poll(Duration.ofMillis(200));
            }

            Set<TopicPartition> all = new HashSet<>(consumerA.assignment());
            all.addAll(consumerB.assignment());

            assertThat(consumerA.assignment()).hasSize(2);
            assertThat(consumerB.assignment()).hasSize(2)
                    .doesNotContainAnyElementsOf(consumerA.assignment());
            assertThat(all).hasSize(4);
        }
    }
}

説明：

  • @EmbeddedKafka は memory 内に broker を起動し、4 つの partition を持つ orders topic を作成します。Broker はテスト method に直接 inject されます。
  • 2 つの KafkaConsumer は同じ group.id = "billing" を使用します。
  • ループは両方の consumer で poll() を交互に呼び出します。consumer は poll() 内でのみ rebalance に参加するためです。1 つの consumer だけが poll する場合、rebalance は完了しません。両方の consumer は同じスレッド上で実行されますが、別々のオブジェクトであるため、「1 つの consumer は 1 つのスレッドだけで使用される」ルールに違反しません。
  • Assertion は各 consumer が 2 つの partition を持つことをチェックし、重複がなく、合計で 4 つの partition をカバーしていることを確認します。

これは異なる partition.assignment.strategy を試すための良い方法でもあります。3 番目の consumer を追加し、1 つの consumer を閉じてから、assignment() の変化を観察してください。

8. よくあるエラー

  • Consumer 数が partition 数より多い場合：余った consumer はアイドル状態になります。concurrency を増やしても、システムは高速化しません。
  • Listener での同期処理が長すぎる場合：max.poll.interval.ms を超え、rebalance ループが発生します。Consumer は削除され、戻ってきて、同じ batch を受け取り、再び削除されます。
  • onPartitionsLost で手動 commit する場合：CommitFailedException が発生するか、新しい consumer の offset が上書きされます。
  • 異なる 2 つの service で同じ group.id を使用する場合：2 つの service は message を共有する代わりに、それぞれが十分に受け取ります。各 service は独立しており、独自の group を持つべきです。
  • Large group で eager protocol での rolling deploy：各 pod 再起動は 2 つの stop-the-world を引き起こします。Cooperative を使用するか、static membership を使用するか、Kafka 4.x の新しいプロトコルを使用してください。

9. 結論

Consumer group を使用すると、Kafka の読み取りを水平に scale できます。上限は partition の数です。Rebalance はその柔軟性の代償です。member が参加、離脱、死亡、または処理が遅すぎるときに発生します。

Rebalance の害を最小限にするために、以下のことを行うべきです：

  • max.poll.interval.ms と max-poll-records を適切に設定する。
  • CooperativeStickyAssignor を使用するか、Kafka 4.x で group.protocol=consumer を使用する。
  • インフラストラクチャが許可する場合、static membership を有効にする。
  • ConsumerAwareRebalanceListener 経由で partition のライフサイクルを処理する。

最後に、常に処理を idempotent に設計してください。at-least-once が既定の保証であり、これを受け入れる必要があるためです。

kafkaspring-kafkaconsumer-grouprebalancespring-boot
著者のアプリReflex Arena11 brain-speed games, live 1v1 duels, monthly seasonsiOS · Android

同じトピックの記事

Kafka · 2026/10/12 · 8 分で読めます · 312

Change Data CaptureとDebeziumとKafka

Debeziumを使ってPostgreSQLのWALを読み込み、すべてのデータ変更をKafkaにプッシュし、Spring Bootで消費。データ書き込みコードの変更は不要です。