Files de messages et Kafka : découpler pour tenir

Le centre de tri ne fait pas la tournée

En décembre, une ville reçoit plusieurs fois plus de courrier qu'en juin. Personne n'embauche plusieurs fois plus de facteurs pour un seul mois.

Ce qui absorbe le pic, c'est le centre de tri. Les sacs arrivent quand ils arrivent, on les vide dans des casiers, et les tournées partent au rythme des facteurs. L'expéditeur, lui, n'a jamais attendu que le destinataire soit chez lui pour poster sa lettre.

Deux rythmes différents, un tampon entre les deux, et une pile de courrier qui monte ou qui descend selon la saison. C'est très exactement ce qu'une file de messages installe dans un système.

Une file ne rend aucun traitement plus rapide. Elle rend le producteur indépendant du rythme, de la disponibilité et des pannes du consommateur.

Le billet Messagerie inter-services et supervision centralisée a comparé les quatre modes de communication entre services sur cinq critères. Celui-ci prend le seul qui change vraiment la forme du système, et va au fond de son représentant le plus répandu : le log distribué, partitionné, rejouable.

Vue d'ensemble : ce qu'une file achète, et ce qu'elle coûte

Trois choses justifient qu'on ajoute un intermédiaire entre deux services.

Absorber les pics. Le producteur écrit à son rythme, le consommateur lit au sien. Entre les deux, le courtier stocke. Un pic de dix minutes se traite en trente, sans que personne renvoie d'erreur.

Découpler les rythmes. Un service qui publie un événement ne connaît plus ses destinataires. On en branche un cinquième sans toucher au premier.

Survivre à l'indisponibilité du consommateur. Le service de facturation redémarre pendant vingt minutes. Les messages attendent dans le casier au lieu de se perdre en erreurs de délai dépassé.

Rien de tout cela n'est gratuit. La latence de bout en bout augmente, l'ordre global disparaît, les doublons deviennent une hypothèse de travail permanente, et il y a un composant de plus à sauvegarder, sécuriser et superviser à trois heures du matin.

La suite du billet déroule sept points, dans cet ordre : la différence entre une file et un log, la manière dont un message trouve sa destination, l'anatomie de Kafka, la clé de partition qui décide de l'ordre, les trois garanties de livraison, la métrique unique qu'on surveille en exploitation, et les cas où il vaut mieux se passer de file.

Le découplage est le produit principal de la file. Tout le reste, y compris la performance, en découle.

File classique ou log distribué

Au centre de tri, deux gestes cohabitent. Le casier qu'on vide : la lettre part, le casier redevient vide, personne ne pourra relire ce qui y était. Et le registre d'arrivée, qu'on garde des mois et qu'on peut rouvrir à n'importe quelle page.

La file classique tient du casier. Un message est déposé, un consommateur le retire, l'acquitte, et le courtier l'efface. Plusieurs consommateurs concurrents se partagent le travail, chacun prenant le message suivant. Le courtier maintient un état par message : livré, acquitté, à réessayer, mort. RabbitMQ, ActiveMQ ou Amazon SQS travaillent ainsi.

Le log distribué tient du registre. Les messages sont ajoutés en fin de fichier, jamais retirés à la lecture, et conservés selon une politique de rétention indépendante de leur consommation. Le courtier ne suit pas l'état de chaque message : il ne connaît qu'un numéro de position par consommateur. Kafka et Pulsar travaillent ainsi.

La conséquence pratique est plus lourde qu'il n'y paraît. Dans un log, brancher un nouveau consommateur sur des données déjà publiées ne demande aucune coopération du producteur. On relit depuis le début, on reconstruit un état, on rejoue un bug corrigé sur trois semaines de trafic. C'est le socle technique de l'event sourcing, traité dans Microservices, Event Sourcing et CQRS.

Une question tranche ce choix bien avant celle du débit : quelqu'un aura-t-il besoin, un jour, de relire ce qui est déjà passé.

Les échanges : comment un message trouve son casier

Au tri, la règle d'aiguillage est écrite sur le casier, pas sur la lettre. Le facteur ne décide de rien : il applique le plan de tri.

Dans le vocabulaire AMQP, celui de RabbitMQ, un producteur ne publie jamais dans une file. Il publie dans un échange, avec une clé de routage. L'échange est relié aux files par des liaisons, chacune portant sa propre clé. Le type de l'échange décide de la façon dont les deux clés se comparent.

