Retards retry Spring Kafka : comment corriger

Retry non bloquant dans Spring Kafka : Diagnostic et résolution des problèmes de lag

Lors de la mise en œuvre d'une retry asynchrone pour Kafka avec @RetryableTopic dans Spring Boot 3.2 et Spring Kafka 3.1, un lag s'accumule sur les topics de retry sous charge. Le topic principal traite les messages uniformément, mais retry-0 accumule des messages pendant 10 à 12 minutes, puis se vide partiellement avant de se bloquer à nouveau. Cela entraîne un traitement inégal.

Les métriques montrent des pics sporadiques de requêtes fetch au lieu d'un polling constant. Le consumer interroge périodiquement le broker pour les heartbeats, mais la lecture réelle se produit de manière sporadique avec une latence élevée.

Métriques clés pour l'analyse

Pour le diagnostic, collectez ces métriques Spring Kafka :

• kafka_consumer_time_between_poll_avg / max — intervalles entre les polls.
• kafka_consumer_last_poll_seconds_ago — temps écoulé depuis le dernier poll.
• kafka_consumer_poll_idle_ratio_avg — proportion de polls inactifs.
• kafka_consumer_fetch_manager_fetch_rate / fetch_latency_avg — vitesse et latence des fetch.
• spring_kafka_listener_seconds_count / sum — temps de traitement du listener.

Les graphiques du fetch-rate et des invocations de listener confirment : le traitement s'active par à-coups, le poll idle est élevé.

Diagnostic via Thread Dump

Le thread dump a révélé ~180 threads en TIMED_WAITING sur ScheduledThreadPoolExecutor$DelayedWorkQueue.take. Cela indique une surcharge du scheduler, utilisé pour la reprise différée des partitions.

Mécanisme de pause dans RetryableTopic

@RetryableTopic utilise ContainerPartitionPausingBackOffManager : lors du traitement du premier message de retry, il vérifie le backoff, met en pause la partition pour now() + delay, et planifie la reprise via ListenerContainerPauseService.

Code de pause :

private final TaskScheduler scheduler;

public void pausePartition(MessageListenerContainer messageListenerContainer, 
                           TopicPartition partition,
                           Duration pauseDuration) {
    Instant resumeAt = Instant.now().plusMillis(pauseDuration.toMillis());
    messageListenerContainer.pausePartition(partition);
    this.scheduler.schedule(() -> {
        messageListenerContainer.resumePartition(partition);
    }, resumeAt);
}

Sous forte charge, la file d'attente des tâches du scheduler déborde. Par défaut, spring.task.scheduling.pool.size=1, créant un goulot d'étranglement : les tâches de reprise s'accumulent, le fetch ne démarre pas à temps.

Solution : Augmenter le pool du scheduler

Définissez spring.task.scheduling.pool.size=10 dans application.properties.

Résultats :

• Le lag de retry-0 s'est stabilisé à un niveau acceptable (ne tombe pas à 0 en raison du backoff).
• Les intervalles de poll ont raccourci, le fetch-rate a augmenté.
• Les invocations de listener sont devenues constantes.

Graphique du lag après correction : reste plat sans croissance. La latence de fetch a diminué, les polls sont plus fréquents.

Points clés à retenir

• Scheduler limité : le pool par défaut d'un thread ne peut pas gérer les pauses/reprises massives sous charge de retry.
• Métriques d'abord : surveillez fetch-rate, poll-idle, listener-time avant le thread dump.
• Transactions secondaires : ack-mode=record, max.poll.records=1 n'affectent pas le lag.
• Concurrence OK : Spring Kafka s'adapte correctement par partitions, même avec des topics multipliés.
• Alternative : pour les charges élevées, envisagez une retry personnalisée avec vérifications d'horodatage dans une boucle au lieu de pauses.

Recommandations de surveillance

Ajoutez des tableaux de bord pour :

• Taux de commit/rollback pour exactly-once.
• Taille de la file d'attente du scheduler (métrique personnalisée).
• Événements de pause/reprise des partitions.
• Delta entre l'horodatage de backoff et l'heure actuelle.

Testez sous charge avant la production. Le code source de Spring Kafka fournit un aperçu complet des mécanismes.

— Editorial Team