Consumer Group und Rebalance in Kafka mit Spring Boot

1. Übersicht

Wenn Sie mehrere Instanzen desselben Service ausführen, die aus Kafka lesen, müssen Sie wissen, welche Nachricht von welcher Instanz verarbeitet wird. Kafka löst dies durch Consumer Groups und den Rebalance-Mechanismus.

Dieser Artikel erklärt, wie Kafka Partitionen in einer Gruppe verteilt und in welchen Situationen ein Rebalance auftritt. Anschließend konfigurieren Sie Spring Boot so, dass Rebalances weniger häufig auftreten und keine Daten verloren gehen oder doppelt verarbeitet werden. Am Ende dieses Artikels gibt es einen Test mit Embedded Kafka, um die Partitionsaufteilung zu überprüfen.

2. Wie Consumer Groups funktionieren

Eine Consumer Group ist eine Menge von Consumern mit derselben group.id. Kafka garantiert:

Jede Partition eines Topics wird genau einem Consumer in der Gruppe zu jedem Zeitpunkt zugewiesen.
Ein Consumer kann mehrere Partitionen erhalten.
Verschiedene Gruppen lesen unabhängig voneinander. Jede Gruppe hat ihren eigenen Offset.

Angenommen, das Topic orders hat 4 Partitionen:

Gruppe "billing" mit 2 Consumern:
  consumer-A -> orders-0, orders-1
  consumer-B -> orders-2, orders-3

Gruppe "billing" mit 5 Consumern:
  consumer-A -> orders-0
  consumer-B -> orders-1
  consumer-C -> orders-2
  consumer-D -> orders-3
  consumer-E -> (inaktiv, keine Partition)

Eine wichtige Konsequenz ist, dass die Anzahl der Partitionen die maximale Parallelität einer Gruppe begrenzt. Das Hinzufügen von Consumern über die Anzahl der Partitionen hinaus hilft nicht, schneller zu verarbeiten.

Die Koordination wird von einem Broker übernommen, der als Group Coordinator bezeichnet wird. Consumer senden in regelmäßigen Abständen Heartbeats an den Coordinator, um zu zeigen, dass sie noch aktiv sind.

2.1. Abhängigkeit

Fügen Sie Spring for Apache Kafka hinzu. Spring Boot 3.x verwaltet die Version bereits:

<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. Ein einfacher Listener

@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" platziert den Consumer in der Gruppe billing.
concurrency = "2" erstellt 2 Consumer (2 Threads) in der gleichen Instanz. Wenn Sie 2 Instanzen des Service ausführen, hat die Gruppe 4 Consumer, jeder Consumer erhält 1 Partition.

Das Log beim Start zeigt die Ergebnis der Aufteilung:

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

3. Was ist Rebalance und wann tritt es auf?

Rebalance ist der Prozess, in dem eine Gruppe Partitionen unter ihren Mitgliedern neu verteilt. Es tritt auf, wenn:

Ein neuer Consumer tritt bei, z.B. Sie skalieren einen weiteren Pod oder deployen eine neue Instanz.
Ein Consumer tritt ordnungsgemäß aus, z.B. bei einem sauberen Shutdown mit close().
Ein Consumer wird als tot betrachtet: der Coordinator empfängt für session.timeout.ms keine Heartbeats.
Consumer verarbeitet zu langsam: das Intervall zwischen zwei poll()-Aufrufen überschreitet max.poll.interval.ms. Der Consumer verlässt die Gruppe von selbst.
Topic ändert sich: Anzahl der Partitionen wird erhöht oder ein neues Topic passt zum abonnierten Pattern.

Das Problem bei dem alten Eager-Protokoll ist, dass alle Consumer alle ihre Partitionen zurückgeben müssen, bevor sie neu zugewiesen werden. Die gesamte Gruppe stoppt die Verarbeitung während dieser Zeit, dieses Phänomen wird oft als “Stop-the-World” bezeichnet. Bei großen Gruppen oder Rolling Deployments kann ein Rebalance sich wiederholt auslösen.

4. Wichtige Timeout-Parameter

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

Parameter	Bedeutung
`session.timeout.ms`	Maximale Zeit ohne Heartbeat, bevor der Consumer entfernt wird
`heartbeat.interval.ms`	Intervall für das Senden von Heartbeats, normalerweise ≤ 1/3 von `session.timeout.ms`
`max.poll.interval.ms`	Maximales Intervall zwischen zwei `poll()`-Aufrufen
`max-poll-records`	Maximale Anzahl von Records pro `poll()`-Aufruf