Attention au piège de vocabulaire : l'échange topic d'AMQP et le topic de Kafka n'ont aucun rapport. Le premier est une règle d'aiguillage par motif. Le second est simplement le nom d'un log, et l'aiguillage y est décidé par le hachage d'une clé, pas par un motif.

Quand l'aiguillage vit dans le courtier, changer une règle de routage ne demande aucun déploiement. Quand il vit dans la clé de partition, il est figé dans le code du producteur.

Kafka, brique par brique

Six mots suffisent à décrire Kafka, et le reste en découle.

Le topic est un log nommé, un flux d'événements de même nature : commandes, paiements, clics.

La partition est le découpage physique de ce log. Un topic à quatre partitions, ce sont quatre journaux distincts, écrits uniquement par ajout en fin de fichier, sur quatre courtiers potentiellement différents. C'est l'unité de parallélisme, et c'est elle qui fait tenir le débit.

Le décalage (offset) est le numéro d'ordre d'un message dans sa partition. Il ne bouge jamais. Un message publié reste à la position 481 234 de la partition 2 jusqu'à son expiration.

Le producteur écrit en fin de partition. Il choisit la partition par la clé du message, ou à tour de rôle s'il n'y a pas de clé.

Le groupe de consommateurs est l'unité d'abonnement. Kafka attribue chaque partition à une seule instance du groupe. Deux groupes différents lisent les mêmes données sans se voir, chacun avec son propre curseur.

La réplication copie chaque partition sur plusieurs courtiers. L'une des copies est le chef de partition : elle seule sert les lectures et les écritures, les autres suivent. Si le chef tombe, une réplique synchronisée prend sa place.

Le schéma montre le plafond qu'on découvre en production : une partition n'est lue que par une seule instance d'un groupe. Le groupe facturation peut encore monter à quatre instances, une par partition. La cinquième resterait inactive, quel que soit le retard à rattraper.

Un mot d'actualité pour éviter une lecture périmée : la coordination du cluster reposait historiquement sur ZooKeeper, remplacé par le protocole KRaft intégré, et Apache Kafka 4.0 est la première version à fonctionner sans ZooKeeper du tout.

Combien de partitions, et combien de disque

Le nombre de partitions se calcule, il ne se devine pas. Prenons un flux de 20 000 messages par seconde, un message d'un kilooctet, un consommateur capable de traiter 2 000 messages par seconde par instance, et sept jours de rétention.

20 000 msg/s x 1 Ko                    =      20 Mo/s de flux entrant
20 000 msg/s / 2 000 msg/s par instance =         10 instances nécessaires
                                        =>   10 partitions au minimum

20 Mo/s x 86 400 s                     = 1 728 000 Mo/jour, soit 1,73 To/jour
1,73 To x 7 jours de rétention         =      12,1 To de données utiles
12,1 To x 3 (facteur de réplication)   =      36,3 To de disque à provisionner

Trente-six téraoctets pour vingt mégaoctets par seconde. Le facteur de réplication commande à lui seul la facture de stockage.

Trop peu de partitions plafonne le débit de consommation. Trop de partitions allonge le rééquilibrage et multiplie les fichiers ouverts sur chaque courtier. Le bon nombre vient du calcul ci-dessus, pas d'une valeur par défaut.

L'ordre vit dans la partition, donc dans la clé

Au tri postal, on confie toute une rue au même facteur. Pas par sentiment : parce que c'est la seule façon de garantir que les trois lettres du numéro 12 arrivent dans l'ordre où elles ont été postées.

Kafka garantit l'ordre à l'intérieur d'une partition, et nulle part ailleurs. Un topic à dix partitions ne fournit aucun ordre global, et il n'existe aucun réglage pour l'obtenir.

L'ordre se pilote donc par la clé. Le producteur calcule le hachage de la clé du message, modulo le nombre de partitions, et obtient la destination. Même clé, même partition, donc même ordre.

L'exemple canonique est le compte utilisateur. compte_créé, adresse_modifiée, compte_fermé doivent arriver dans cet ordre au consommateur, sinon l'état reconstruit est faux. Il suffit de publier ces trois événements avec l'identifiant du compte comme clé.

Deux pièges viennent avec cette décision.

Le changement de nombre de partitions casse l'affectation. Passer de 10 à 12 partitions change le résultat du modulo pour la plupart des clés. Pendant la transition, les anciens et les nouveaux messages d'une même clé vivent dans deux partitions différentes, et l'ordre est perdu pour de bon sur cette fenêtre.

