Consumer-Gruppen und Rebalancing in Kafka: Warum Consumer stoppen

1. Überblick

Sie haben einen Service, der Nachrichten aus Kafka liest und seit vielen Tagen stabil läuft. Eines Tages steigt der Consumer-Lag plötzlich an. Die Logs sind voll von Revoking previously assigned partitions und CommitFailedException. Der Schuldige ist normalerweise das Rebalancing der Consumer-Gruppe.

Dieser Artikel erklärt, wie eine Consumer-Gruppe Partitionen aufteilt, wann Rebalancing auftritt und warum es Consumer zum Stillstand bringt. Nach dem Lesen werden Sie wissen, wie Sie Spring Kafka konfigurieren, um die Anzahl der Rebalancings zu reduzieren und die Stillstandszeit zu minimieren.

2. Wie Consumer-Gruppen funktionieren

Eine Consumer-Gruppe ist eine Sammlung von Consumern, die eine gemeinsame group.id verwenden. Kafka teilt die Partitionen eines Topics an die Mitglieder der Gruppe auf. Jede Partition wird zu jedem Zeitpunkt nur einem Consumer zugewiesen.

Angenommen, das Topic orders hat 6 Partitionen. Sie führen 3 Instanzen von order-service aus:

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

Wenn Sie 7 Instanzen ausführen, sitzt eine Instanz untätig herum, da keine Partitionen mehr zugewiesen werden können.

Zunächst fügen Sie die Abhängigkeit hinzu:

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

Ein einfacher Listener in 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("Bestellung {} von Partition {} empfangen", record.key(), record.partition());
    }
}

Die Annotation @KafkaListener erstellt einen Listener-Container. Innerhalb des Containers befindet sich ein Thread, der die poll()-Schleife ausführt. Alle Instanzen mit groupId = "order-service" nehmen an derselben Gruppe teil. Ein Broker spielt die Rolle des Group Coordinators und verwaltet die Liste der Mitglieder und die Partitionsverteilung.

3. Was ist Rebalancing und wann tritt es auf?

Rebalancing ist der Prozess, bei dem der Group Coordinator Partitionen den Mitgliedern neu zuteilt. Szenarien, die Rebalancing verursachen:

1. Neuer Consumer tritt bei: Hochfahren, Bereitstellung neuer Instanzen.
2. Consumer verlässt die Gruppe absichtlich: Herunterfahren, Rollender Neustart.
3. Consumer wird als offline angesehen, weil er keinen Heartbeat innerhalb von session.timeout.ms sendet. Standard ist 45 Sekunden.
4. Consumer ruft poll() nicht innerhalb von max.poll.interval.ms auf. Standard ist 5 Minuten.
5. Die Anzahl der Partitionen eines Topics ändert sich, oder ein musterbasiertes Abo passt zu weiteren neuen Topics.

Die beiden Timeout-Mechanismen in den Punkten 3 und 4 sind unterschiedlich und werden oft verwechselt:

Konfiguration	Standard	Was wird geprüft
heartbeat.interval.ms	3000	Rhythmus des Heartbeat-Versands (von Background-Thread)
session.timeout.ms	45000	Läuft der Prozess noch
max.poll.interval.ms	300000	Hängt der Verarbeitungs-Thread

Der Heartbeat wird von einem separaten Background-Thread gesendet. Der Prozess läuft daher noch, wenn der Verarbeitungs-Thread hängt. Umgekehrt erkennt max.poll.interval.ms korrekt den Fall, in dem Ihr Message-Verarbeitungscode zu langsam ist.

4. Warum werden Consumer gestoppt

4.1. Eager Rebalancing: Die gesamte Gruppe stoppt

Mit dem traditionellen Protokoll und Eager-Assignoren (RangeAssignor, RoundRobinAssignor, StickyAssignor) verläuft jedes Rebalancing wie folgt:

1. Alle Consumer widerrufen alle Partitionen, die sie halten.
2. Alle Consumer senden JoinGroup-Anfragen erneut und warten aufeinander.
3. Der Leader der Gruppe berechnet einen neuen Verteilungsplan, und der Coordinator sendet das Ergebnis zurück.
4. Consumer erhalten neue Partitionen und setzen die Verarbeitung fort.

Während die Schritte 1–3 ablaufen, verarbeitet niemand in der Gruppe Nachrichten. Nur eine neue Instanz, die gestartet wird, bringt die ganze Gruppe zum Stillstand. Dies ist das Phänomen „Stop-the-World”.

4.2. Zu langsame Verarbeitung: Überschreitung von max.poll.interval.ms

Die häufigste Ursache in der Praxis ist langsamer Verarbeitungscode. Das folgende Beispiel ruft einen externen Service auf, der etwa 1 Sekunde pro Nachricht dauert:

@KafkaListener(topics = "orders", groupId = "order-service")
public void onOrder(ConsumerRecord<String, String> record) throws InterruptedException {
    paymentService.charge(record.value()); // angenommen ~1 Sekunde
}

