Surveillance de Kafka : consumer lag et métriques à surveiller

1. Aperçu

Un système Kafka peut sembler « en bonne santé » : les brokers tournent, les producteurs envoient des messages. Mais les consommateurs peuvent être en retard de plusieurs millions de messages sans que personne ne le sache. La métrique qui reflète cet état s’appelle consumer lag.

Dans cet article, vous verrez comment le consumer lag est calculé et comment le mesurer de trois façons : CLI, AdminClient en Java, et Micrometer dans Spring Boot. Ensuite, nous examinerons les métriques broker, producer et consumer les plus importantes et comment configurer des alertes pour elles.

2. Qu’est-ce que le consumer lag ?

Chaque partition dans Kafka est un log en ajout seul. Il y a deux offsets auxquels vous devez faire attention :

• Log-end offset (LEO) : l’offset du message suivant qui sera écrit dans la partition.
• Committed offset : l’offset que le groupe de consommateurs a validé, c’est-à-dire la position à partir de laquelle le groupe continuera à lire en cas de redémarrage.

Le lag d’une partition se calcule ainsi :

lag = log-end offset - committed offset

Le lag du groupe de consommateurs est la somme des lags sur toutes les partitions lues par le groupe.

Un lag de 0 signifie que le consommateur est à jour. Des fluctuations légères du lag sont normales, car les messages arrivent continuellement. Un lag qui augmente régulièrement au fil du temps est un signe préoccupant. Dans ce cas, le consommateur traite plus lentement que le producteur n’écrit.

Remarque : Avec les topics utilisant des transactions ou la compaction, les offsets peuvent avoir des « vides » (enregistrements de contrôle, messages supprimés). Le lag calculé en fonction des offsets peut donc être légèrement supérieur au nombre de messages réellement à lire. Ce n’est pas une erreur.

3. Vérifier le lag avec CLI

Le plus rapide est d’utiliser le script fourni avec Kafka :

kafka-consumer-groups.sh --bootstrap-server localhost:9092 \
  --describe --group order-service

Voici le résultat :

GROUP          TOPIC   PARTITION  CURRENT-OFFSET  LOG-END-OFFSET  LAG   CONSUMER-ID       HOST
order-service  orders  0          15230           15232           2     consumer-1-a1b2   /10.0.0.5
order-service  orders  1          14980           18410           3430  consumer-1-c3d4   /10.0.0.6
order-service  orders  2          15102           15102           0     consumer-1-e5f6   /10.0.0.7

La partition 1 a 3430 messages de retard, tandis que les deux autres partitions sont presque à jour. Un tel décalage de lag est généralement dû à deux raisons. D’abord, il y a peut-être une clé « chaude » concentrée dans une partition. Ensuite, le consommateur sur l’hôte 10.0.0.6 peut avoir un problème.

CLI est très utile pour enquêter sur les problèmes. Cependant, vous ne pouvez pas exécuter cette commande chaque minute manuellement. La section suivante automatisera la mesure du lag.

4. Calculer le lag avec AdminClient

Ajoutez la dépendance kafka-clients (Spring Boot gère la version) :

<dependency>
    <groupId>org.apache.kafka</groupId>
    <artifactId>kafka-clients</artifactId>
</dependency>

Commencez par extraire le calcul dans une fonction pure pour faciliter les tests :

public final class LagCalculator {

    private LagCalculator() {}

    public static Map<TopicPartition, Long> compute(Map<TopicPartition, Long> committed,
                                                    Map<TopicPartition, Long> endOffsets) {
        Map<TopicPartition, Long> lag = new HashMap<>();
        committed.forEach((tp, committedOffset) -> {
            Long end = endOffsets.get(tp);
            if (end != null) {
                lag.put(tp, Math.max(0L, end - committedOffset));
            }
        });
        return lag;
    }
}

Cette fonction ne calcule le lag que pour les partitions ayant à la fois un committed offset et un end offset. Math.max(0L, …) a une raison d’être. Le committed offset et le end offset sont obtenus par deux requêtes différentes, donc parfois le committed offset peut être « plus récent » que le end offset vient d’être récupéré. Un résultat négatif serait sans sens, nous le ramenons donc à 0.

