Résolution des problèmes de latence élevée des consommateurs dans votre pipeline Kafka

Une latence élevée des consommateurs signifie que des enregistrements sont disponibles dans Kafka avant que votre application ait fini de les utiliser. Ce retard peut se manifester par un décalage du consommateur, des tableaux de bord obsolètes, des alertes retardées ou des tâches en aval qui manquent leur fenêtre prévue. La partie dérangeante est que Kafka peut être en bonne santé alors que le pipeline est toujours lent. Le consommateur peut attendre une base de données, faire trop de travail par sondage, valider les offsets trop souvent, ou lutter contre des rééquilibrages causés par de longues pauses de traitement.

Ce guide aborde d'abord le côté consommateur car c'est là que la plupart des incidents de latence deviennent visibles. L'objectif est de trouver le segment lent avant de modifier les paramètres.

Comprendre le décalage du consommateur et la latence

Le décalage du consommateur est la métrique principale indiquant des problèmes de latence. Il représente la différence entre le dernier offset produit dans une partition et l'offset que le groupe de consommateurs a lu et validé avec succès. Un décalage élevé signifie que vos consommateurs prennent du retard.

Métriques clés à surveiller :

Décalage du consommateur : Nombre total de messages non lus par partition.
Taux de récupération vs. Taux de production : Si le taux de récupération du consommateur est constamment inférieur au taux de production, le décalage augmentera.
Latence de validation : Temps nécessaire aux consommateurs pour enregistrer leur progression.

Phase 1 : Analyser le comportement de récupération du consommateur

La raison la plus courante d'une latence élevée est une récupération inefficace des données. Les consommateurs doivent extraire les données des courtiers, et si la configuration est sous-optimale, ils peuvent passer trop de temps à attendre ou à récupérer trop peu de données.

Réglage de `fetch.min.bytes` et `fetch.max.wait.ms`

Ces deux paramètres influencent directement la quantité de données qu'un consommateur attend avant de demander une récupération, équilibrant la latence et le débit.

fetch.min.bytes : La quantité minimale de données que le courtier doit retourner (en octets). Une valeur plus grande encourage le regroupement par lots, ce qui augmente le débit mais peut légèrement augmenter la latence si la taille requise n'est pas immédiatement disponible.
- Meilleure pratique : Pour les pipelines à haut débit et faible latence, vous pouvez garder cette valeur relativement basse (par exemple, 1 octet) pour garantir un retour immédiat, ou l'augmenter si des goulots d'étranglement de débit sont observés.
fetch.max.wait.ms : Combien de temps le courtier attendra pour accumuler fetch.min.bytes avant de répondre. Une attente plus longue maximise la taille du lot mais ajoute directement à la latence si le volume requis n'est pas présent.
- Compromis : Réduire ce temps (par exemple, de 500 ms par défaut à 50 ms) réduit considérablement la latence mais peut entraîner des récupérations plus petites et moins efficaces.

Ajustement de `max.poll.records`

Ce paramètre contrôle le nombre d'enregistrements retournés dans un seul appel Consumer.poll().

max.poll.records=500

Si max.poll.records est défini trop bas, le consommateur passe un temps excessif à boucler sur les appels poll() sans traiter des volumes de données significatifs, augmentant la surcharge. S'il est trop élevé, le traitement du grand lot peut prendre plus de temps que le délai d'expiration de la session, provoquant des rééquilibrages inutiles.

Conseil pratique : Commencez avec une valeur modérée comme 100 à 500 et surveillez le temps de traitement réel pour chaque sondage. Ne réglez pas cela par supposition. Si un lot de 500 enregistrements prend quatre minutes parce que chaque enregistrement écrit sur une API lente, augmenter max.poll.records rendra le consommateur moins stable, pas plus rapide.

Phase 2 : Enquêter sur le temps de traitement et les validations

Même si les données sont récupérées rapidement, une latence élevée résulte si le temps passé à traiter le lot récupéré dépasse le temps entre les récupérations.

Goulots d'étranglement dans la logique de traitement