Heartbeats werden von einem separaten Background-Thread gesendet. Daher bleibt der Consumer “am Leben” gemäß session.timeout.ms, selbst wenn der Haupt-Thread beschäftigt ist. Jedoch misst max.poll.interval.ms nur den Verarbeitungs-Thread. Wenn ein Batch zu lange dauert, wird der Consumer trotzdem aus der Gruppe entfernt.

Ein typischer Fehler im Log sieht so aus:

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

Die Formel, die erfüllt sein muss:

max-poll-records × Zeit für 1 Record  <  max.poll.interval.ms

Hinweis: Reparieren Sie das Problem nicht, indem Sie max.poll.interval.ms auf mehrere Stunden erhöhen. Ein wirklich eingefrorener Consumer wird eine Partition lange Zeit halten, ohne dass jemand sie verarbeitet. Reduzieren Sie stattdessen max-poll-records oder verschieben Sie schwere Aufgaben aus dem Consumer-Thread.

5. Auswirkungen von Rebalance reduzieren

5.1. Cooperative Rebalancing

Mit CooperativeStickyAssignor werden nur Partitionen zurückgefordert, die tatsächlich verschoben werden müssen. Die verbleibenden Consumer verarbeiten weiterhin ihre Partitionen, während das Rebalance stattfindet.

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

Beispiel: eine Gruppe mit A (orders-0..3), dann tritt B bei.

Eager:       A gibt alle 4 Partitionen ab -> neu verteilt: A {0,1}, B {2,3}
Cooperative: A gibt nur {2,3} ab, verarbeitet {0,1} weiterhin -> B erhält {2,3}

Hinweis: Nicht direkt von Eager zu Cooperative für eine laufende Gruppe in einem Deploy wechseln. Sie benötigen zwei Rolling Deployments. Deklarieren Sie beim ersten Mal beide Assignors, z.B. CooperativeStickyAssignor,RangeAssignor. Beim zweiten Mal entfernen Sie dann RangeAssignor.

Ab Kafka 4.0 ist das neue Rebalance-Protokoll (KIP-848) GA. Sie aktivieren es mit group.protocol=consumer. Dabei führt der Broker die Aufteilung durch, Rebalances sind standardmäßig inkrementell, und die Client-seitigen Parameter partition.assignment.strategy und session.timeout.ms werden nicht mehr verwendet. Falls Ihr Cluster dies unterstützt, ist dies die erste Option, die Sie in Betracht ziehen sollten.

5.2. Static Membership

Jedes Mal, wenn ein Pod neu startet, erhält der Consumer eine neue member.id, was zwei Rebalances auslöst (Austritt dann Eintritt). Static Membership weist dem Consumer eine feste Identität zu:

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

Wenn der Consumer innerhalb von session.timeout.ms zurückkommt, weist der Coordinator ihm die gleichen Partitionen zu, ohne dass ein Rebalance nötig ist. Dies funktioniert gut mit Kubernetes StatefulSets, wo HOSTNAME stabil ist (billing-0, billing-1…).

Hinweis: group.instance.id muss eindeutig in der Gruppe sein. Mit concurrency > 1 fügt Spring Kafka automatisch das Suffix -n für jeden Sub-Consumer hinzu. Wenn zwei Instanzen die gleiche ID verwenden, wird die spätere Instanz die frühere ‘rauswerfen’ (FencedInstanceIdException). Verwenden Sie Static Membership nicht mit Deployments mit zufälligen Pod-Namen, da Sie keinen Vorteil erhalten würden.

6. Offset-Handling während Rebalance

Das größte Risiko beim Rebalance ist die doppelte Verarbeitung. Consumer A hat einen Record verarbeitet, aber noch nicht committed; die Partition wird zu B verschoben, und B liest ab dem letzten committierten Offset.

Spring Kafka committed standardmäßig nach jedem Batch (AckMode.BATCH) und committed automatisch ausstehende Offsets, bevor Partitionen zurückgefordert werden. Wenn Sie eingreifen müssen, z.B. um Buffer zu leeren oder Protokolle zu schreiben, verwenden Sie 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("Werde Partition {} verlieren, flush ausstehende Daten", partitions);
    }

    @Override
    public void onPartitionsRevokedAfterCommit(Consumer<?, ?> consumer,
                                               Collection<TopicPartition> partitions) {
        log.info("Offset committed, Partition {} abgegeben", partitions);
    }

    @Override
    public void onPartitionsAssigned(Consumer<?, ?> consumer,
                                     Collection<TopicPartition> partitions) {
        log.info("Partition {} erhalten", partitions);
    }

    @Override
    public void onPartitionsLost(Consumer<?, ?> consumer,
                                 Collection<TopicPartition> partitions) {
        log.warn("Partition {} verloren ohne Commit", partitions);
    }
}