Vient ensuite la classe qui appelle AdminClient :

public class ConsumerLagChecker {

    private final Admin admin;

    public ConsumerLagChecker(Admin admin) {
        this.admin = admin;
    }

    public Map<TopicPartition, Long> lagOf(String groupId) throws Exception {
        Map<TopicPartition, Long> committed = admin.listConsumerGroupOffsets(groupId)
                .partitionsToOffsetAndMetadata()
                .get(10, TimeUnit.SECONDS)
                .entrySet().stream()
                .filter(e -> e.getValue() != null)
                .collect(Collectors.toMap(Map.Entry::getKey, e -> e.getValue().offset()));

        Map<TopicPartition, OffsetSpec> request = committed.keySet().stream()
                .collect(Collectors.toMap(tp -> tp, tp -> OffsetSpec.latest()));

        Map<TopicPartition, Long> endOffsets = admin.listOffsets(request)
                .all()
                .get(10, TimeUnit.SECONDS)
                .entrySet().stream()
                .collect(Collectors.toMap(Map.Entry::getKey, e -> e.getValue().offset()));

        return LagCalculator.compute(committed, endOffsets);
    }
}

Voici comment cela fonctionne étape par étape :

• listConsumerGroupOffsets retourne le committed offset du groupe. La valeur peut être null si la partition n’a jamais été commitée, nous la filtrons donc.
• listOffsets avec OffsetSpec.latest() récupère le log-end offset de chaque partition.
• Chaque appel a un délai d’attente de 10 secondes. Cela empêche le thread de se bloquer indéfiniment en cas de problème du cluster.

Utilisation :

