Consumer group et rebalance dans Kafka : pourquoi le consumer s'arrête

1. Vue d’ensemble

Vous avez un service qui lit des messages de Kafka et fonctionne correctement depuis plusieurs jours. Un jour, le consumer lag augmente soudainement. Les logs sont remplies de Revoking previously assigned partitions et CommitFailedException. Le coupable est généralement le rebalance du consumer group.

Cet article explique comment le consumer group divise les partitions, quand un rebalance se produit et pourquoi il arrête le consumer. Une fois lu, vous saurez comment configurer Spring Kafka pour réduire le nombre de rebalances et minimiser le temps d’arrêt à chaque fois.

2. Comment fonctionne le consumer group

Un consumer group est un ensemble de consumers partageant un group.id. Kafka divise les partitions du topic entre les membres du group. Chaque partition est assignée à un seul consumer à la fois.

Supposons que le topic orders a 6 partitions. Vous exécutez 3 instances de order-service :

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

Si vous exécutez 7 instances, une instance restera inactive car il n’y a pas de partition à assigner.

D’abord, ajoutez la dépendance :

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

Un listener simple dans 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("Commande reçue {} depuis la partition {}", record.key(), record.partition());
    }
}

L’annotation @KafkaListener crée un listener container. À l’intérieur du container se trouve un thread exécutant la boucle poll(). Toutes les instances avec groupId = "order-service" participent au même group. Un broker joue le rôle de group coordinator gérant la liste des membres et la répartition des partitions.

3. Qu’est-ce que le rebalance et quand se produit-il

Le rebalance est le processus par lequel le group coordinator réassigne les partitions aux membres. Situations causant un rebalance :

1. Un nouveau consumer se joint : scale up, déployer une nouvelle instance.
2. Un consumer quitte le group volontairement : shutdown, rolling restart.
3. Un consumer est considéré comme mort car il n’envoie pas de heartbeat dans session.timeout.ms. La valeur par défaut est 45 secondes.
4. Un consumer n’appelle pas poll() dans max.poll.interval.ms. La valeur par défaut est 5 minutes.
5. Le nombre de partitions du topic change, ou la souscription selon pattern correspond à de nouveaux topics.

Les deux mécanismes de timeout des points 3 et 4 sont différents et souvent confondus :

Configuration	Par défaut	Vérifie quoi
heartbeat.interval.ms	3000	Fréquence d’envoi du heartbeat (depuis un thread d’arrière-plan)
session.timeout.ms	45000	Le processus est-il encore vivant
max.poll.interval.ms	300000	Le thread de traitement est-il bloqué

Le heartbeat est envoyé par un thread d’arrière-plan séparé. Le processus reste donc « vivant » alors que le thread de traitement est bloqué. En revanche, max.poll.interval.ms détecte correctement les cas où votre code de traitement des messages est trop lent.

4. Pourquoi le consumer s’arrête

4.1. Eager rebalance : arrêter tout le group

Avec le protocole traditionnel et les assignors de type eager (RangeAssignor, RoundRobinAssignor, StickyAssignor), chaque rebalance se déroule comme suit :

1. Tous les consumers révoient toutes les partitions qu’ils détenaient.
2. Tous les consumers renvoient une demande JoinGroup et s’attendent mutuellement.
3. Le leader du group calcule le nouveau plan de partage, puis le coordinator envoie le résultat.
4. Les consumers reçoivent les nouvelles partitions et continuent le traitement.

Pendant que les étapes 1-3 se déroulent, personne dans le group ne traite les messages. Même une seule nouvelle instance qui démarre arrête tout le group. C’est le phénomène « stop-the-world ».

4.2. Traitement trop lent : dépasser max.poll.interval.ms

La cause la plus courante en pratique est le code de traitement lent. L’exemple suivant appelle un service externe qui prend 1 seconde par message :

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

Par défaut max.poll.records = 500. Un appel à poll() peut retourner 500 enregistrements, soit environ 500 secondes de traitement. Ce nombre dépasse 300 secondes de max.poll.interval.ms. Le consumer quitte alors le group de lui-même, et vous verrez un log comme suit :

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