Si la logique de votre application consommateur implique des appels externes lourds (par exemple, écritures en base de données, recherches API) qui ne sont pas parallélisés dans la boucle de consommation, le temps de traitement va exploser.

Étapes de dépannage :

Mesurer le temps de traitement : Utilisez des métriques pour suivre le temps réel écoulé entre la réception du lot et la fin de toutes les opérations en aval avant la validation.
Parallélisation : Si le traitement est lent, envisagez d'utiliser des pools de threads internes dans votre application consommateur pour traiter les enregistrements simultanément après leur récupération, mais avant de valider les offsets.

Révision de la stratégie de validation

La validation des offsets peut introduire de la latence si elle se produit trop fréquemment, car chaque validation nécessite une coordination avec Kafka. Le plus grand risque, cependant, est généralement l'exactitude. Valider trop tôt peut entraîner une perte de travail après un crash. Valider trop tard peut rejouer du travail après un crash.

enable.auto.commit : Convient pour les lecteurs simples, les expériences et les pipelines non critiques. Pour les consommateurs de production qui mettent à jour des bases de données, appellent des API ou publient des événements dérivés, les validations manuelles sont généralement plus faciles à raisonner.
auto.commit.interval.ms : Cela dicte la fréquence de validation des offsets (par défaut 5 secondes).

Si le traitement est rapide et stable, un intervalle plus long (par exemple, 10 à 30 secondes) réduit la surcharge de validation. Cependant, si votre application plante fréquemment, un intervalle plus court préserve plus de travail en cours, bien qu'il augmente le trafic réseau et la latence potentielle.

Avertissement sur les validations manuelles : Si vous utilisez des validations manuelles (enable.auto.commit=false), assurez-vous que commitSync() est utilisé avec parcimonie. commitSync() bloque le thread du consommateur jusqu'à ce que la validation soit acquittée, impactant sévèrement la latence si elle est appelée après chaque message ou petit lot.

Phase 3 : Mise à l'échelle et allocation des ressources

Si les configurations semblent optimisées, le problème fondamental pourrait être un parallélisme insuffisant ou une saturation des ressources.

Mise à l'échelle des threads consommateurs

Les consommateurs Kafka montent en échelle en augmentant le nombre d'instances de consommateurs dans un groupe, jusqu'au nombre de partitions qu'ils consomment. Si vous avez 20 partitions et 5 instances de consommateurs, Kafka attribuera normalement plusieurs partitions à chaque consommateur. Cela peut être parfaitement sain. La limite est qu'une partition dans un groupe de consommateurs est traitée par un seul consommateur à la fois, donc une seule partition chaude ne peut pas être corrigée simplement en ajoutant plus de membres au groupe.

Règle générale : Le nombre d'instances de consommateurs ne doit généralement pas dépasser le nombre de partitions de tous les sujets auxquels ils sont abonnés. Plus d'instances que de partitions entraîne des threads inactifs.

Santé du courtier et du réseau

La latence peut provenir de l'extérieur du code consommateur :

CPU/Mémoire du courtier : Si les courtiers sont surchargés, leur temps de réponse aux demandes de récupération augmente, provoquant des délais d'attente et des retards du consommateur.
Saturation du réseau : Un trafic réseau élevé entre les consommateurs et les courtiers peut ralentir les transferts TCP, en particulier lors de la récupération de gros lots.

Utilisez des outils de surveillance pour vérifier l'utilisation du CPU du courtier et les E/S réseau pendant les périodes de fort décalage.

Lire la forme du décalage

La forme du décalage vous indique où chercher. Une seule partition en retard signifie généralement que le problème est étroit. Peut-être qu'une clé achemine trop de trafic vers une partition. Peut-être qu'un enregistrement déclenche un chemin de code lent. Peut-être que l'hôte exécutant cette affectation de partition est défectueux. Dans cette situation, ajouter plus de consommateurs peut ne rien faire car Kafka ne peut pas diviser cette partition unique entre plusieurs consommateurs du même groupe.

