Consumer group et rebalance dans Kafka avec Spring Boot

1. Vue d’ensemble

Lorsque vous exécutez plusieurs instances du même service lisant depuis Kafka, vous devez savoir quel message est traité par quelle instance. Kafka résout ce problème grâce aux consumer groups et au mécanisme de rebalance.

Cet article explique comment Kafka divise les partitions au sein d’un groupe et les situations qui causent un rebalance. Ensuite, vous configurerez Spring Boot pour que les rebalances se produisent moins souvent et ne causent pas de perte ou de duplication de données. À la fin, il y a un test avec Embedded Kafka pour vérifier comment les partitions sont distribuées.

2. Comment fonctionne un consumer group

Un consumer group est un ensemble de consumers ayant le même group.id. Kafka garantit :

• Chaque partition du topic est assignée à exactement un consumer du groupe à tout moment.
• Un consumer peut recevoir plusieurs partitions.
• Les différents groupes lisent indépendamment. Chaque groupe a ses propres offsets.

Supposons que le topic orders a 4 partitions :

Groupe "billing" avec 2 consumers :
  consumer-A -> orders-0, orders-1
  consumer-B -> orders-2, orders-3

Groupe "billing" avec 5 consumers :
  consumer-A -> orders-0
  consumer-B -> orders-1
  consumer-C -> orders-2
  consumer-D -> orders-3
  consumer-E -> (inactif, sans partition)

Une conséquence importante est que le nombre de partitions limite le parallélisme maximal d’un groupe. Ajouter des consumers au-delà du nombre de partitions n’aide pas à accélérer le traitement.

La coordination est effectuée par un broker, appelé group coordinator. Le consumer envoie régulièrement un heartbeat au coordinator pour signaler qu’il est actif.

2.1. Dépendance

Ajoutez Spring for Apache Kafka. Spring Boot 3.x gère déjà la version :

<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. Un listener simple

@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" place le consumer dans le groupe billing.
• concurrency = "2" crée 2 consumers (2 threads) dans la même instance. Si vous exécutez 2 instances du service, le groupe aura 4 consumers, chaque consumer recevant 1 partition.

Les logs au démarrage montrent le résultat de la distribution :

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

3. Qu’est-ce qu’un rebalance et quand se produit-il

Rebalance est le processus par lequel le groupe redistribue les partitions parmi ses membres. Cela se produit lorsque :

1. Un nouveau consumer rejoint, par exemple lorsque vous augmentez le nombre de pods ou déployez une nouvelle instance.
2. Un consumer se retire normalement, par exemple lors d’un arrêt appelant close().
3. Un consumer est considéré comme mort : le coordinator ne reçoit pas de heartbeat pendant session.timeout.ms.
4. Un consumer traite trop lentement : l’intervalle entre deux appels poll() dépasse max.poll.interval.ms. Le consumer se retire du groupe de lui-même.
5. Le topic change : augmentation du nombre de partitions, ou un nouveau topic correspond au pattern auquel vous êtes abonné.

Le problème avec le protocole eager ancien est que chaque consumer doit rendre toutes ses partitions avant d’être réassigné. Tout le groupe arrête le traitement pendant ce temps, ce phénomène est généralement appelé « stop-the-world ». Avec un grand groupe ou un déploiement de type rolling, les rebalances peuvent se produire continuellement.

4. Paramètres de timeout importants

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

Paramètre	Signification
session.timeout.ms	Temps maximum sans heartbeat avant que le consumer soit exclu
heartbeat.interval.ms	Intervalle d’envoi des heartbeats, généralement ≤ 1/3 de session.timeout.ms
max.poll.interval.ms	Intervalle maximal entre deux appels poll()
max-poll-records	Nombre maximal de records à chaque appel poll()

Les heartbeats sont envoyés par un thread d’arrière-plan séparé. Ainsi, le consumer reste « vivant » selon session.timeout.ms même si le thread principal est occupé. Cependant, max.poll.interval.ms ne mesure que le thread de traitement. Si un batch prend trop longtemps à traiter, le consumer est quand même expulsé du groupe.

Une erreur typique dans les logs ressemble à ceci :

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

La formule à satisfaire est :

max-poll-records × temps de traitement de 1 record  <  max.poll.interval.ms

Note : Ne « guérissez » pas en augmentant max.poll.interval.ms à plusieurs heures. Un consumer vraiment bloqué tiendra la partition pendant longtemps sans traitement. Réduisez plutôt max-poll-records, ou déplacez les tâches lourdes en dehors du flux de consommation.

5. Réduire l’impact du rebalance

5.1. Cooperative rebalancing

Avec CooperativeStickyAssignor, seules les partitions qui ont réellement besoin d’être transférées sont révoquées. Les autres consumers continuent à traiter leurs partitions pendant que le rebalance se déroule.

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

Exemple : le groupe a A (orders-0..3), puis B rejoint.

Eager:       A rend les 4 partitions -> redistribution : A {0,1}, B {2,3}
Cooperative: A rend seulement {2,3}, continue à traiter {0,1} -> B reçoit {2,3}

Note : Vous ne devez pas passer directement un groupe en cours d’exécution de eager à cooperative en un seul déploiement. Vous avez besoin d’un déploiement rolling deux fois. Première fois, déclarez les deux assignors, par exemple CooperativeStickyAssignor,RangeAssignor. La seconde fois seulement, retirez RangeAssignor.

À partir de Kafka 4.0, le nouveau protocole de rebalance (KIP-848) est GA. Vous l’activez avec group.protocol=consumer. Ensuite, la distribution est effectuée par le broker, le rebalance par défaut est déjà progressif, et les paramètres partition.assignment.strategy et session.timeout.ms côté client ne sont plus utilisés. Si votre cluster le supporte, c’est l’option à considérer en premier.

