Comment gérer Kafka par programmation

Introduction

Apache Kafka est une plateforme de traitement d'événements et de flux distribuée open-source écrite en Java. Elle est conçue pour traiter des flux de données en temps réel exigeants et est intrinsèquement évolutive, avec un débit et une disponibilité élevés. Le projet fournit des scripts shell pour produire et consommer des messages à destination et en provenance d'un cluster Kafka, ainsi que pour des tâches administratives, telles que la gestion des topics et des partitions. Bien que ceux-ci soient utiles pour explorer et expérimenter, les applications du monde réel accèdent à Kafka de manière programmatique. À cette fin, Kafka propose de nombreuses bibliothèques clientes pour les langages de programmation et environnements largement utilisés.

Dans ce tutoriel, vous apprendrez à gérer les ressources dans un cluster Kafka en utilisant l'API KafkaAdminClient. Vous apprendrez également à récupérer des informations sur le cluster de manière programmatique et à créer, lister et supprimer des topics. Vous découvrirez également l'utilitaire en ligne de commande kcat, qui vous permet d'accéder à votre cluster Kafka sans dépendre de Java.

Enfin, vous apprendrez à configurer Kafka Cruise Control, un démon permettant d’optimiser automatiquement les processus internes d’un cluster Kafka, ce qui permet d’obtenir une plus grande efficacité et fiabilité.

Prérequis

Pour compléter ce tutoriel, vous aurez besoin de :

Une machine avec au moins 4 Go de RAM et 2 processeurs. Dans le cas d'un serveur Ubuntu, suivez la configuration initiale du serveur pour les instructions de configuration.
Installez le kit de développement Java (JDK) 11 sur votre droplet ou votre machine locale. Pour des instructions sur l'installation de Java sur Ubuntu, consultez le tutoriel Comment installer Java avec Apt sur Ubuntu.
Un cluster Apache Kafka avec trois brokers. Vous pouvez suivre le tutoriel Comment configurer un cluster Kafka multi-nœuds utilisant KRaft pour les instructions de configuration.
Un projet Java avec un producteur Kafka configuré selon le tutoriel Comment configurer un producteur Kafka pour source de données via CLI.
Familiarité avec la disposition standard des répertoires des projets Java. Pour plus d'informations, voir le sujet Introduction à la disposition standard des répertoires dans la documentation officielle de Maven.
Python 3 installé sur votre Droplet ou machine locale, avec un nouvel environnement virtuel configuré. Dans le cas d'un serveur Ubuntu, consultez le tutoriel Comment Installer Python 3 et Configurer un Environnement de Programmation sur un Serveur Ubuntu. Vous devez uniquement compléter l'Étape 1 et l'Étape 2. Dans ce tutoriel, l'environnement virtuel sera considéré comme ~/venv.

Étape 1 - Utilisation de Kafka AdminClient

Dans le cadre des prérequis, vous avez créé un projet Java avec les dépendances nécessaires pour accéder à Kafka de manière programmatique et produire des messages dans le sujet java_demo. Dans cette étape, vous allez créer une classe qui utilise la classe AdminClient de Kafka pour gérer le cluster de manière programmatique.

Accédez au répertoire où le projet dokafka est stocké. La structure du projet spécifie que le code source est stocké sous src/main/java/com/dokafka.

Vous allez stocker la classe dans un fichier nommé AdminClientDemo.java. Créez-le et ouvrez-le pour l’édition en exécutant :

nano src/main/java/com/dokafka/AdminClientDemo.java

Ajoutez les lignes suivantes :

package com.dokafka;

import org.apache.kafka.clients.admin.*;
import org.apache.kafka.common.Node;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.*;

public class AdminClientDemo {
  private static final Logger log = LoggerFactory.getLogger(AdminClientDemo.class);

  public static void main(String[] args) {
    String bootstrapServers = "kafka1.your_domain:9092";

    Properties properties = new Properties();
    properties.put("bootstrap.servers", bootstrapServers);
    final KafkaAdminClient client = AdminClient.create(properties);

    try {
      Collection<Node> nodes = client.describeCluster().nodes().get();
      if (nodes == null)
        log.info("There seem to be no nodes in the cluster!");
      else
        log.info(String.format("Count of nodes: %s\n", nodes.size()));
    } catch (Exception e) {
      log.error("An error occurred", e);
    }
  }
}

