Comprendre Kafka : Haute Disponibilité

Comprendre Kafka : Haute Disponibilité

Table des matières

    1. L'origine de la haute disponibilité

  • 1.1. Pourquoi la réplication est nécessaire

  • 1.2. Élection du leader

    1. Analyse de la conception HA de Kafka

  • 2.1. Distribution uniforme des réplicas dans le cluster

  • 2.2. Stratégie de réplication des données

    1. Structure ZooKeeper liée à la HA

  • 3.1. Admin

  • 3.2. Broker

  • 3.3. Controller

    1. Publication de messages par le producteur

  • 4.1. Méthode d'écriture

  • 4.2. Routage des messages

  • 4.3. Processus d'écriture

    1. Stockage des messages par le broker

  • 5.1. Méthode de stockage

  • 5.2. Stratégie de rétention

    1. Création et suppression de topics

  • 6.1. Création d'un topic

  • 6.2. Suppression d'un topic

    1. Failover du broker
    1. Failover du controller
  1. L'origine de la haute disponibilité

1.1. Pourquoi la réplication est nécessaire

Avant la version 0.8 de Kafka, le système ne disposait pas de mécanisme de réplication. Lorsqu'un broker tombait en panne, toutes les partitions sur ce broker devenaient inaccessibles, ce qui contredisait les objectifs de persistance des données et de garantie de livraison de Kafka. De plus, les producteurs ne pouvaient plus stocker de données sur ces partitions.

Si le producteur utilisait un mode synchrone, il tenterait de renvoyer le message jusqu'à un nombre maximal de tentatives (configuré par message.send.max.retries, avec une valeur par défaut de 3) avant de lever une exception. L'utilisateur pouvait alors choisir d'arrêter l'envoi de données ou de continuer. La première option bloquerait les données, tandis que la seconde entraînerait une perte de données qui auraient dû être envoyées à ce broker.

Dans le mode asynchrone, le producteur tenterait de renvoyer le message jusqu'à message.send.max.retries tentatives, enregistrerait l'exception et continuerait d'envoyer les données suivantes. Cela entraînerait une perte de données, découverte uniquement via les journaux. Par ailleurs, le producteur Kafka ne fournit pas d'interface callback pour le mode asynchrone.

Il est donc évident que sans réplication, la panne d'une machine ou d'un broker réduirait la disponibilité globale du système. Avec l'augmentation de la taille du cluster, la probabilité de telles anomalies augmente considérablement, rendant le mécanisme de réplication essentiel pour les systèmes de production.

1.2. Élection du leader

Avec l'introduction de la réplication, une partition peut avoir plusieurs réplicas. Il est alors nécessaire d'en élire un leader, auquel les producteurs et consommateurs interagissent directement. Les autres réplicas agissent comme des followers, répliquant les données depuis le leader.

Pour garantir la cohérence des données entre les multiples réplicas d'une même partition (afin qu'en cas de panne, les autres réplicas puissent continuer à servir sans duplication ni perte de données), un mécanisme d'élection de leader est nécessaire. Sans leader, si tous les réplicas pouvaient lire/écrire simultanément, il faudrait assurer une synchronisation de données entre tous les pairs (N×N canaux), ce qui compliquerait considérablement la garantie de cohérence et d'ordre des données, augmentant ainsi la complexité de mise en œuvre et le risque d'anomalies.

L'introduction d'un leader simplifie le système : seul le leader gère les lectures/écritures, tandis que les followers ne récupèrent les données que depuis le leader (N canaux), rendant le système plus simple et efficace.

  1. Analyse de la conception HA de Kafka

2.1. Distribution uniforme des réplicas dans le cluster

Pour mieux équilibrer la charge, Kafka tente de répartir uniformément toutes les partitions sur l'ensemble du cluster. Un déploiement typique implique un nombre de partitions supérieur au nombre de brokers. Pour améliorer la tolérance aux pannes de Kafka, il est également nécessaire de disperser les réplicas d'une partition sur différentes machines.