5.2. Static membership

Chaque fois que le pod redémarre, le consumer reçoit un nouveau member.id, et cela déclenche un rebalance deux fois (quitter puis rejoindre). Static membership assigne un identifiant fixe au consumer :

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

Si le consumer revient dans le délai session.timeout.ms, le coordinator réassigne exactement les mêmes partitions sans rebalance. C’est particulièrement adapté aux StatefulSets Kubernetes, où HOSTNAME est stable (billing-0, billing-1…).

Note : group.instance.id doit être unique dans le groupe. Avec concurrency > 1, Spring Kafka ajoute automatiquement un suffixe -n à chaque sous-consumer. Si deux instances utilisent le même ID, l’instance qui arrive après « expulsera » la précédente (FencedInstanceIdException). N’utilisez pas static membership avec un Deployment ayant des noms de pods aléatoires, car vous n’en tirerez aucun bénéfice.

6. Gestion des offsets lors du rebalance

Le plus grand risque du rebalance est la duplication du traitement. Le consumer A a fini de traiter un record mais ne l’a pas encore commité, la partition est transférée à B, et B lit à partir du dernier offset commité.

Spring Kafka commite par défaut après chaque batch (AckMode.BATCH) et commite automatiquement les offsets en attente avant que la partition soit révoquée. Si vous avez besoin d’intervenir, par exemple pour vider un buffer ou écrire un log, vous utilisez 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("Perte imminente de partition {}, vidage des données en mémoire", partitions);
    }

    @Override
    public void onPartitionsRevokedAfterCommit(Consumer<?, ?> consumer,
                                               Collection<TopicPartition> partitions) {
        log.info("Offset commité et partition {} restituée", partitions);
    }

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

    @Override
    public void onPartitionsLost(Consumer<?, ?> consumer,
                                 Collection<TopicPartition> partitions) {
        log.warn("Partition {} perdue sans temps de commiter", partitions);
    }
}

• onPartitionsRevokedBeforeCommit s’exécute avant que le container ne commite l’offset. C’est le moment de terminer le travail en attente.
• onPartitionsRevokedAfterCommit s’exécute après que l’offset a été commité.
• onPartitionsLost est appelé quand le consumer est expulsé du groupe, par exemple en raison du dépassement de max.poll.interval.ms. À ce stade, la partition peut appartenir à un autre consumer, donc ne commitez pas.

Enregistrez le listener dans la 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(...) préserve toute la configuration d’application.yml. La ligne suivante ajoute simplement le listener de rebalance.

Note : Peu importe la qualité de la configuration, Kafka ne garantit que at-least-once pour les consumers normaux. Rendez votre traitement idempotent, par exemple en vérifiant si orderId a déjà été payé avant d’appeler PaymentService. Ne supposez pas que le rebalance ne causera jamais de duplication.

7. Test de la distribution des partitions

Le test suivant utilise @EmbeddedKafka pour vérifier que deux consumers du même groupe partagent équitablement les 4 partitions :

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

Explication :

• @EmbeddedKafka démarre un broker en mémoire, crée le topic orders avec 4 partitions. Le broker est injecté directement en tant que paramètre de la méthode de test.
• Deux KafkaConsumer partagent group.id = "billing".
• La boucle appelle poll() alternativement sur les deux consumers, car le consumer n’effectue le rebalance que dans poll(). Si seul le consumer A poll, le rebalance ne se complètera pas. Les deux consumers s’exécutent sur le même thread mais sont deux objets séparés, donc cela ne viole pas la règle « un consumer ne doit être utilisé que par un thread ».
• Les assertions vérifient que chaque consumer a 2 partitions, ne se chevauchent pas, et que toutes les 4 partitions sont couvertes en total.

C’est aussi une bonne façon de tester différentes partition.assignment.strategy. Essayez d’ajouter un troisième consumer, de fermer un consumer, puis d’observer comment assignment() change.

8. Erreurs courantes

• Plus de consumers que de partitions : les consumers en excès resteront inactifs. Augmenter concurrency n’accélère pas le système.
• Traitement synchrone trop long dans le listener : dépasser max.poll.interval.ms cause une boucle de rebalance. Le consumer est expulsé, revient, reçoit le même batch, puis est à nouveau expulsé.
• Commit manuel dans onPartitionsLost : cause une CommitFailedException, ou pire, écrase l’offset du nouveau consumer.
• Utiliser le même group.id pour deux services différents : les deux services vont se partager les messages au lieu que chaque côté en reçoive. Chaque service indépendant doit avoir son propre groupe.
• Déploiement rolling avec protocole eager sur un grand groupe : chaque redémarrage de pod cause deux arrêts complets. Utilisez cooperative, static membership ou le nouveau protocole de Kafka 4.x.

9. Conclusion

Consumer group vous permet de scaler la lecture depuis Kafka horizontalement, avec comme limite supérieure le nombre de partitions. Le rebalance est le prix à payer pour cette flexibilité. Il se produit lorsqu’un membre entre, sort, meurt ou traite trop lentement.

Pour que le rebalance cause moins de dégâts, vous devriez faire ce qui suit :

• Configurez correctement max.poll.interval.ms et max-poll-records.
• Utilisez CooperativeStickyAssignor, ou group.protocol=consumer sur Kafka 4.x.
• Activez static membership quand l’infrastructure le permet.
• Gérez le cycle de vie des partitions via ConsumerAwareRebalanceListener.

Enfin, concevez toujours votre traitement comme idempotent, car at-least-once est la garantie par défaut que vous devez accepter.