Produire et consommer des messages avec les outils de ligne de commande Kafka

Découvrez comment utiliser les outils de ligne de commande Kafka pour vous connecter à un cluster Managed Service pour Apache Kafka, produire des messages et les consommer.

Avant de commencer

Avant de commencer ce tutoriel, créez un cluster Managed Service pour Apache Kafka. Si vous disposez déjà d'un cluster, vous pouvez ignorer cette étape.

Créer un cluster

Console

Accédez à la page Managed Service pour Apache Kafka > Clusters.
accéder aux clusters
Cliquez sur Créer.
Dans le champ Nom du cluster, saisissez un nom pour le cluster.
Dans la liste Région, sélectionnez un emplacement pour le cluster.
Pour Configuration réseau, configurez le sous-réseau où le cluster est accessible :
1. Pour Project (Projet), sélectionnez votre projet.
2. Pour Réseau, sélectionnez le réseau VPC.
3. Pour Sous-réseau, sélectionnez le sous-réseau.
4. Cliquez sur OK.
Cliquez sur Créer.

Une fois que vous avez cliqué sur Créer, l'état du cluster est Creating. Lorsque le cluster est prêt, l'état est Active.

gcloud

Pour créer un cluster Kafka, exécutez la commande managed-kafka clusters create.

gcloud managed-kafka clusters create KAFKA_CLUSTER \
--location=REGION \
--cpu=3 \
--memory=3GiB \
--subnets=projects/PROJECT_ID/regions/REGION/subnetworks/SUBNET_NAME \
--async

Remplacez les éléments suivants :

KAFKA_CLUSTER : nom du cluster Kafka
REGION : emplacement du cluster
PROJECT_ID : ID de votre projet
SUBNET_NAME : sous-réseau dans lequel vous souhaitez créer le cluster, par exemple default

Pour en savoir plus sur les emplacements compatibles, consultez Emplacements Managed Service pour Apache Kafka.

La commande s'exécute de manière asynchrone et renvoie un ID d'opération :

Check operation [projects/PROJECT_ID/locations/REGION/operations/OPERATION_ID] for status.

Pour suivre la progression de l'opération de création, utilisez la commande gcloud managed-kafka operations describe :

gcloud managed-kafka operations describe OPERATION_ID \
  --location=REGION

Lorsque le cluster est prêt, le résultat de cette commande inclut l'entrée state: ACTIVE. Pour en savoir plus, consultez Surveiller l'opération de création de cluster.

Rôles requis

Pour obtenir les autorisations nécessaires pour créer et configurer une VM cliente, demandez à votre administrateur de vous accorder les rôles IAM suivants sur le projet :

Administrateur d'instances Compute (v1) (roles/compute.instanceAdmin.v1)
Administrateur de projet IAM (roles/resourcemanager.projectIamAdmin)
Lecteur de rôles (roles/iam.roleViewer)
Utilisateur du compte de service (roles/iam.serviceAccountUser)

Pour en savoir plus sur l'attribution de rôles, consultez la page Gérer l'accès aux projets, aux dossiers et aux organisations.

Vous pouvez également obtenir les autorisations requises avec des rôles personnalisés ou d'autres rôles prédéfinis.

Créer une VM cliente

Créez une instance de machine virtuelle (VM) Linux dans Compute Engine qui peut accéder au cluster Kafka. Lorsque vous configurez la VM, définissez les options suivantes :

Région : Créez la VM dans la même région que votre cluster Kafka.
Sous-réseau Créez la VM dans le même réseau VPC que le sous-réseau que vous avez utilisé dans la configuration de votre cluster Kafka. Pour en savoir plus, consultez Afficher les sous-réseaux d'un cluster.
Niveaux d'accès Attribuez le niveau d'accès https://www.googleapis.com/auth/cloud-platform à la VM. Ce champ d'application autorise la VM à envoyer des requêtes à l'API Managed Kafka.

Les étapes suivantes montrent comment définir ces options.

Console

Accédez à la page Créer une instance dans la console Google Cloud .

Créer une instance
Dans le volet Configuration de la machine, procédez comme suit :
1. Dans le champ Nom, spécifiez un nom pour votre instance. Pour en savoir plus, consultez Convention d'attribution de noms aux ressources.
2. Dans la liste Région, sélectionnez la même région que votre cluster Kafka.
3. Dans la liste Zone, sélectionnez une zone.
Dans le menu de navigation, cliquez sur Mise en réseau. Dans le volet Mise en réseau qui s'affiche, procédez comme suit :
1. Accédez à la section Interfaces réseau.
2. Pour développer l'interface réseau par défaut, cliquez sur la flèche .
3. Dans le champ Réseau, sélectionnez le réseau VPC.
4. Dans la liste Sous-réseau, sélectionnez le sous-réseau.
5. Cliquez sur OK.
Dans le menu de navigation, cliquez sur Sécurité. Dans le volet Sécurité qui s'affiche, procédez comme suit :
1. Pour Niveaux d'accès, sélectionnez Définir l'accès pour chaque API.
2. Dans la liste des niveaux d'accès, recherchez la liste déroulante Cloud Platform et sélectionnez Activé.
Cliquez sur Créer pour créer la VM.