Tout d'abord, vous définissez la classe AdminClientDemo et importez les classes que vous utiliserez. Vous instanciez également un Logger en tant que membre de la classe. De manière similaire à ProducerDemo, dans la méthode main() vous déclarez d'abord l'adresse du cluster Kafka (bootstrapServers). N'oubliez pas de remplacer kafka1.your_domain par votre nom de domaine réel.

Ensuite, vous instanciez un objet Properties, qui contient des paires de clés et de valeurs représentant la configuration pour faire fonctionner votre consommateur Kafka. Vous définissez la propriété bootstrap.servers à l'adresse du cluster Kafka.

Après cela, vous déclarez et instanciez un AdminClient :

final KafkaAdminClient client = AdminClient.create(properties);

Le client peut effectuer des actions administratives dans le cluster, telles que lister, créer et supprimer des topics, des partitions et des offsets. Ici, vous l'utilisez pour récupérer le nombre de nœuds présents dans le cluster :

...
    try {
      Collection<Node> nodes = client.describeCluster().nodes().get();
      if (nodes == null)
        log.info("There seem to be no nodes in the cluster!");
      else
        log.info(String.format("Count of nodes: %s\n", nodes.size()));
    } catch (Exception e) {
      log.error("An error occurred", e);
    }
...

Si aucun objet Node n'est renvoyé, vous enregistrez un message approprié. Sinon, vous affichez le nombre de nœuds Kafka reçus. Lorsque vous avez terminé, enregistrez et fermez le fichier.

Ensuite, vous allez créer un script pour compiler et exécuter AdminClientDemo. Vous le stockerez dans un fichier appelé run-adminclient.sh. Créez-le et ouvrez-le pour l'édition :

nano run-adminclient.sh

Ajoutez les lignes suivantes :