En réalité, si tous les réplicas d'une partition se trouvaient sur le même broker, une panne de ce broker rendrait tous les réplicas de la partition inutilisables, annulant ainsi les bénéfices de la haute disponibilité. De plus, si un broker tombe en panne, il faut garantir que sa charge soit uniformément répartie sur les autres brokers survivants.

L'algorithme de distribution des réplicas de Kafka est le suivant :

  1. Trier tous les brokers (supposons n brokers) et les partitions à assigner
  2. Assigner la ième partition au broker (i mod n)
  3. Assigner le jème réplica de la ième partition au broker ((i + j) mod n)

2.2. Stratégie de réplication des données

La haute fiabilité de Kafka repose sur sa stratégie robuste de réplication.

2.2.1. Stratégie de synchronisation de la transmission des messages

Quand un producteur publie un message dans une partition, il trouve d'abord via ZooKeeper le leader de cette partition. Quelle que soit la valeur de Replication Factor du topic, le producteur n'envoie le message qu'au leader de la partition. Le leader écrit le message dans son propre journal local. Chaque follower récupère les données depuis le leader. De cette manière, l'ordre des données stockées par les followers reste cohérent avec celui du leader.

Après avoir reçu le message et l'avoir écrit dans son journal, le leader envoie un ACK. Une fois que le leader a reçu des ACKs de tous les réplicas de l'ISR (In-Sync Replicas), le message est considéré comme commité. Le leader augmente alors le HW (High Watermark) et envoie un ACK au producteur.

Pour améliorer les performances, chaque follower envoie un ACK au leader immédiatement après avoir reçu les données, plutôt qu'après leur écriture sur le disque. Par conséquent, pour les messages déjà commités, Kafka ne garantit que leur présence dans la mémoire de multiples réplicas, pas leur persistance sur disque, ce qui ne garantit pas qu'ils puissent toujours être consommés après une anomalie.

Les consommateurs lisent également les messages depuis le leader, et seuls les messages commités leur sont exposés.

2.2.2. Nombre de copies nécessaires avant ACK

Pour Kafka, un broker est considéré comme "actif" si deux conditions sont remplies :

  • Il doit maintenir une session avec ZooKeeper (via le mécanisme de Heartbeat).
  • Les followers doivent pouvoir répliquer les messages du leader rapidement, sans être "trop en retard".

Le leader suit la liste des réplicas qui lui restent synchrones, appelée ISR (In-Sync Replicas). Si un follower tombe en panne ou est trop en retard, le leader le retire de l'ISR. L'expression "trop en retard" fait référence au fait que le nombre de messages que le follower a répliqués en retard sur le leader dépasse une valeur prédéfinie (configurable via replica.lag.max.messages dans server.properties, valeur par défaut 4000) ou que le follower n'a pas envoyé de requête de fetch pendant un certain temps (configurable via replica.lag.time.max.ms, valeur par défaut 10000).

Le mécanisme de réplication de Kafka n'est ni une réplication purement synchrone, ni purement asynchrone. Une réplication synchrone complète exigerait que tous les followers fonctionnels aient répliqué le message avant qu'il soit considéré comme commité, ce qui affecterait considérablement le débit (un débit élevé étant une caractéristique importante de Kafka). Dans une réplication asynchrone, les followers répliquent les données de manière asynchrone depuis le leader, et le message est considéré comme commité dès qu'il est écrit dans le journal du leader. Dans ce cas, si tous les followers sont en retard par rapport au leader et que ce dernier tombe en panne, des données seront perdues. L'utilisation de l'ISR par Kafka équilibre bien la garantie de non-perte de données et le débit.

Les followers peuvent répliquer les données par lots depuis le leader, ce qui améliore considérablement les performances de réplication (écriture disque par lots) et réduit l'écart entre followers et leader.

Il est à noter que Kafka ne résout que les problèmes de panne/réinitialisation, pas les problèmes "Byzantins". Un message n'est considéré comme commité que s'il a été répliqué depuis le leader par tous les followers de l'ISR. Cela évite que des données soient écrites dans le leader sans avoir été répliquées par aucun follower avant une panne, entraînant une perte de données (non consommables par les consommateurs). Pour le producteur, il peut choisir d'attendre ou non que le message soit commité, via le paramètre request.required.acks. Ce mécanisme garantit qu'aussi longtemps qu'il y a au moins un follower dans l'ISR, un message commité ne sera pas perdu.