Standardmäßig ist max.poll.records = 500. Ein poll()-Aufruf kann 500 Datensätze zurückgeben, was etwa 500 Sekunden Verarbeitung bedeutet. Diese Zahl überschreitet die 300 Sekunden von max.poll.interval.ms. Der Consumer verlässt die Gruppe dann von selbst, und Sie sehen ein Protokoll wie dieses:

[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.

Danach schlägt der Offset-Commit fehl:

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.

Dies führt zu einer Kette von Problemen:

• Die Partition dieses Consumers wird jemand anderem zugewiesen.
• Da der Offset nicht committed wurde, wird der neue Besitzer der Partition diese Nachrichten erneut verarbeiten.
• Der alte Consumer tritt erneut bei und verursacht ein weiteres Rebalancing.
• Wenn der nächste Batch immer noch langsam ist, wiederholt sich die Schleife. Der Lag steigt kontinuierlich.

5. Behebung

5.1. Anpassung der Batch-Größe und Poll-Zeit

Regel zum Merken: max.poll.records × Zeit pro Datensatz muss deutlich kleiner als max.poll.interval.ms sein.

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

Mit 50 Datensätzen und 1 Sekunde pro Datensatz dauert ein Batch etwa 50 Sekunden. Diese Zahl liegt sicher im Limit von 120 Sekunden.

Hinweis: Erhöhen Sie max.poll.interval.ms nicht auf mehrere Stunden, um „das Feuer zu löschen”. Dieser Wert ist auch die maximale Zeit für Kafka, um zu erkennen, dass ein Consumer wirklich eingefroren ist. Außerdem muss Rebalancing auf den langsamsten Consumer warten, der poll() wieder aufruft. Ein großer Wert verlängert also das Rebalancing.

Wenn jede Nachricht grundsätzlich lange dauert, z.B. wegen langsamer API-Aufrufe, sollten Sie ein Timeout für die externen Aufrufe festlegen. Eine andere Möglichkeit besteht darin, Nachrichten in eine interne Warteschlange zu verschieben, anstatt den Poll-Thread zu lange zu halten.

5.2. Cooperative Rebalancing

CooperativeStickyAssignor verwendet inkrementelles Rebalancing. Consumer widerrufen nur die Partitionen, die wirklich versetzt werden müssen, während andere Partitionen normal verarbeitet werden.

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

Wenn Sie order-service-4 zur Gruppe mit 3 Instanzen aus Abschnitt 2 hinzufügen, werden nur wenige Partitionen auf die neue Instanz verschoben:

order-service-1  -> orders-0, orders-1        (unverändert, kein Stopp)
order-service-2  -> orders-2                  (verliert orders-3)
order-service-3  -> orders-4                  (verliert orders-5)
order-service-4  -> orders-3, orders-5

Hinweis: Sie können nicht direkt von Eager zu Cooperative auf einer laufenden Gruppe wechseln. Sie benötigen zwei Rolling Deployments. Beim ersten Mal konfigurieren Sie beide: CooperativeStickyAssignor,RangeAssignor. Beim zweiten Mal entfernen Sie RangeAssignor. Wenn die Mitglieder keinen gemeinsamen Assignor haben, meldet die Gruppe einen Fehler.

5.3. Static Membership

Bei jedem Rolling Restart verlässt eine Instanz die Gruppe und kehrt zurück. Das Ergebnis sind zwei Rebalancings pro Pod. Static Membership löst dies. Sie weisen jeder Instanz eine feste Kennung über group.instance.id zu:

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

Consumer mit group.instance.id senden beim Herunterfahren kein LeaveGroup. Kehren sie innerhalb von session.timeout.ms mit der gleichen ID zurück, gibt der Coordinator genau die alten Partitionen zurück, ohne Rebalancing.

Hinweis: HOSTNAME muss über Neustarts hinweg stabil sein, z.B. der Pod-Name eines StatefulSet auf Kubernetes. Bei normalen Deployments ändert sich der Pod-Name jedes Mal, daher hat Static Membership keine Wirkung. Wenn zwei Instanzen mit derselben ID gleichzeitig laufen, stoßen sie sich gegenseitig aus der Gruppe (FencedInstanceIdException). Wenn concurrency > 1 ist, fügt Spring Kafka automatisch ein Suffix -0, -1 usw. zur ID für jeden Consumer-Kind hinzu.

Der Kompromiss dabei: Wenn eine Instanz wirklich ausfällt, werden ihre Partitionen so lange von niemandem verarbeitet, bis session.timeout.ms abgelaufen ist.

5.4. Neues Consumer Group-Protokoll

Seit Kafka 4.0 ist das neue Rebalancing-Protokoll (KIP-848) produktionsreif. Sie aktivieren es mit group.protocol=consumer. Die Partitionsverteilung wird vollständig an den Broker verlagert und erfolgt immer inkrementell. Consumer müssen nicht mehr aufeinander über eine gemeinsame JoinGroup-Barriere warten.

Mit diesem Protokoll wird die partition.assignment.strategy auf der Client-Seite nicht mehr verwendet. Stattdessen wählt der Broker den Assignor (die Client-Konfiguration ist group.remote.assignor). Wenn Ihr Cluster und Client neu genug sind, ist dies eine überlegenswerte Option. Sie sollten die kafka-clients-Version überprüfen, die Spring Boot herunterlädt, bevor Sie diese aktivieren.

6. Umgang mit Partitionsentzug

Egal wie sehr Sie optimieren, Rebalancing wird immer noch stattfinden. Es ist wichtig, Offsets vor dem Partitionsentzug zu committen, um doppelte Verarbeitung zu reduzieren.

Spring Kafka committed ausstehende Offsets automatisch, bevor Partitionen entzogen werden. Wenn Sie KafkaConsumer direkt verwenden, müssen Sie dies selbst mit ConsumerRebalanceListener tun:

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) {
        // Hier können Sie den lokalen State für die neuen Partitionen neu laden
    }
}