Une clé mal choisie fabrique une partition brûlante. Si 40 % du trafic porte l'identifiant d'un seul client grand compte, 40 % du flux atterrit sur une seule partition, donc sur un seul courtier et une seule instance de consommateur. Le reste du cluster regarde.

La clé de partition est une décision de conception, pas un réglage d'exploitation. Elle fixe à la fois l'ordre garanti et la répartition de la charge, et elle est très difficile à corriger après coup.

Les garanties de livraison

Trois régimes existent, et pas un de plus : la lettre simple, la lettre suivie qu'on renvoie au moindre doute, et le recommandé avec accusé de réception.

Au plus une fois. Le consommateur valide son décalage avant de traiter le message, le producteur n'attend pas d'accusé du courtier. Rapide, simple, et un message peut disparaître sans que personne le sache.

Au moins une fois. Le producteur réessaie tant qu'il n'a pas d'accusé, le consommateur ne valide son décalage qu'après avoir terminé son traitement. Aucun message n'est perdu, certains sont traités deux fois. C'est le régime par défaut de la plupart des systèmes en production.

Exactement une fois. Kafka le propose depuis la version 0.11 (2017) : producteur idempotent, qui déduplique les réémissions, et transactions qui rendent atomiques une lecture, un traitement et une écriture à l'intérieur de Kafka. Depuis Kafka 3.0, le producteur est idempotent par défaut.

La nuance décide de beaucoup de choses : la garantie ne franchit pas la frontière de Kafka. Si ton consommateur écrit dans une base externe ou appelle une API de paiement, tu retombes en pratique sur du « au moins une fois », et c'est à ton code de rendre le traitement idempotent. C'est le sujet du billet suivant, Limitation de débit et idempotence.

# producteur : ne rien perdre
acks=all
enable.idempotence=true
# topic : accepter l'écriture seulement si deux répliques suivent
min.insync.replicas=2

Avec un facteur de réplication de 3 et min.insync.replicas=2, le cluster continue d'accepter les écritures avec un courtier en panne, et les refuse avec deux. Le refus est ici le comportement souhaité : mieux vaut une erreur visible qu'une écriture non répliquée.

« Exactement une fois » décrit une propriété de bout en bout que le courtier seul ne peut pas fournir. Sans consommateur idempotent, la promesse s'arrête à la frontière du cluster.

Le décalage du consommateur, la métrique qui compte

Dans un centre de tri, un seul indicateur dit si la nuit s'est bien passée : la hauteur de la pile de sacs non traités au petit matin.

L'équivalent s'appelle le décalage du consommateur : la différence entre le dernier décalage publié dans une partition et le dernier décalage validé par le groupe. Il se lit partition par partition.

Un décalage stable, même élevé, décrit un système qui tient. Un décalage qui croît linéairement décrit un système qui a déjà perdu, et l'heure de la rupture se calcule.

kafka-consumer-groups.sh --bootstrap-server broker:9092 \
  --describe --group facturation

Le rattrapage se pose comme le reste, avec une soustraction et une division.

Décalage constaté        : 4 000 000 messages
Débit de production      :    18 000 messages/s
Débit de consommation    :    20 000 messages/s
Vidange nette            :     2 000 messages/s
4 000 000 / 2 000        =     2 000 s, soit environ 33 minutes de rattrapage

Si la consommation était descendue à 17 000 messages par seconde, la vidange nette deviendrait négative et le calcul n'aurait plus de solution. La rétention devient alors le compte à rebours : à sept jours de conservation, les plus vieux messages non traités commencent à disparaître.

Deux alertes suffisent en pratique. Une sur la valeur absolue du décalage maximum, une sur sa dérivée. La seconde prévient plusieurs heures avant la première.

Un décalage qui monte désigne presque toujours un consommateur ralenti par sa base de données. Le courtier se contente de rendre le problème visible plus tôt.

Quand ne pas mettre de file

Un centre de tri pour trois lettres par jour entre deux bureaux voisins coûte plus cher que le facteur qui traverse la rue.

Quatre situations où l'appel synchrone reste le bon choix.

L'appelant a besoin de la réponse pour continuer. Vérifier un mot de passe, calculer un prix, valider un panier. Poser une file ici revient à réinventer une requête synchrone avec deux topics et un identifiant de corrélation, en moins fiable.

Le volume ne le justifie pas. Deux services, quelques appels par minute, un couplage assumé. Le courtier ajoute une infrastructure, une supervision et une compétence d'exploitation pour un bénéfice nul.