2.2.3. Algorithme d'élection du leader

L'élection du leader est essentiellement un verrou distribué. Deux méthodes principales permettent d'implémenter un verrou distribué basé sur ZooKeeper :

  • Unicité du nom de nœud : plusieurs clients créent un nœud, seul le client ayant réussi à créer le nœud obtient le verrou.
  • Nœud temporaire séquentiel : tous les clients créent leur propre nœud temporaire séquentiel dans un répertoire donné, seul celui avec le plus petit numéro d'ordre obtient le verrou.

Une méthode très courante d'élection de leader est le "vote majoritaire" (Majority Vote), mais Kafka ne l'utilise pas. Dans ce mode, avec 2f+1 réplicas (incluant le leader et les followers), il faut garantir que f+1 réplicas aient répliqué le message avant qu'il soit commité. Pour garantir une élection correcte du nouveau leader, le nombre de réplicas en panne ne peut pas dépasser f. Parmi n'importe quel groupe de f+1 réplicas restants, au moins un contient tous les messages les plus récents. Cette méthode présente un avantage majeur : la latence du système dépend seulement des brokers les plus rapides, pas du plus lent. Cependant, pour garantir une tolérance aux pannes élevée, de nombreux réplicas sont nécessaires, ce qui peut entraîner une baisse de performance avec de grands volumes de données.

Kafka maintient dynamiquement dans ZooKeeper un ISR (in-sync replicas), où tous les membres sont synchronisés avec le leader. Seuls les membres de l'ISR peuvent être élus leader. Dans ce mode, pour f+1 réplicas, une partition peut tolérer f pannes de réplicas sans perdre les messages déjà commités. Dans la plupart des scénarios d'utilisation, ce mode est très avantageux.

2.2.4. Gestion du cas où tous les réplicas sont inactifs