#!/bin/bash
mvn clean
mvn package
java -cp target/dokafka-1.0-SNAPSHOT.jar:target/lib/* com.dokafka.AdminClientDemo

Enregistrez et fermez le fichier, puis marquez-le comme exécutable :

chmod +x run-adminclient.sh

Enfin, essayez de l'exécuter :

./run-adminclient.sh

La sortie sera longue, et sa fin devrait ressembler à ceci :

...
[main] INFO com.dokafka.AdminClientDemo - Count of nodes: 3

AdminClientDemo s'est connecté avec succès à votre installation Kafka et a récupéré le nombre de nœuds.

Créer et lister des sujets

KafkaAdminClient fournit les méthodes createTopics() et listTopics(), que vous allez maintenant utiliser pour créer un sujet et ensuite lister tous ceux qui sont présents dans le cluster.

Tout d'abord, ouvrez AdminClientDemo pour l'édition en exécutant :

nano src/main/java/com/dokafka/AdminClientDemo.java

Modifiez le code pour qu'il ressemble à ceci :

...
    try {
      NewTopic newTopic = new NewTopic("newTopic", 1, (short) 1);
      CreateTopicsResult result = client.createTopics(
        Collections.singleton(newTopic)
      );
      result.all().get();

      ListTopicsOptions options = new ListTopicsOptions();
      options.listInternal(true);
      Collection<TopicListing> topics = client.listTopics(options).listings().get();
      for (TopicListing topic: topics) {
        log.info("Topic: " + topic.name());
      }
    } catch (Exception e) {
      log.error("An error occurred", e);
    }
...

Vous créez d'abord une instance de NewTopic, qui décrit un sujet devant être créé dans le cluster. Vous passez son nom ("newTopic") et le nombre de partitions et de répliques, que vous devez convertir en short. La méthode createTopics() accepte une Collection d'instances de NewTopic, donc vous construisez une liste avec un seul élément contenant uniquement newTopic.

Puisque cette méthode est asynchrone, le flux d'exécution retournera à main() immédiatement après son appel tandis qu'elle continue de s'exécuter en arrière-plan. Comme vous allez lister les sujets ensuite, vous devez attendre qu'elle se termine avec result.all().get(). result contient une liste des sujets créés, et obtenir all() d'entre eux nécessite que l'opération se termine avant de continuer.

Ensuite, vous créez une instance de ListTopicsOptions, qui contient la configuration pour le processus de récupération des sujets. Vous l'appelez via la méthode listInteral() et passez true pour permettre la récupération des sujets que Kafka considère uniquement pour un usage interne. Ensuite, vous le passez à client.listTopics() et récupérez les listings(), à travers lesquels vous itérez dans une boucle et consignez les noms des sujets.

Enregistrez et fermez le fichier lorsque vous avez terminé, puis exécutez le script :

./run-adminclient.sh

Tous les sujets du groupe seront listés à la fin de la sortie :

...
[main] INFO com.dokafka.AdminClientDemo - Topic: newTopic
[main] INFO com.dokafka.AdminClientDemo - Topic: java_demo

Suppression des sujets

Pour supprimer un sujet (ou plusieurs), vous pouvez utiliser la méthode deleteTopics() de KafkaAdminClient. Vous allez maintenant l'utiliser pour supprimer le newTopic que vous venez de créer. Ouvrez AdminClientDemo pour l'édition :

nano src/main/java/com/dokafka/AdminClientDemo.java

Remplacez le code de création de sujet par les lignes surlignées :

...
      DeleteTopicsResult deleted = client.deleteTopics(Collections.singleton("newTopic"));
      deleted.all().get();
      log.info("Topic newTopic deleted!");

      ListTopicsOptions options = new ListTopicsOptions();
      options.listInternal(true);
      Collection<TopicListing> topics = client.listTopics(options).listings().get();
      for (TopicListing topic: topics) {
        log.info("Topic: " + topic.name());
      }
...

Ici, vous passez une Collection de chaînes représentant les noms des sujets à client.deleteTopics(). Ensuite, vous attendez qu'il se termine en obtenant deleted.all(), ce qui ne retourne que lorsque tous les sujets ont été traités. Enregistrez et fermez le fichier lorsque vous avez terminé, puis exécutez le script run-adminclient.sh :

./run-adminclient.sh

Vous verrez que le nouveau sujet n'est pas présent dans la liste des sujets :

[main] INFO com.dokafka.AdminClientDemo - Topic newTopic deleted!
[main] INFO com.dokafka.AdminClientDemo - Topic: java_demo

À cette étape, vous avez utilisé le KafkaAdminClient pour accéder à votre cluster et récupérer des informations à son sujet de manière programmée. Vous avez également vu comment lister, créer et supprimer des sujets dans le cluster. Vous allez maintenant apprendre à utiliser l'outil CLI kcat pour accéder à votre cluster.

Étape 2 - Utiliser kcat pour gérer le cluster

Dans cette étape, vous apprendrez comment télécharger et installer kcat, un outil en ligne de commande pour accéder à et configurer des clusters Kafka sans dépendance à Java.

Tout d'abord, vous devrez installer le paquet approprié pour votre système d'exploitation. Si vous êtes sur MacOS, vous pouvez télécharger kcat avec Brew :

brew install kcat

Sur les systèmes Debian et Ubuntu, vous pouvez l'installer comme d'habitude via apt :

sudo apt install kafkacat

kafkacat est un ancien nom de kcat, mais il est conservé dans le gestionnaire de paquets pour des raisons de compatibilité. Pour les instructions d'installation sur d'autres distributions, consultez la documentation officielle.

Produire et Consommer des Messages

Une des opérations de base que les scripts fournis avec Kafka permettent est de diffuser des données à partir d'un sujet. Avec kcat, vous pouvez diffuser depuis plusieurs sujets à la fois :

kcat -b your_broker_address:9092 first_topic second_topic ...

L'exécution de la commande suivante diffusera les messages du sujet java_demo dans la console :

kcat -b kafka1.your_domain:9092 java_demo

Le script Kafka-console-consumer.sh vous permet également de lire tous les messages sur le sujet, en commençant depuis le début. Passez -t pour obtenir la même chose avec kcat :

kcat -b kafka1.your_domain:9092 -t java_demo

La sortie contiendra les messages que vous avez produits dans le cadre des prérequis :

% Auto-selecting Consumer mode (use -P or -C to override)
% Reached end of topic java_demo [1] at offset 0
% Reached end of topic java_demo [2] at offset 0
Hello World!
% Reached end of topic java_demo [0] at offset 0
% Reached end of topic java_demo [3] at offset 0
% Reached end of topic java_demo [4] at offset 1
% Reached end of topic java_demo [5] at offset 0

Vous pouvez également afficher les messages consommés au format JSON en passant -J :

kcat -b kafka1.your_domain:9092 -t java_demo -J

La sortie ressemblera à ceci :

% Auto-selecting Consumer mode (use -P or -C to override)
% Reached end of topic java_demo [2] at offset 0
% Reached end of topic java_demo [0] at offset 0
% Reached end of topic java_demo [1] at offset 0
{"topic":"java_demo","partition":4,"offset":0,"tstype":"create","ts":1714922509999,"broker":1,"key":null,"payload":"Hello World!"}
% Reached end of topic java_demo [3] at offset 0
% Reached end of topic java_demo [5] at offset 0
% Reached end of topic java_demo [4] at offset 1

Pour produire des messages vers un sujet, passez -P pour basculer en mode producteur :

kcat -b kafka1.your_domain:9092 -t java_demo -P

De la même manière que kafka-console-producer.sh, kcat acceptera des messages séparés par ENTRÉE. Pour quitter l'invite, vous pouvez appuyer sur CTRL + C et suivre avec ENTRÉE.

Vous pouvez utiliser la capacité de modélisation de kcat pour afficher plus d'informations sur un message en même temps. Pour passer un format de votre choix, utilisez -f comme montré dans la commande suivante :

kcat -b kafka1.your_domain:9092 -t java_demo -f 'Topic %t[%p], offset: %o, key: %k, payload: %S bytes: %s\n'

Ici, vous lui indiquez de lire tous les messages depuis le début du sujet java_demo et d'afficher le nom du sujet (%t), le numéro de partition (%p), le décalage (%o) et la clé du message (%k), ainsi que la longueur de la valeur (%S) et le message lui-même (%s).

La sortie ressemblera à ceci :

% Auto-selecting Consumer mode (use -P or -C to override)
% Reached end of topic java_demo [2] at offset 0
% Reached end of topic java_demo [1] at offset 0
Topic java_demo[4], offset: 0, key: , payload: 12 bytes: Hello World!
% Reached end of topic java_demo [0] at offset 0
% Reached end of topic java_demo [3] at offset 0
% Reached end of topic java_demo [4] at offset 1
% Reached end of topic java_demo [5] at offset 0

kcat a utilisé le modèle de chaîne et a produit des informations supplémentaires sur l'enregistrement du message.

Lister les métadonnées du cluster

Vous pouvez lister les métadonnées du cluster de manière organisée en passant -L :

kcat -b kafka1.your_domain:9092 -L

La sortie listera les courtiers dans le cluster, ainsi que tous les sujets et leurs partitions :

 3 brokers:
  broker 1 at kafka1.your_domain:9092
  broker 2 at kafka2.your_domain:9092 (controller)
  broker 3 at kafka3.your_domain:9092
 1 topics:
  topic "java_demo" with 6 partitions:
    partition 0, leader 3, replicas: 3,1 isrs: 3,1
    partition 1, leader 1, replicas: 1,2 isrs: 1,2
    partition 2, leader 2, replicas: 2,3 isrs: 2,3
    partition 3, leader 2, replicas: 2,1 isrs: 2,1
    partition 4, leader 1, replicas: 1,3 isrs: 1,3
    partition 5, leader 3, replicas: 3,2 isrs: 3,2

De manière similaire au script Kafka-topics.sh, kcat liste les leaders de partition, les réplicas et les ensembles de réplicas synchronisés.

Vous pouvez également passer -J pour obtenir la sortie au format JSON :

kcat -b kafka1.your_domain:9092 -L -J

La sortie ressemblera à ceci :

{
    "originating_broker": {
        "id": 2,
        "name": "kafka2.your_domain:9092/2"
    },
    "query": {
        "topic": "*"
    },
    "controllerid": 3,
    "brokers": [
        {
            "id": 1,
            "name": "kafka1.your_domain:9092"
        },
        {
            "id": 2,
            "name": "kafka2.your_domain:9092"
        },
        {
            "id": 3,
            "name": "kafka3.your_domain:9092"
        }
    ],
    "topics": [
        {
            "topic": "java_demo",
            "partitions": [
                {
                    "partition": 0,
                    "leader": 3,
                    "replicas": [
                        {
                            "id": 3
                        }
                    ],
                    "isrs": [
                        {
                            "id": 3
                        }
                    ]
                },
                ...
            ]
        }
    ]
}

À cette étape, vous avez installé kcat, un outil permettant d'accéder et de gérer facilement les clusters Kafka sans dépendre de Java. Vous avez vu comment récupérer des informations sur le cluster et ses topics, ainsi que comment produire et consommer des messages. Vous allez maintenant apprendre à automatiser les rééquilibrages appropriés dans le cluster avec Cruise Control.

Étape 3 - Automatiser les rééquilibrages avec Kafka Cruise Control

Cruise Control est un projet open-source développé par LinkedIn qui surveille en continu l'activité des brokers Kafka dans un cluster et rééquilibre les charges de travail pour optimiser l'utilisation des ressources et le débit. Dans cette étape, vous apprendrez comment l'activer pour votre cluster et suivre ses opérations.

Par défaut, Cruise Control dispose de cibles sensées pour garantir que le cluster se comporte de manière optimale, appelées objectifs. Parmi les principaux objectifs, on trouve la garantie que chaque sujet dispose d'un nombre approprié de répliques, le maintien de l'équilibre et du respect des limites spécifiées pour l'utilisation du CPU, du réseau et du disque, ainsi que la disponibilité d'un nombre suffisant de répliques des partitions de sujets. La mise en œuvre d'objectifs personnalisés est également prise en charge.

Vous devrez compiler Cruise Control à partir des sources. Tout d'abord, clonez le dépôt Git officiel en exécutant :

git clone https://github.com/linkedin/cruise-control.git

Naviguez vers celui-ci :

cd cruise-control

Ensuite, utilisez Gradle pour compiler le projet :

./gradlew jar

Le processus de construction prendra un certain temps. La fin de la sortie ressemblera à ceci :

...
BUILD SUCCESSFUL in 2m 41s
17 actionable tasks: 17 executed

Le régulateur de vitesse est maintenant compilé avec son rapporteur de métriques, qui se trouve sous cruise-control-metrics-reporter/build/libs/ dans un fichier JAR. Le rapporteur envoie des métriques concernant un courtier dans un sujet sur le cluster que Cruise Control peut surveiller.

Ensuite, exécutez la commande suivante pour copier toutes les dépendances dans le répertoire cible :

./gradlew jar copyDependantLibs

La sortie sera :

BUILD SUCCESSFUL in 15s
17 actionable tasks: 1 executed, 16 up-to-date

Ensuite, vous devrez configurer chaque broker Kafka du cluster pour utiliser le rapporteur de métriques spécial. Copiez-le dans le répertoire libs/ où Kafka est installé en exécutant :

cp cruise-control-metrics-reporter/build/libs/* /home/kafka/kafka/libs/

Ensuite, vous devrez modifier la configuration du courtier Kafka pour utiliser le nouveau rapporteur. Ouvrez le fichier server.properties du courtier pour l'éditer :

nano /home/kafka/kafka/config/kraft/server.properties

Ajoutez la ligne suivante à la fin du fichier :

...
metric.reporters=com.linkedin.kafka.cruisecontrol.metricsreporter.CruiseControlMetricsReporter

Enregistrez et fermez le fichier une fois que vous avez terminé. Redémarrez le courtier en exécutant :

sudo systemctl restart kafka

Tous les brokers du cluster doivent avoir le rapporteur de métriques Cruise Control activé, donc répétez les étapes ci-dessus sur les autres nœuds pour obtenir le JAR du rapporteur de métriques. Après quelques minutes, vous pouvez lister les topics du cluster en utilisant kcat pour vérifier que le rapporteur de métriques a commencé à fonctionner :

...
  topic "__CruiseControlMetrics" with 6 partitions:
    partition 0, leader 3, replicas: 3,2 isrs: 3,2
    partition 1, leader 2, replicas: 2,3 isrs: 2,3
    partition 2, leader 3, replicas: 3,2 isrs: 3,2
    partition 3, leader 2, replicas: 2,3 isrs: 2,3
    partition 4, leader 2, replicas: 2,3 isrs: 2,3
    partition 5, leader 3, replicas: 3,2 isrs: 3,2

Afin d'optimiser correctement les courtiers, Cruise Control doit connaître les spécifications matérielles des courtiers individuels. Cette configuration est stockée dans un fichier appelé capacity.json, situé sous config. Ouvrez-le pour le modifier :

nano config/capacity.json

Dans son état par défaut, le fichier ressemble à ceci :

{
  "brokerCapacities":[
    {
      "brokerId": "-1",
      "capacity": {
        "DISK": "100000",
        "CPU": "100",
        "NW_IN": "10000",
        "NW_OUT": "10000"
      },
      "doc": "This is the default capacity. Capacity unit used for disk is in MB, cpu is in percentage, network throughput is in KB."
    },
    {
      "brokerId": "0",
      "capacity": {
        "DISK": "500000",
        "CPU": "100",
        "NW_IN": "50000",
        "NW_OUT": "50000"
      },
      "doc": "This overrides the capacity for broker 0."
    }
  ]
}

Le tableau brokerCapacities contient des éléments liés aux identifiants des courtiers. Chacun contient des champs pour l'espace disque, l'allocation CPU et le débit réseau vers et depuis le courtier.

Vous avez déployé trois brokers Kafka dans le cadre des prérequis. Modifiez le fichier en ajoutant les spécifications correctes pour chacun d'eux :

{
  "brokerCapacities":[
    {
      "brokerId": "-1",
      "capacity": {
        "DISK": "100000",
        "CPU": "100",
        "NW_IN": "10000",
        "NW_OUT": "10000"
      },
      "doc": "This is the default capacity. Capacity unit used for disk is in MB, cpu is in percentage, network throughput is in KB."
    },
    {
      "brokerId": "1",
      "capacity": {
        "DISK": "100000",
        "CPU": "100",
        "NW_IN": "10000",
        "NW_OUT": "10000"
      },
      "doc": ""
    },
    {
      "brokerId": "2",
      "capacity": {
        "DISK": "100000",
        "CPU": "100",
        "NW_IN": "10000",
        "NW_OUT": "10000"
      },
      "doc": ""
    },
    {
      "brokerId": "3",
      "capacity": {
        "DISK": "100000",
        "CPU": "100",
        "NW_IN": "10000",
        "NW_OUT": "10000"
      },
      "doc": ""
    }
  ]
}

Définissez les capacités des disques en fonction de la taille de vos Droplets, puis enregistrez et fermez le fichier.

Ensuite, vous allez configurer Cruise Control pour se connecter à votre cluster en mode KRaft. Vous allez modifier le fichier cruisecontrol.properties dans config. Ouvrez-le pour édition en exécutant ce qui suit :

nano config/cruisecontrol.properties

Trouvez les lignes suivantes :

...
# The Kafka cluster to control.
bootstrap.servers=localhost:9092
...

Le paramètre bootstrap.servers spécifie à quel courtier, et à quel endroit de son cluster, se connecter. Modifiez-le pour qu'il ressemble à ceci :

...
# The Kafka cluster to control.
bootstrap.servers=kafka1.your_domain:9092

# Switch to KRaft mode
kafka.broker.failure.detection.enable=true
...

Le paramètre kafka.broker.failure.detection.enable indique à Cruise Control d'ignorer les paramètres de ZooKeeper et d'utiliser le mode KRaft nativement. Remplacez kafka1.votre_domaine par le nom de domaine de votre broker Kafka.

Ensuite, trouvez les lignes suivantes :

...
# The configuration for the BrokerCapacityConfigFileResolver (supports JBOD, non-JBOD, and heterogeneous CPU core capacities)
#capacity.config.file=config/capacity.json
capacity.config.file=config/capacityJBOD.json
...

Rendre le premier paramètre de configuration actif :

...
# The configuration for the BrokerCapacityConfigFileResolver (supports JBOD, non-JBOD, and heterogeneous CPU core capacities)
capacity.config.file=config/capacity.json
#capacity.config.file=config/capacityJBOD.json
...

Ici, vous définissez le fichier capacity.json (se trouvant sous config/) comme la spécification des capacités du courtier.

Enregistrez et fermez le fichier lorsque vous avez terminé. Maintenant que le contrôle de vitesse est configuré, exécutez-le dans un terminal séparé en utilisant la commande suivante :

./kafka-cruise-control-start.sh config/cruisecontrol.properties

Il y aura beaucoup de flux de sortie en continu tandis que Cruise Control surveille et optimise continuellement votre cluster.

Utilisation de l'interface en ligne de commande Cruise Control

Cruise Control expose une API REST sur le port 9090 pour les tâches de configuration et d'administration. Cependant, le projet maintient également cccli, un outil Python qui enveloppe l'API dans une interface en ligne de commande intuitive. Dans cette section, vous allez l'installer et apprendre à récupérer des informations sur votre cluster.

Accédez à l'environnement virtuel Python 3 que vous avez créé dans le cadre des prérequis :

cd ~/venv

Activez-le en exécutant :

./bin/activate

Ensuite, utilisez pip pour installer l'outil :

pip install cruise-control-client

La fin de la sortie sera similaire à ceci :

...
Installing collected packages: urllib3, idna, charset-normalizer, certifi, requests, cruise-control-client
Successfully installed certifi-2024.2.2 charset-normalizer-3.3.2 cruise-control-client-1.1.3 idna-3.7 requests-2.31.0 urllib3-2.2.1

L'installateur a lié la commande cccli au paquet Python approprié.

Pour obtenir des statistiques sur la charge actuelle du cluster, exécutez la commande suivante en passant l'adresse de votre instance Cruise Control à -a :

cccli -a localhost:9090 load

La sortie détaillera les paramètres qu'elle surveille, fournis par le rapporteur de métriques :

Starting long-running poll of http://localhost:9090/kafkacruisecontrol/load?allow_capacity_estimation=False


              HOST         BROKER              RACK         DISK_CAP(MB)            DISK(MB)/_(%)_            CORE_NUM         CPU(%)          NW_IN_CAP(KB/s)       LEADER_NW_IN(KB/s)     FOLLOWER_NW_IN(KB/s)         NW_OUT_CAP(KB/s)        NW_OUT(KB/s)       PNW_OUT(KB/s)    LEADERS/REPLICAS
kafka1.your_domain,             1,kafka1.your_domain,          100000.000,              0.023/00.00,                  1,         0.999,               10000.000,                   0.012,                   0.002,               10000.000,              0.028,              0.080,            23/45
kafka2.your_domain,             2,kafka2.your_domain,          100000.000,              0.238/00.00,                  1,         0.819,               10000.000,                   0.028,                   0.009,               10000.000,              0.066,              0.105,            27/48
kafka3.your_domain,             3,kafka3.your_domain,          100000.000,              0.383/00.00,                  1,         1.292,               10000.000,                   0.041,                   0.006,               10000.000,              0.081,              0.108,            28/49

Le régulateur de vitesse peut être configuré pour réparer automatiquement le cluster en cas de défaillance d'un courtier, de violation d'objectif ou d'anomalie de métrique. Exécutez la commande suivante pour activer le processus de réparation en cas de défaillance d'un courtier :

cccli -a localhost:9090 admin --enable-self-healing-for broker_failure

Ils montreront l'état ancien et nouveau du paramètre modifié :

{
selfHealingEnabledBefore: {BROKER_FAILURE=false}, selfHealingEnabledAfter: {BROKER_FAILURE=true}
}

À cette étape, vous avez téléchargé et configuré le rapporteur de métriques Cruise Control sur chaque nœud broker Kafka de votre cluster. Ensuite, vous avez configuré et démarré Cruise Control, qui surveille et optimise continuellement les opérations de votre cluster. Vous avez également vu comment installer et utiliser cccli, un outil pour surveiller et configurer Cruise Control à la volée depuis la ligne de commande. Pour plus d'informations sur ce qu'il offre, consultez la documentation officielle.

Conclusion

Vous pouvez maintenant gérer un cluster Kafka depuis votre code en utilisant l'API KafkaAdminClient. Vous avez également appris à utiliser kcat et à configurer Cruise Control pour augmenter l'efficacité et la fiabilité de votre cluster Kafka.

L'auteur a choisi la Apache Software Foundation pour recevoir un don dans le cadre du programme Write for Donations.