onPartitionsRevokedBeforeCommit läuft, bevor der Container den Offset committed. Hier können Sie unvollendete Arbeiten abschließen.
onPartitionsRevokedAfterCommit läuft, nachdem der Offset committed wurde.
onPartitionsLost wird aufgerufen, wenn ein Consumer aus der Gruppe entfernt wird, z.B. wegen Überschreitung von max.poll.interval.ms. Die Partition kann dann bereits einem anderen Consumer gehören, also committen Sie nicht.

Registrieren Sie den Listener bei der 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(...) behält alle Konfigurationen aus application.yml bei. Die nächste Zeile fügt nur den Rebalance-Listener hinzu.

Hinweis: Unabhängig davon, wie gut Sie es konfigurieren, Kafka garantiert nur at-least-once für normale Consumer. Machen Sie Ihre Verarbeitung idempotent, z.B. prüfen Sie, ob die orderId bereits bezahlt wurde, bevor Sie PaymentService aufrufen. Setzen Sie nicht darauf, dass ein Rebalance niemals zu Duplikaten führt.

7. Testen der Partitionsaufteilung

Der folgende Test verwendet @EmbeddedKafka, um zu überprüfen, dass zwei Consumer in der gleichen Gruppe 4 Partitionen gleichmäßig teilen:

@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);
        }
    }
}

Erklärung:

@EmbeddedKafka startet einen In-Memory-Broker und erstellt das Topic orders mit 4 Partitionen. Der Broker wird direkt als Parameter der Test-Methode injiziert.
Zwei KafkaConsumer verwenden die gleiche group.id = "billing".
Die Schleife ruft poll() wechselweise auf beiden Consumern auf, da Consumer nur während poll() an Rebalances teilnehmen. Wenn nur Consumer A pollst, ist das Rebalance nicht vollständig. Die zwei Consumer laufen auf dem gleichen Thread, sind aber separate Objekte, daher wird gegen die Regel “ein Consumer darf nur von einem Thread verwendet werden” nicht verstoßen.
Die Assertions prüfen, dass jeder Consumer 2 Partitionen hat, diese nicht überlappen und insgesamt alle 4 Partitionen abdecken.

Dies ist auch eine gute Möglichkeit, verschiedene partition.assignment.strategy zu testen. Fügen Sie einen dritten Consumer hinzu, schließen Sie einen Consumer und beobachten Sie, wie sich assignment() ändert.

8. Häufige Fehler

Mehr Consumer als Partitionen: die überschüssigen Consumer bleiben untätig. Das Erhöhen der concurrency macht das System nicht schneller.
Synchrone Verarbeitung zu lange im Listener: Überschreitung von max.poll.interval.ms verursacht eine Rebalance-Schleife. Der Consumer wird entfernt, kommt zurück, erhält den gleichen Batch, und wird wieder entfernt.
Manuelles Commit in onPartitionsLost: verursacht CommitFailedException oder schlimmer, überschreibt den Offset des neuen Consumers.
Gleiche group.id für zwei verschiedene Services: die beiden Services werden sich Nachrichten “teilen”, statt jeder erhält das Komplette. Jeder Service sollte unabhängig seine eigene Gruppe haben.
Rolling Deploy mit Eager Protocol bei großen Gruppen: Jeder Pod-Neustart verursacht zwei Stop-the-World-Ereignisse. Verwenden Sie stattdessen Cooperative, Static Membership oder das neue Protokoll von Kafka 4.x.

9. Fazit

Consumer Groups ermöglichen es Ihnen, Kafka-Lesevorgänge horizontal zu skalieren, wobei die Obergrenze die Anzahl der Partitionen ist. Rebalance ist der Preis für diese Flexibilität. Es tritt auf, wenn Mitglieder beitreten, austreten, sterben oder zu langsam verarbeiten.

Um die Auswirkungen von Rebalances zu minimieren, sollten Sie Folgendes tun:

Konfigurieren Sie max.poll.interval.ms und max-poll-records angemessen.
Verwenden Sie CooperativeStickyAssignor oder group.protocol=consumer auf Kafka 4.x.
Aktivieren Sie Static Membership, wenn die Infrastruktur es erlaubt.
Handhaben Sie den Partitions-Lebenszyklus über ConsumerAwareRebalanceListener.

Schließlich sollten Sie Ihre Verarbeitung immer idempotent gestalten, denn at-least-once ist die Standardgarantie, die Sie akzeptieren müssen.