gcloud

Pour créer l'instance de VM, utilisez la commande gcloud compute instances create.

gcloud compute instances create VM_NAME \
  --scopes=https://www.googleapis.com/auth/cloud-platform \
  --subnet=projects/PROJECT_ID/regions/REGION/subnetworks/SUBNET \
  --zone=ZONE

Remplacez les éléments suivants :

VM_NAME : nom de la VM
PROJECT_ID : ID de votre projet
REGION : région dans laquelle vous avez créé le cluster Kafka, par exemple us-central1
SUBNET : sous-réseau du même réseau VPC que celui utilisé dans la configuration du cluster
ZONE : zone de la région dans laquelle vous avez créé le cluster, par exemple us-central1-c

Pour en savoir plus sur la création d'une VM, consultez Créer une instance de VM dans un sous-réseau spécifique.

Attribuer des rôles IAM

Attribuez les rôles IAM (Identity and Access Management) suivants au compte de service Compute Engine par défaut :

Client Managed Kafka (roles/managedkafka.client)
Créateur de jetons du compte de service (roles/iam.serviceAccountTokenCreator)
Créateur de jetons OpenID du compte de service (roles/iam.serviceAccountOpenIdTokenCreator)

Console

Dans la console Google Cloud , accédez à la page IAM.

Accéder à IAM
Recherchez la ligne Compte de service Compute Engine par défaut, puis cliquez sur Modifier le compte principal.
Cliquez sur Ajouter un autre rôle, puis sélectionnez le rôle Client Kafka géré. Répétez cette étape pour les rôles Créateur de jetons du compte de service et Créateur de jetons OpenID du compte de service.
Cliquez sur Enregistrer.

gcloud

Pour attribuer des rôles IAM, utilisez la commande gcloud projects add-iam-policy-binding.

gcloud projects add-iam-policy-binding PROJECT_ID \
  --member="serviceAccount:PROJECT_NUMBER-compute@developer.gserviceaccount.com" \
  --role=roles/managedkafka.client

gcloud projects add-iam-policy-binding PROJECT_ID\
  --member="serviceAccount:PROJECT_NUMBER-compute@developer.gserviceaccount.com" \
  --role=roles/iam.serviceAccountTokenCreator

gcloud projects add-iam-policy-binding PROJECT_ID \
  --member="serviceAccount:PROJECT_NUMBER-compute@developer.gserviceaccount.com" \
  --role=roles/iam.serviceAccountOpenIdTokenCreator

Remplacez les éléments suivants :

PROJECT_ID : ID de votre projet
PROJECT_NUMBER : votre numéro de projet

Pour obtenir le numéro du projet, exécutez la commande gcloud projects describe :

gcloud projects describe PROJECT_ID

Pour en savoir plus, consultez Trouver le nom, le numéro et l'ID du projet.

Se connecter à la VM

Utilisez SSH pour vous connecter à l'instance de VM.

Console

Accédez à la page Instances de VM.

Accéder à la page Instances de VM
Dans la liste des instances de VM, recherchez le nom de la VM, puis cliquez sur SSH.

gcloud

Pour vous connecter à la VM, utilisez la commande gcloud compute ssh.

gcloud compute ssh VM_NAME \
  --project=PROJECT_ID \
  --zone=ZONE

Remplacez les éléments suivants :

VM_NAME : nom de la VM
PROJECT_ID : ID de votre projet
ZONE : zone dans laquelle vous avez créé la VM

Une configuration supplémentaire peut être requise pour la première utilisation de SSH. Pour en savoir plus, consultez À propos des connexions SSH.

Installer les outils de ligne de commande Kafka

Dans votre session SSH, exécutez les commandes suivantes pour installer les outils de ligne de commande Kafka.

Installez Java, qui est nécessaire pour exécuter les outils de ligne de commande Kafka, et wget pour faciliter le téléchargement des dépendances. Les commandes suivantes supposent que vous utilisez un environnement Debian Linux.
```
sudo apt-get install default-jre wget
```

Installez les outils de ligne de commande Kafka.

wget -O kafka_2.13-3.7.2.tgz https://dlcdn.apache.org/kafka/3.7.2/kafka_2.13-3.7.2.tgz
tar xfz kafka_2.13-3.7.2.tgz