Un décalage uniforme sur toutes les partitions pointe vers une limite partagée. Le service peut avoir besoin de plus d'instances, la base de données en aval peut être saturée, ou les courtiers peuvent être lents à servir les récupérations. Si le décalage augmente à la même heure chaque jour, recherchez des tâches planifiées, des producteurs par lots, la pression de compactage, des sauvegardes ou des événements de mise à l'échelle automatique. La latence Kafka est souvent un effet secondaire de quelque chose en dehors de Kafka.

Séparez également "enregistrements en retard" de "temps de retard". Un sujet avec de petits événements peut montrer un nombre effrayant d'enregistrements mais rattraper son retard en quelques secondes. Un sujet avec de gros enregistrements ou un traitement coûteux peut montrer un nombre de décalage plus petit mais représenter des minutes de retard métier. Si votre pile de surveillance peut estimer le temps de décalage à partir des horodatages des enregistrements, représentez-le graphiquement à côté du décalage d'offset. Si elle ne le peut pas, échantillonnez quelques enregistrements avec kafka-console-consumer.sh dans un groupe temporaire et comparez les horodatages des événements avec l'heure réelle.

Correctifs courants qui se retournent contre vous

Le premier mauvais correctif est d'augmenter max.poll.interval.ms jusqu'à ce que les rééquilibrages cessent. Cela peut être valide lorsque le traitement est naturellement long, mais cela peut aussi cacher un consommateur bloqué plus longtemps. Si le consommateur est bloqué sur un appel en aval pendant vingt minutes, un intervalle plus long retarde la récupération.

Le deuxième mauvais correctif est d'augmenter les partitions pendant un incident sans vérifier le modèle de clé. Plus de partitions peuvent améliorer le parallélisme futur, mais cela modifie l'affectation des partitions pour les nouveaux enregistrements et peut affecter les hypothèses de classement. Cela ne divise pas non plus les enregistrements qui se trouvent déjà dans les partitions existantes.

Le troisième mauvais correctif est de passer à des réinitialisations d'offset --to-latest pour rendre les tableaux de bord verts. Cela saute du travail. Parfois, l'entreprise l'accepte, comme pour les événements d'analyse jetables lors d'une panne. Pour la facturation, l'exécution, les alertes de sécurité ou les changements d'état visibles par l'utilisateur, sauter des enregistrements en retard peut créer un incident beaucoup plus important que la latence elle-même.

Quand la mise à l'échelle des consommateurs aide

La mise à l'échelle aide lorsque le groupe a plus de partitions que de consommateurs actifs et que le travail est raisonnablement équilibré entre ces partitions. Si un sujet a 24 partitions et 6 consommateurs, passer à 12 consommateurs peut réduire la latence car chaque instance gère moins de partitions. Passer de 24 consommateurs à 40 consommateurs n'aidera pas ce même groupe ; les consommateurs supplémentaires resteront inactifs car il n'y a que 24 partitions à attribuer.

La mise à l'échelle n'aide pas beaucoup lorsque tous les consommateurs attendent la même dépendance saturée. Si chaque consommateur écrit dans une table de base de données qui est déjà verrouillée, plus de consommateurs peuvent augmenter la contention et aggraver la latence. Dans ce cas, le regroupement par lots des écritures, la modification des index, l'ajout de contre-pression ou la séparation des charges de travail chaudes peuvent être plus importants que les paramètres Kafka.

Surveillez les rééquilibrages lors de la mise à l'échelle. Un déploiement progressif qui démarre et arrête les consommateurs de manière trop agressive peut créer des pics de latence même lorsque le nombre final de répliques est correct. L'appartenance statique avec group.instance.id peut réduire les mouvements de partition inutiles pour certains services à longue durée d'exécution, mais nécessite une gestion minutieuse de l'identité des instances. Le rééquilibrage coopératif peut également réduire les perturbations par rapport au rééquilibrage impatient, selon la configuration du client et de l'assignateur.

Quand la latence est vraiment un risque de rétention