Ensuite, la commande de commit d’offset échoue :

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.

Les conséquences entraînent une série de problèmes :

• La partition de ce consumer est assignée à quelqu’un d’autre.
• L’offset n’a pas pu être commis, donc celui qui reçoit la partition traite à nouveau ces messages.
• L’ancien consumer rejoint, causant un autre rebalance.
• Si le batch suivant est toujours lent, la boucle se répète indéfiniment. Le lag augmente sans cesse.

5. Comment remédier

5.1. Ajuster la taille du batch et le timing du poll

Règle à retenir : max.poll.records × temps de traitement par record doit être bien inférieur à max.poll.interval.ms.

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

Avec 50 enregistrements et 1 seconde par enregistrement, un batch prend environ 50 secondes. Ce chiffre est bien en dessous de la limite de 120 secondes.

Remarque : N’augmentez pas max.poll.interval.ms à plusieurs heures pour « combattre l’incendie ». Cette valeur est aussi le temps maximum pour que Kafka détecte un consumer vraiment gelé. De plus, un rebalance doit attendre que le consumer le plus lent appelle poll() à nouveau, donc une valeur grande prolonge le rebalance.

Si chaque message prend naturellement du temps à traiter, par exemple un appel API lent, vous devez définir un timeout pour l’appel externe. Une autre façon est de pousser le message vers une file d’attente interne, au lieu de garder le thread poll trop longtemps.

5.2. Rebalance coopératif

CooperativeStickyAssignor utilise le rebalance incrémental. Les consumers ne révoient que les partitions qui doivent vraiment être transférées, tandis que les autres partitions continuent à être traitées normalement.

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

Lors de l’ajout de order-service-4 au group de 3 instances de la section 2, seules quelques partitions sont transférées à la nouvelle instance :

order-service-1  -> orders-0, orders-1        (conservées, pas d'arrêt)
order-service-2  -> orders-2                  (perd orders-3)
order-service-3  -> orders-4                  (perd orders-5)
order-service-4  -> orders-3, orders-5

Remarque : Vous ne pouvez pas basculer directement de eager à cooperative sur un group en cours d’exécution. Vous devez faire un rolling deploy deux fois. La première fois, configurez les deux : CooperativeStickyAssignor,RangeAssignor. La seconde fois, supprimez RangeAssignor. Si les membres n’ont pas d’assignor commun, le group signalera une erreur.

5.3. Adhésion statique

À chaque rolling restart, une instance quitte le group puis revient. Le résultat est deux rebalances pour chaque pod. L’adhésion statique résout cela. Vous attribuez à chaque instance un identifiant fixe via group.instance.id :

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

Les consumers avec group.instance.id n’enverront pas LeaveGroup lors de l’arrêt. S’il revient dans session.timeout.ms avec le même id, le coordinator rend les bonnes vieilles partitions sans rebalance.

Remarque : HOSTNAME doit être stable à travers les redémarrages, par exemple le nom du pod d’un StatefulSet sur Kubernetes. Avec un Deployment classique, le nom du pod change à chaque fois, donc l’adhésion statique n’a aucun effet. Deux instances exécutées simultanément avec le même id s’évinceront du group (FencedInstanceIdException). Quand concurrency > 1, Spring Kafka ajoute automatiquement des suffixes -0, -1… à l’id pour chaque sous-consumer.

L’inconvénient de cette approche : si une instance meurt vraiment, ses partitions ne seront pas traitées jusqu’à l’expiration de session.timeout.ms.

5.4. Nouveau protocole de consumer group

À partir de Kafka 4.0, le nouveau protocole de rebalance (KIP-848) est prêt pour la production. Vous l’activez avec group.protocol=consumer. La répartition des partitions est entièrement transférée au broker et se fait toujours de manière incrémentale. Les consumers n’ont plus à s’attendre mutuellement via un « barrage » JoinGroup commun.

Avec ce protocole, partition.assignment.strategy côté client n’est plus utilisé. À la place, le broker choisit l’assignor (la configuration côté client est group.remote.assignor). Si votre cluster et votre client sont suffisamment récents, c’est une option à considérer. Vous devez vérifier la version de kafka-clients que Spring Boot utilise avant d’activer.

6. Gestion lors de la révocation des partitions

Peu importe comment vous l’optimisez, le rebalance se produira toujours. L’important est de commiter les offsets traités avant de perdre les partitions, pour réduire le traitement dupliqué.

Spring Kafka valide automatiquement les offsets en attente avant la révocation des partitions. Si vous utilisez KafkaConsumer directement, vous devez le faire vous-même avec 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) {
        // Vous pouvez recharger l'état local pour les nouvelles partitions ici
    }
}