try (Admin admin = Admin.create(Map.of(
        AdminClientConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092"))) {
    new ConsumerLagChecker(admin).lagOf("order-service")
            .forEach((tp, lag) -> System.out.println(tp + " -> " + lag));
}

orders-0 -> 2
orders-1 -> 3430
orders-2 -> 0

4.1. Tests pour LagCalculator

Utilisez JUnit 5 et AssertJ (inclus dans spring-boot-starter-test) :

class LagCalculatorTest {

    private final TopicPartition p0 = new TopicPartition("orders", 0);
    private final TopicPartition p1 = new TopicPartition("orders", 1);

    @Test
    void shouldComputeLagPerPartition() {
        Map<TopicPartition, Long> lag = LagCalculator.compute(
                Map.of(p0, 100L, p1, 50L),
                Map.of(p0, 120L, p1, 50L));

        assertThat(lag).containsEntry(p0, 20L).containsEntry(p1, 0L);
    }

    @Test
    void shouldClampNegativeLagToZero() {
        Map<TopicPartition, Long> lag = LagCalculator.compute(
                Map.of(p0, 130L),
                Map.of(p0, 120L));

        assertThat(lag).containsEntry(p0, 0L);
    }

    @Test
    void shouldSkipPartitionWithoutEndOffset() {
        Map<TopicPartition, Long> lag = LagCalculator.compute(
                Map.of(p0, 10L), Map.of());

        assertThat(lag).isEmpty();
    }
}

5. Envoyer le lag à Micrometer dans Spring Boot

Ajoutez Actuator et le registre Prometheus :

<dependency>
    <groupId>org.springframework.boot</groupId>
    <artifactId>spring-boot-starter-actuator</artifactId>
</dependency>
<dependency>
    <groupId>io.micrometer</groupId>
    <artifactId>micrometer-registry-prometheus</artifactId>
</dependency>

management.endpoints.web.exposure.include=health,prometheus

5.1. Les métriques côté client fournies

Quand MeterRegistry est disponible, Spring Boot attache automatiquement MicrometerConsumerListener à ConsumerFactory. Toutes les métriques internes du consommateur Kafka apparaissent sur /actuator/prometheus, notamment :

kafka_consumer_fetch_manager_records_lag_max{client_id="consumer-order-service-1",...} 3430.0
kafka_consumer_fetch_manager_records_lag{topic="orders",partition="1",...} 3430.0

records-lag-max est le lag maximal que ce consommateur a observé. Le client Kafka le calcule lui-même à partir de la position de récupération, pas du committed offset.

5.2. Jauge de lag mesurée de l’extérieur

Les métriques côté client ont un grand défaut. Quand le consommateur meurt, la métrique disparaît aussi. C’est juste au moment où le lag augmente le plus vite que vous n’avez plus de données. Pour cette raison, vous devriez mesurer le lag en plus à partir d’un processus indépendant en utilisant AdminClient :

@Component
public class ConsumerLagMetrics {

    private final ConsumerLagChecker checker;
    private final MeterRegistry registry;
    private final Map<TopicPartition, AtomicLong> gauges = new ConcurrentHashMap<>();

    public ConsumerLagMetrics(KafkaAdmin kafkaAdmin, MeterRegistry registry) {
        this.checker = new ConsumerLagChecker(Admin.create(kafkaAdmin.getConfigurationProperties()));
        this.registry = registry;
    }

    @Scheduled(fixedDelay = 30_000)
    public void refresh() throws Exception {
        checker.lagOf("order-service").forEach((tp, lag) ->
                gauges.computeIfAbsent(tp, this::register).set(lag));
    }

    private AtomicLong register(TopicPartition tp) {
        AtomicLong holder = new AtomicLong();
        Gauge.builder("kafka.consumer.group.lag", holder, AtomicLong::get)
                .tag("group", "order-service")
                .tag("topic", tp.topic())
                .tag("partition", String.valueOf(tp.partition()))
                .register(registry);
        return holder;
    }
}

N’oubliez pas d’activer la planification avec @EnableScheduling sur une classe @Configuration. Toutes les 30 secondes, la jauge kafka_consumer_group_lag est mise à jour. Cette valeur est basée sur le committed offset, elle reste donc correcte même si tous les consommateurs du groupe se sont arrêtés.

En production, beaucoup d’équipes utilisent des outils dédiés comme kafka-exporter ou Burrow à la place du code ci-dessus. Leur principe reste identique : comparer le committed offset avec le log-end offset.

Remarque : Ne placez pas cette jauge dans le service consommateur lui-même s’il y a plusieurs instances. Chaque instance exporterait la même valeur de lag, et le tableau de bord agrégé serait incorrect. Exécutez-le dans un service de surveillance séparé, ou seulement sur une instance.

6. Les métriques à surveiller

Le lag indique qu’il y a un problème. Les métriques ci-dessous vous aident à trouver le problème où.

6.1. Broker

Métrique (JMX)	Signification	Valeur souhaitée
UnderReplicatedPartitions	Partitions avec réplicas non synchronisées	0
OfflinePartitionsCount	Partitions sans leader	0
ActiveControllerCount	Nombre de contrôleurs actifs (cluster entier)	exactement 1
IsrShrinksPerSec / IsrExpandsPerSec	Rétrécissement/expansion ISR	près de 0, pas de fluctuations constantes
RequestHandlerAvgIdlePercent	Pourcentage d’inactivité des threads de traitement de requête	> 0.3
NetworkProcessorAvgIdlePercent	Pourcentage d’inactivité des threads réseau	> 0.3

OfflinePartitionsCount > 0 est une panne grave : cette partition ne peut pas être lue ni écrite. Un UnderReplicatedPartitions prolongé indique généralement une surcharge du broker, un disque lent ou un problème réseau.

6.2. Producer

• record-error-rate : taux d’envois échoués après épuisement des tentatives. Toute valeur différente de 0 mérite investigation.
• request-latency-avg : latence des requêtes vers le broker. S’il augmente, vous verrez le débit diminuer.
• record-retry-rate : beaucoup de tentatives signifient que le broker ou le réseau est instable.
• buffer-available-bytes : quand le buffer s’épuise, send() se bloque.

6.3. Consumer

• records-lag-max : déjà discuté ci-dessus.
• records-consumed-rate : taux de consommation. Comparez-le avec le taux auquel le producteur écrit dans le topic.
• rebalance-rate-per-hour et last-rebalance-seconds-ago : les rééquilibrages constants font « geler » le consommateur. C’est un coupable courant de l’augmentation du lag.
• commit-latency-avg : commit lent réduit le débit, surtout en commit synchrone.
• last-poll-seconds-ago : si cela dépasse max.poll.interval.ms, le consommateur est expulsé du groupe.

7. Définir des alertes pertinentes

L’erreur la plus courante est de définir un seuil absolu comme « lag > 1000 alors alerter ». Avec un topic à 50.000 messages/seconde, 1000 n’a aucun sens. Avec un topic de quelques messages/minute, 1000 pourrait déjà signifier des heures de retard.

Il y a deux meilleures approches. La première est d’alerter quand le lag augmente continuellement :

groups:
  - name: kafka-consumer
    rules:
      - alert: KafkaConsumerLagGrowing
        expr: sum by (group, topic) (deriv(kafka_consumer_group_lag[10m])) > 0
          and sum by (group, topic) (kafka_consumer_group_lag) > 500
        for: 15m
        labels:
          severity: warning
        annotations:
          summary: "Le lag de {{ $labels.group }} sur {{ $labels.topic }} augmente continuellement depuis 15 minutes"

Cette règle ne s’active que lorsque les deux conditions sont vraies : le lag tend à augmenter (deriv > 0) et a dépassé un seuil minimum. La condition doit persister pendant 15 minutes. De cette façon, vous ignorez les pics courts après un déploiement.

La deuxième approche consiste à convertir le lag en temps. Divisez le lag par le taux de consommation pour estimer le retard du consommateur en secondes. Ce nombre est directement lié au SLA. Par exemple, « les commandes doivent être traitées en 2 minutes » est bien plus compréhensible que « lag sous 5000 ».

De plus, vous devez avoir une alerte distincte pour le cas où la métrique disparaît, par exemple en utilisant absent(kafka_consumer_group_lag). L’absence de données ne signifie pas que tout va bien.

8. Erreurs courantes

• Compter uniquement sur records-lag-max. Quand le consommateur meurt, cette métrique disparaît. Ayez toujours une source de mesure du lag externe.
• Lire le lag d’un groupe sans consommateur actif. CLI affiche toujours le committed offset, mais la colonne CONSUMER-ID est vide. Une augmentation du lag ici signifie que personne n’effectue le traitement.
• Augmenter le nombre de consommateurs au-delà du nombre de partitions. Les consommateurs excédentaires resteront inactifs, le lag ne diminuera pas. Pour augmenter le parallélisme, vous devez augmenter le nombre de partitions.
• Ignorer le rééquilibrage. Le traitement d’un batch trop long dépasse max.poll.interval.ms. Le consommateur est retiré, le groupe se rééquilibre, puis se répète. Le lag augmente en dents de scie. Réduisez max.poll.records ou déplacez le traitement lourd vers un autre thread.
• Appeler AdminClient trop souvent. Interroger les offsets toutes les secondes pour des centaines de groupes crée une charge inutile sur le broker. Un cycle de 15-60 secondes est suffisant pour la plupart des systèmes.

9. Conclusion

Le consumer lag est la métrique la plus directe pour voir si le système Kafka suit les données. Le lag se calcule en soustrayant le committed offset du log-end offset. Vous pouvez visualiser le lag avec kafka-consumer-groups.sh, le calculer avec AdminClient ou l’exporter vers Prometheus via Micrometer.

Pour une surveillance efficace, combinez les métriques côté client avec une source de mesure indépendante. En parallèle du lag, suivez les métriques du broker comme UnderReplicatedPartitions et OfflinePartitionsCount, ainsi que la fréquence de rééquilibrage côté consommateur. Enfin, alertez selon la tendance et le temps de latence plutôt que selon un nombre de lag fixe. De cette façon, vous éviterez les faux positifs tout en ne manquant pas les vrais problèmes.