Erklärung der einzelnen Teile:

• markProcessed() zeichnet den nächsten Offset auf, der gelesen werden soll, d.h. offset + 1, nachdem ein Datensatz verarbeitet wurde.
• onPartitionsRevoked() wird im Poll-Thread selbst ausgeführt, kurz bevor Partitionen entzogen werden. Es committed nur Offsets der entzogenen Partitionen. Dies ist entscheidend für Cooperative Rebalancing, da dort nur einige Partitionen entzogen werden.
• commitSync() blockiert, bis der Broker bestätigt, um sicherzustellen, dass der Offset gespeichert ist, bevor ein anderer Consumer die Partition übernimmt.

Sie registrieren den Listener beim Abonnieren: consumer.subscribe(List.of("orders"), listener).

Testen Sie dieses Verhalten mit dem MockConsumer aus kafka-clients:

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

Der Test simuliert, dass nur orders-0 entzogen wird. Das erwartete Ergebnis ist, dass Offset 42 für orders-0 committed wird. orders-1 wird weiterhin gehalten, daher werden keine Offsets committed.

Hinweis: Code in onPartitionsRevoked() muss schnell sein. Dieser Callback wird während des Rebalancing ausgeführt. Wenn er langsam ist, muss die ganze Gruppe warten.

7. Häufige Fehler

• Asynchrone Verarbeitung, aber weiterhin Commit pro Poll. Sie drücken Datensätze in einen ExecutorService, und der Container committed Offsets direkt nach poll(). Wenn Rebalancing auftritt, können teilweise verarbeitete Nachrichten verloren gehen. Committen Sie nur Offsets von vollständig verarbeiteten Datensätzen.
• Verwechslung von session.timeout.ms mit max.poll.interval.ms. Das Erhöhen des Session-Timeouts hilft nicht, wenn die Ursache langsame Verarbeitung ist.
• Einstellung von concurrency größer als die Anzahl der Partitionen. Überschüssige Consumer haben nichts zu tun. Sie machen nur die Gruppe größer und das Rebalancing länger.
• Lange GC-Pausen oder CPU wird bei Containern gedrosselt. Der Heartbeat-Thread kommt nicht hinterher, Consumer werden aus der Gruppe entfernt, obwohl der Code nicht langsam ist. Überprüfen Sie JVM-Metriken und CPU-Limits.
• Keine idempotente Verarbeitung. Rebalancing kann immer zu doppelter Verarbeitung einiger Nachrichten führen. Gehen Sie nicht von „exactly once” aus, wenn Sie nicht Transaktionen oder einen Deduplication-Mechanismus wie die Überprüfung, ob eine orderId bereits verarbeitet wurde, verwenden.

8. Fazit

Rebalancing ist ein notwendiger Mechanismus, damit Consumer-Gruppen elastisch skaliert werden und fehlertolerant sind. Der Preis dafür ist, dass Consumer kurzzeitig pausieren müssen. Die häufigste Ursache für kontinuierliche Consumer-Stopps ist, wenn die poll()-Schleife max.poll.interval.ms überschreitet. Der Consumer wird dann aus der Gruppe entfernt, Commits schlagen fehl und Rebalancing wiederholt sich endlos.

Um dies zu minimieren, können Sie:

• Halten Sie jeden Batch klein mit max.poll.records.
• Wechseln Sie zu CooperativeStickyAssignor oder dem neuen Consumer-Group-Protokoll.
• Verwenden Sie Static Membership für Deployments mit stabilen Identifiern.
• Committen Sie Offsets zeitgerecht in onPartitionsRevoked().

Schließlich entwerfen Sie die Message-Verarbeitung immer idempotent. Rebalancing kann nicht vollständig eliminiert werden, nur weniger häufig und weniger schädlich gemacht werden.