Une latence élevée devient urgente lorsque le décalage approche de la fenêtre de rétention du sujet. Kafka supprime les anciens segments en fonction de la politique de rétention, et non du fait que tous les consommateurs les aient lus. Si un consommateur a six heures de retard sur un sujet qui conserve sept jours de données, vous avez le temps de réparer l'application. S'il a six jours de retard sur ce même sujet, vous avez besoin d'un plan de récupération avant que les enregistrements non lus les plus anciens n'expirent.

Lors de ce type d'incident, estimez le taux de rattrapage. Si le groupe réduit le décalage de 50 000 enregistrements par minute et qu'il a 5 millions d'enregistrements de retard, il peut rattraper son retard dans une fenêtre réalisable. Si le décalage continue de croître, le groupe ne récupère pas. Vous devrez peut-être mettre les producteurs en pause, ajouter une capacité de consommateur temporaire, supprimer une dépendance en aval lente du chemin chaud, ou prendre une décision consciente sur les données qui peuvent être ignorées.

La meilleure surveillance de la latence des consommateurs montre à la fois le retard opérationnel et la marge de rétention. "Ce groupe a 20 minutes de retard" est utile. "Ce groupe a 18 heures avant l'expiration des données non lues" est le nombre qui fait venir les bonnes personnes dans la salle.

Un runbook pratique sur la latence

Commencez par le décalage au niveau de la partition, pas seulement le décalage total :

kafka-consumer-groups.sh --bootstrap-server kafka-1:9092 --describe --group realtime-enricher

Si le décalage est concentré dans une seule partition, recherchez un déséquilibre de clé ou une instance de consommateur plus lente que les autres. Si le décalage est uniformément réparti, recherchez un goulot d'étranglement partagé : trop peu de consommateurs, des appels en aval lents, une latence de récupération du courtier, ou un pic de producteur qui a dépassé la capacité normale. Exécutez la commande deux fois, à une ou deux minutes d'intervalle, afin de savoir si le groupe rattrape son retard ou prend encore plus de retard.

Ensuite, mesurez quatre temporisations dans l'application : le temps d'attente dans poll(), le temps passé à traiter les enregistrements retournés, le temps passé à écrire dans les systèmes en aval, et le temps passé à valider les offsets. Ces chiffres vous indiquent quel paramètre est important. Si poll() attend trop longtemps alors que le trafic est faible, réduisez fetch.max.wait.ms ou gardez fetch.min.bytes bas. Si le traitement domine, les paramètres de récupération Kafka sont une distraction. Si les validations dominent, arrêtez de valider chaque enregistrement avec des validations synchrones.

Pour les services à faible latence, je commence généralement par un regroupement par lots de récupération conservateur, puis je l'augmente uniquement lorsque la surcharge du courtier ou du réseau est clairement le problème :

fetch.min.bytes=1
fetch.max.wait.ms=50
max.poll.records=100
enable.auto.commit=false

Ce n'est pas une configuration universelle optimale. C'est un point de départ lisible. Un consommateur ETL par lots peut préférer des récupérations plus grandes et des max.poll.records plus grands. Un service de notation de fraude peut préférer des lots plus petits car un appel API lent peut bloquer tout le lot.

Soyez particulièrement prudent lors de l'ajout de threads de travail après poll(). Le traitement parallèle peut aider, mais les offsets ne doivent être validés qu'après que tous les enregistrements antérieurs pour la partition concernée sont traités en toute sécurité. Si les threads de travail se terminent dans le désordre et que vous validez l'offset le plus élevé trop tôt, un crash peut ignorer silencieusement des enregistrements qui étaient encore en cours. Un modèle courant consiste à suivre l'achèvement par partition et à valider uniquement l'offset contigu le plus élevé terminé.

La liste de contrôle est simple : inspectez le décalage par partition, mesurez les phases de l'application, réglez le comportement de récupération uniquement lorsque le comportement de récupération est le problème, et mettez à l'échelle les consommateurs uniquement lorsqu'il y a suffisamment de partitions pour utiliser les instances supplémentaires. Cet ordre évite la plupart des travaux de réglage inutiles.