Explication de chaque partie :

• markProcessed() enregistre le prochain offset à lire, à savoir offset + 1, après qu’un enregistrement ait été traité.
• onPartitionsRevoked() s’exécute sur le thread poll lui-même, juste avant que les partitions ne soient retirées. Il valide uniquement les offsets des partitions qui sont révoquées. C’est important avec le rebalance coopératif, car alors seule une partie des partitions est révoquée.
• commitSync() bloque jusqu’à ce que le broker confirme, s’assurant que l’offset est enregistré avant qu’un autre consumer ne reçoive la partition.

Vous enregistrez le listener lors de l’abonnement : consumer.subscribe(List.of("orders"), listener).

Testez ce comportement avec MockConsumer fourni dans 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();
    }
}

Le test simule la révocation de seulement orders-0. Le résultat attendu est que l’offset 42 soit commis pour orders-0. Tandis que orders-1 est toujours détenu, donc aucun offset n’a été commis.

Remarque : Le code dans onPartitionsRevoked() doit être rapide. Ce callback s’exécute pendant le rebalance. S’il est lent, tout le group doit attendre.

7. Erreurs courantes

• Traitement asynchrone mais commit selon poll. Vous poussez l’enregistrement vers un ExecutorService, puis le container valide l’offset immédiatement après poll(). Si un rebalance se produit, le message en cours de traitement peut être perdu. Ne validez l’offset que des enregistrements complètement traités.
• Confondre session.timeout.ms avec max.poll.interval.ms. Augmenter le session timeout n’aide pas si la cause est le traitement lent.
• Définir concurrency supérieur au nombre de partitions. Les consumers supplémentaires n’ont rien à faire. Ils rendent simplement le group plus grand et le rebalance plus long.
• Pause GC longue ou CPU limité sur le container. Le thread heartbeat ne peut pas suivre, le consumer est expulsé du group même si le code n’est pas lent. Vérifiez les métriques JVM et les limites CPU.
• Ne pas concevoir le traitement comme idempotent. Le rebalance peut toujours causer un traitement dupliqué de certains messages. Ne supposez pas « exactly once » si vous n’utilisez pas les transactions ou si vous n’avez pas de mécanisme de déduplication, par exemple vérifier si orderId a déjà été traité.

8. Conclusion

Le rebalance est un mécanisme nécessaire pour que le consumer group se redimensionne et tolère les pannes. Le prix à payer est que les consumers doivent s’arrêter temporairement. La cause la plus courante d’arrêt répété du consumer est que la boucle poll() dépasse max.poll.interval.ms. Le consumer est alors expulsé du group, la validation échoue et le rebalance se répète indéfiniment.

Pour minimiser cela, vous pouvez :

• Garder chaque batch petit avec max.poll.records.
• Passer à CooperativeStickyAssignor ou au nouveau protocole de consumer group.
• Utiliser l’adhésion statique pour les déploiements avec un identifiant stable.
• Valider l’offset au bon moment dans onPartitionsRevoked().

Enfin, concevez toujours le traitement des messages comme idempotent. Le rebalance ne peut pas être complètement éliminé, on peut seulement le rendre moins fréquent et moins nuisible.