L'atomicité est exigée entre deux systèmes. Une file ne fournit pas de transaction répartie. Le sujet a ses propres réponses, traitées dans Transactions distribuées : 2PC, 3PC, SAGA et verrous.

L'ordre global est indispensable. Un log partitionné ne le donne pas. Le forcer sur une seule partition supprime le parallélisme, et donc la raison d'avoir choisi Kafka.

Le critère le plus souvent oublié tient à l'équipe : personne n'y sait diagnostiquer un rééquilibrage bloqué à trois heures du matin. Un courtier qu'on ne sait pas exploiter est moins fiable qu'une API bien faite.

Scénario B : Kafka chez LinkedIn

Kafka n'est pas né dans un laboratoire. Il est né chez LinkedIn, d'un problème que les outils de l'époque ne couvraient pas.

Le papier « Kafka: a Distributed Messaging System for Log Processing », présenté par Jay Kreps, Neha Narkhede et Jun Rao à l'atelier NetDB en 2011, pose le constat d'origine. D'un côté, les courtiers de messages classiques offraient des garanties fines, au prix d'un état maintenu message par message côté serveur, et s'écroulaient sur des volumes de données d'activité. De l'autre, les collecteurs de journaux étaient conçus pour l'analyse hors ligne et ne servaient pas des consommateurs quasi temps réel.

Les auteurs décrivent les choix de conception qui permettent de tenir les deux à la fois : un log persistant et séquentiel plutôt qu'un index de messages, un consommateur qui tire les données au lieu de les recevoir, un décalage tenu par le consommateur et non par le courtier, l'envoi par lots, et un accès disque séquentiel qui s'appuie sur le cache du système d'exploitation au lieu de le contourner.

Un de ces choix est contre-intuitif et porte tout le reste : le courtier ne garde aucun état de livraison par message. C'est précisément ce qui lui permet d'en accepter beaucoup.

En septembre 2015, LinkedIn a publié un billet d'ingénierie, « How We're Improving and Advancing Kafka at LinkedIn », annonçant plus de mille milliards de messages publiés par jour sur ses clusters, avec des pointes à 4,5 millions de messages par seconde. Ce que montre le chiffre compte plus que sa taille : les décisions de conception de 2011 sont exactement celles qui rendent ce volume atteignable quatre ans plus tard.

Le cas est rare : on connaît publiquement la motivation d'origine, les arbitrages de conception et le comportement à très grande échelle du même système. Et la trajectoire dit une chose simple : ôter de l'intelligence au courtier a été la condition de sa montée en charge.

Points clés

• Une file achète du découplage, pas de la vitesse : le producteur cesse de dépendre du rythme et des pannes du consommateur, au prix d'une latence et d'un composant de plus.
• La file classique efface à la lecture, le log distribué conserve et se relit. La question qui tranche est celle de la relecture, pas celle du débit.
• L'ordre n'existe qu'à l'intérieur d'une partition : la clé de partition décide à la fois de l'ordre garanti et de l'équilibre de la charge.
• « Au moins une fois » est le régime réel de presque tous les systèmes. Il impose un consommateur idempotent, et « exactement une fois » ne franchit pas la frontière de Kafka.
• Le décalage du consommateur et sa dérivée suffisent à savoir si le système tient. Le rattrapage se calcule, la rétention fixe le compte à rebours.

Dans la série

Palier 2 : Faire circuler. Domaine : Flux & fiabilité.

• Précédent : Cache distribué et CDN : rapprocher la donnée du lecteur
• Suivant : Limitation de débit et idempotence : encaisser les rejeux
• Vue d'ensemble : System design : par où commencer

Pour aller plus loin

• Limitation de débit et idempotence : encaisser les rejeux, pour rendre un consommateur capable d'encaisser les doublons du « au moins une fois ».
• Transactions distribuées : 2PC, 3PC, SAGA et verrous, quand plusieurs services doivent aboutir ou échouer ensemble.
• Messagerie inter-services et supervision centralisée, pour comparer la file aux trois autres modes de communication sur cinq critères.
• Microservices, Event Sourcing et CQRS, pour ce qu'on construit une fois qu'un log rejouable existe.
• How We're Improving and Advancing Kafka at LinkedIn, le billet d'ingénierie de septembre 2015 d'où viennent les chiffres du scénario B.
• Documentation Apache Kafka, référence pour la réplication, les transactions et les réglages cités ici.