Lorsqu'au moins un follower est dans l'ISR, Kafka peut garantir que les données commitées ne seront pas perdues. Cependant, si tous les réplicas d'une partition sont inactifs, il est impossible de garantir l'absence de perte de données. Deux solutions sont possibles :

  1. Attendre qu'un quelconque réplica de l'ISR "revienne en vie" et l'élire leader.
  2. Élire comme leader le premier réplica "revenu en vie" (pas nécessairement dans l'ISR).

Cela nécessite un compromis simple entre disponibilité et cohérence. Attendre qu'un réplica de l'ISR "revienne en vie" peut entraîner une indisponibilité plus longue. Si tous les réplicas de l'ISR ne peuvent pas "revenir en vie" ou si leurs données sont perdues, la partition restera permanente inaccessible. Si un réplica non-ISR est élu leader, il ne garantit pas qu'il contient tous les messages commités, mais il deviendra leader et source de données pour les consommateurs. Kafka 0.8.* utilise la deuxième approche. Selon la documentation de Kafka, les versions futures permettront aux utilisateurs de choisir entre ces deux approches via configuration, en fonction des besoins spécifiques en disponibilité ou en cohérence forte.

2.2.5. Élection du leader

La solution la plus simple et intuitive serait que tous les followers surveillent un nœud dans ZooKeeper. En cas de panne du leader, son nœud éphémère serait supprimé automatiquement, et tous les followers tenteraient de créer ce nœud. Celui qui réussit (ZooKeeper garantit qu'un seul peut réussir) deviendrait le nouveau leader, les autres réplicas devenant followers.

Cependant, cette méthode présente trois problèmes :

  • Split-brain : causé par les caractéristiques de ZooKeeper, bien que ZooKeeper garantisse que tous les événements de surveillance sont déclenchés dans l'ordre, il ne garantit pas que tous les réplicas "voient" le même état au même moment, ce qui peut entraîner des réponses incohérentes.
  • Effet de troupeau : si le broker en panne héberge de nombreuses partitions, de nombreux événements de surveillance seraient déclenchés, entraînant de nombreux ajustements dans le cluster.
  • Surchargede ZooKeeper : chaque réplica doit enregistrer un surveillant dans ZooKeeper, ce qui peut surcharger ZooKeeper lorsque le cluster compte plusieurs milliers de partitions.

L'approche d'élection du leader de Kafka 0.8.* résout ces problèmes en désignant un controller parmi tous les brokers. Toutes les élections de leader de partition sont décidées par ce controller. Le controller notifie directement les brokers concernés des changements de leader via RPC (plus efficace que l'utilisation des files d'attente de ZooKeeper). Le controller est également responsable de l'ajout/suppression de topics et de la réattribution des réplicas.

  1. Structure ZooKeeper liée à la HA

3.1. Admin

Ce répertoire n'existe que lors d'opérations spécifiques et est supprimé à la fin de l'opération.

  • /admin/reassign_partitions : utilisé pour assigner certaines partitions à différents ensembles de brokers. Pour chaque partition à réassigner, Kafka stocke tous ses réplicas et les ID de brokers correspondants dans ce nœud. Ce nœud est créé par le processus d'administration et est automatiquement supprimé une fois la réattribution réussie.

3.2. Broker

  • /brokers/ids/[brokerId] : stocke les informations des brokers "vivants".
  • /brokers/topics/[topic] : stocke les ID des brokers pour tous les réplicas de toutes les partitions du topic. Le premier réplica est le "preferred replica". Pour une partition donnée, un broker ne peut héberger qu'un seul réplica, donc l'ID du broker peut servir d'ID de réplica.

3.3. Controller

  • /controller → int (ID du broker du controller) : stocke les informations du controller actuel.
  • /controller_epoch → int (epoch) : stocke directement l'epoch du controller sous forme d'entier, contrairement à d'autres nœuds qui utilisent une chaîne JSON.
  1. Publication de messages par le producteur

4.1. Méthode d'écriture

Le producteur utilise un mode push pour publier des messages vers les brokers. Chaque message est ajouté à une partition, ce qui correspond à une écriture séquentielle sur disque (plus efficace qu'une écriture aléatoire en mémoire, garantissant le débit élevé de Kafka).

4.2. Routage des messages

Lorsqu'un producteur envoie un message à un broker, il choisit une partition en fonction d'un algorithme de partitionnement. Le mécanisme de routage est le suivant :


1. Si la partition est spécifiée, elle est utilisée directement.
2. Si la partition n'est pas spécifiée mais une clé est fournie, une partition est sélectionnée en appliquant un hash à la valeur de la clé.
3. Si ni la partition ni la clé ne sont spécifiées, une partition est sélectionnée par tourniquet (round-robin).

4.3. Processus d'écriture

Le processus d'écriture des messages par le producteur suit les étapes suivantes :

  1. Le producteur trouve d'abord le leader de la partition via le nœud "/brokers/.../state" de ZooKeeper.

  2. Le producteur envoie le message à ce leader.

  3. Le leader écrit le message dans son journal local.

  4. Les followers récupèrent les messages depuis le leader, les écrivent dans leur journal local, et envoient un ACK au leader.

  5. Une fois que le leader a reçu des ACKs de tous les réplicas de l'ISR, il augmente le HW (High Watermark, le dernier offset commité) et envoie un ACK au producteur.

  6. Stockage des messages par le broker

5.1. Méthode de stockage

Physiquement, un topic est divisé en une ou plusieurs partitions (correspondant à la configuration num.partitions=3 dans server.properties). Chaque partition correspond physiquement à un dossier stockant tous les messages et fichiers d'index de cette partition.

5.2. Stratégie de rétention

Au-delà de la consommation, Kafka conserve tous les messages. Deux stratégies permettent de supprimer les anciennes données :


1. Basée sur le temps : log.retention.hours=168
2. Basée sur la taille : log.retention.bytes=1073741824
  1. Création et suppression de topics

6.1. Création d'un topic

Le processus de création d'un topic suit les étapes suivantes :

  1. Le controller surveille le nœud "/brokers/topics" de ZooKeeper. Lorsqu'un topic est créé, le controller est notifié et obtient l'allocation des partitions/réplicas du topic.
  2. Le controller lit la liste de tous les brokers disponibles depuis "/brokers/ids". Pour chaque partition dans l'ensemble :
    • Il sélectionne un broker disponible parmi tous les réplicas de la partition (appelé AR) comme nouveau leader, et définit l'AR comme nouveau ISR.
    • Il écrit le nouveau leader et l'ISR dans le nœud "/brokers/topics/[topic]/partitions/[partition]/state".
  3. Le controller envoie une requête LeaderAndISRRequest aux brokers concernés via RPC.

6.2. Suppression d'un topic

Le processus de suppression d'un topic suit les étapes suivantes :

  1. Le controller surveille le nœud "/brokers/topics" de ZooKeeper. Lorsqu'un topic est supprimé, le controller est notifié et obtient l'allocation des partitions/réplicas du topic.

  2. Si delete.topic.enable=false, le processus se termine. Sinon, le controller reçoit une notification via le surveillant enregistré sur "/admin/delete_topics", et envoie une requête StopReplicaRequest aux brokers correspondants via un callback.

  3. Failover du broker

Le processus de failover d'un broker Kafka suit les étapes suivantes :

  1. Le controller surveille le nœud "/brokers/ids/[brokerId]" de ZooKeeper. Lorsqu'un broker tombe en panne, ZooKeeper déclenche l'événement.

  2. Le controller lit la liste des brokers disponibles depuis le nœud "/brokers/ids".

  3. Le controller détermine l'ensemble des partitions (set_p) qui étaient sur le broker en panne.

  4. Pour chaque partition dans set_p :

    • Il lit l'ISR depuis le nœud "/brokers/topics/[topic]/partitions/[partition]/state".
    • Il décide du nouveau leader.
    • Il écrit les informations du nouveau leader, de l'ISR, du controller_epoch et du leader_epoch dans le nœud state.

  5. Il envoie une commande leaderAndISRRequest aux brokers concernés via RPC.

  6. Failover du controller

Lorsqu'un controller tombe en panne, un failover du controller est déclenché. Chaque broker surveille le nœud "/controller" de ZooKeeper. Lorsque le controller tombe en panne, le nœud temporaire disparaît, et tous les brokers vivants reçoivent une notification. Chaque broker tente de créer un nouveau chemin de controller, et seul celui qui réussit devient le nouveau controller.

Lorsqu'un nouveau controller est élu, la méthode KafkaController.onControllerFailover est exécutée, effectuant les opérations suivantes :

  1. Lire et incrémenter l'epoch du controller.
  2. Enregistrer un surveillant sur le chemin reassignedPartitions (/admin/reassign_partitions).
  3. Enregistrer un surveillant sur le chemin preferredReplicaElection (/admin/preferred_replica_election).
  4. Enregistrer un surveillant sur le chemin /brokers/topics via partitionStateMachine.
  5. Si delete.topic.enable=true (valeur par défaut false), partitionStateMachine enregistre un surveillant sur le chemin Delete Topic (/admin/delete_topics).
  6. Enregistrer un surveillant sur le chemin /brokers/ids via replicaStateMachine.
  7. Initialiser l'objet ControllerContext avec tous les topics actuels, la liste des brokers "vivants", et les leaders et ISR de toutes les partitions.
  8. Démarrer replicaStateMachine et partitionStateMachine.
  9. Définir l'état du broker sur RunningAsController.
  10. Envoyer les informations de leadership de chaque partition à tous les brokers "vivants".
  11. Si auto.leader.rebalance.enable=true (valeur par défaut true), démarrer le thread de rééquilibrage des partitions.
  12. Si delete.topic.enable=true et que Delete Topic Path (/admin/delete_topics) contient des valeurs, supprimer les topics correspondants.

Étiquettes: Kafka haute disponibilité réplication ZooKeeper Producteur

Publié le 12 juillet à 03h54

Étiquettes Populaires

©2026 L'Atelier Monstre