Définissez les variables d'environnement suivantes :

export KAFKA_HOME=$(pwd)/kafka_2.13-3.7.2
export PATH=$PATH:$KAFKA_HOME/bin
export CLASSPATH=$CLASSPATH:$KAFKA_HOME/libs/release-and-dependencies/*:$KAFKA_HOME/libs/release-and-dependencies/dependency/*

Configurer l'authentification

Dans votre session SSH, procédez comme suit pour configurer la bibliothèque d'authentification Managed Service pour Apache Kafka.

Téléchargez la bibliothèque et installez-la en local.
```
wget https://github.com/googleapis/managedkafka/releases/download/v1.0.5/release-and-dependencies.zip
sudo apt-get install unzip
unzip -n -j release-and-dependencies.zip -d $KAFKA_HOME/libs/
```
Cette commande installe la bibliothèque dans le répertoire lib du répertoire d'installation de Kafka. Les outils de ligne de commande Kafka recherchent les dépendances Java dans ce répertoire.
À l'aide d'un éditeur de texte, créez un fichier nommé client.properties et collez-y le code suivant :
```
security.protocol=SASL_SSL
sasl.mechanism=OAUTHBEARER
sasl.login.callback.handler.class=com.google.cloud.hosted.kafka.auth.GcpLoginCallbackHandler
sasl.jaas.config=org.apache.kafka.common.security.oauthbearer.OAuthBearerLoginModule required;
```
Enregistrez le fichier. Ce fichier configure un client Kafka avec les paramètres suivants :
- Utilisez SASL_SSL pour une communication sécurisée avec le cluster Kafka.
- Utilisez des jetons de support OAuth 2.0 pour l'authentification.
- Utilisez la classe GcpLoginCallbackHandler fournie par la bibliothèque comme gestionnaire de rappel de connexion pour obtenir des jetons OAuth 2.0.

Produire et consommer des messages

Depuis votre session SSH, exécutez les commandes suivantes pour produire et consommer des messages Kafka.

Définissez l'adresse d'amorçage en tant que variable d'environnement.
```
export BOOTSTRAP=bootstrap.CLUSTER_ID.REGION.managedkafka.PROJECT_ID.cloud.goog:9092
```
Remplacez les éléments suivants :
- CLUSTER_ID : nom du cluster
- REGION : emplacement où vous avez créé le cluster
- PROJECT_ID : ID de votre projet
Pour en savoir plus, consultez Obtenir l'adresse bootstrap.

Affichez la liste des thèmes du cluster.

kafka-topics.sh --list \
  --bootstrap-server $BOOTSTRAP \
  --command-config client.properties

Rédige un message dans un sujet.

echo "hello world" | kafka-console-producer.sh \
  --topic KAFKA_TOPIC_NAME \
  --bootstrap-server $BOOTSTRAP \
  --producer.config client.properties

Remplacez KAFKA_TOPIC_NAME par le nom du sujet.

Consommer les messages du sujet

kafka-console-consumer.sh \
  --topic KAFKA_TOPIC_NAME \
  --from-beginning \
  --bootstrap-server $BOOTSTRAP \
  --consumer.config client.properties

Pour arrêter de consommer des messages, appuyez sur Ctrl+C.

Exécutez un test de performances du producteur.

kafka-producer-perf-test.sh \
  --topic KAFKA_TOPIC_NAME \
  --num-records 1000000 --throughput 1000 --print-metrics --record-size 1024 \
  --producer-props bootstrap.servers=$BOOTSTRAP \
  --producer.config client.properties

Effectuer un nettoyage

Pour éviter que les ressources utilisées dans cette démonstration soient facturées sur votre compte Google Cloud , procédez comme suit :

Console

Supprimez l'instance de VM.
1. Accédez à la page Instances de VM.
  
  Accéder à la page Instances de VM
2. Sélectionnez la VM, puis cliquez sur Supprimer.
Supprimez le cluster Kafka.
1. Accédez à la page Managed Service pour Apache Kafka > Clusters.
  
  accéder aux clusters
2. Sélectionnez le cluster Kafka, puis cliquez sur Supprimer.

gcloud

Pour supprimer la VM, utilisez la commande gcloud compute instances delete.
```
gcloud compute instances delete VM_NAME --zone=ZONE
```
Pour supprimer le cluster Kafka, utilisez la commande gcloud managed-kafka clusters delete.
```
gcloud managed-kafka clusters delete CLUSTER_ID \
  --location=REGION --async
```

Étapes suivantes

Apache Kafka® est une marque déposée d'Apache Software Foundation ou de ses filiales aux États-Unis et/ou dans d'autres pays.