Connecter Pub/Sub à Apache Kafka

Ce document explique comment intégrer Apache Kafka et Pub/Sub à l'aide du connecteur Kafka pour les groupes Pub/Sub.

Remarque : Vous pouvez également utiliser ce connecteur avec Pub/Sub Lite. Consultez Connecter Pub/Sub Lite à Apache Kafka.

À propos du connecteur Pub/Sub Group Kafka

Apache Kafka est une plate-forme Open Source pour les événements de traitement en flux continu. Il est couramment utilisé dans les architectures distribuées pour permettre la communication entre des composants faiblement couplés. Pub/Sub est un service géré permettant d'envoyer et de recevoir des messages de manière asynchrone. Comme avec Kafka, vous pouvez utiliser Pub/Sub pour communiquer entre les composants de votre architecture cloud.

Le connecteur Kafka Pub/Sub Group vous permet d'intégrer ces deux systèmes. Les connecteurs suivants sont inclus dans le fichier JAR du connecteur :

• Le connecteur de récepteur lit les enregistrements d'un ou de plusieurs sujets Kafka et les publie dans Pub/Sub.
• Le connecteur source lit les messages d'un sujet Pub/Sub et les publie dans Kafka.

Voici quelques scénarios dans lesquels vous pouvez utiliser le connecteur Kafka du groupe Pub/Sub :

• Vous migrez une architecture basée sur Kafka vers Google Cloud.
• Vous disposez d'un système d'interface qui stocke les événements dans Kafka en dehors deGoogle Cloud, mais vous utilisez également Google Cloud pour exécuter certains de vos services de backend, qui doivent recevoir les événements Kafka.
• Vous collectez les journaux d'une solution Kafka sur site et les envoyez àGoogle Cloud pour l'analyse des données.
• Vous disposez d'un système d'interface qui utilise Google Cloud, mais vous stockez également des données sur site à l'aide de Kafka.

Le connecteur nécessite Kafka Connect, qui est un framework permettant de diffuser des données en continu entre Kafka et d'autres systèmes. Pour utiliser le connecteur, vous devez exécuter Kafka Connect en même temps que votre cluster Kafka.

Dans ce document, nous partons du principe que vous connaissez bien Kafka et Pub/Sub. Avant de lire ce document, nous vous conseillons de suivre l'un des guides de démarrage rapide de Pub/Sub.

Le connecteur Pub/Sub n'est compatible avec aucune intégration entre Google Cloud les LCA IAM et Kafka Connect.

Premiers pas avec le connecteur

Cette section vous explique comment effectuer les tâches suivantes :

1. Configurez le connecteur Kafka du groupe Pub/Sub.
2. Envoyez des événements de Kafka vers Pub/Sub.
3. Envoyez des messages de Pub/Sub vers Kafka.

Remarque : Cette section utilise Kafka Connect en mode autonome, ce qui convient aux tests. Pour les systèmes de production, le mode distribué est recommandé. Pour en savoir plus sur l'exécution de Kafka Connect en mode distribué, consultez Exécuter Kafka Connect dans la documentation Apache Kafka.

Prérequis

Installer Kafka

Suivez le guide de démarrage rapide d'Apache Kafka pour installer un cluster Kafka à nœud unique sur votre ordinateur local. Suivez ces étapes dans le guide de démarrage rapide :

1. Téléchargez la dernière version de Kafka et extrayez-la.
2. Démarrez l'environnement Kafka.
3. Créez un sujet Kafka.

Authentifier

Le connecteur Kafka du groupe Pub/Sub doit s'authentifier auprès de Pub/Sub pour envoyer et recevoir des messages Pub/Sub. Pour configurer l'authentification, procédez comme suit :

Connectez-vous à votre compte Google Cloud . Si vous débutez sur Google Cloud, créez un compte pour évaluer les performances de nos produits en conditions réelles. Les nouveaux clients bénéficient également de 300 $ de crédits sans frais pour exécuter, tester et déployer des charges de travail.
Installez la Google Cloud CLI.
Si vous utilisez un fournisseur d'identité (IdP) externe, vous devez d'abord vous connecter à la gcloud CLI avec votre identité fédérée.
Pour initialiser la gcloud CLI, exécutez la commande suivante :
gcloud init
Créez ou sélectionnez un projet Google Cloud .
Rôles requis pour sélectionner ou créer un projet
• Sélectionner un projet : la sélection d'un projet ne nécessite pas de rôle IAM spécifique. Vous pouvez sélectionner n'importe quel projet pour lequel un rôle vous a été attribué.
• Créer un projet : pour créer un projet, vous devez disposer du rôle Créateur de projet (roles/resourcemanager.projectCreator), qui contient l'autorisation resourcemanager.projects.create. Découvrez comment attribuer des rôles.
Remarque : Si vous ne comptez pas conserver les ressources créées dans cette procédure, créez un projet au lieu d'en sélectionner un existant. Après avoir suivi ces étapes, vous pouvez supprimer le projet. Cela entraîne la suppression de toutes les ressources qui lui sont associées.
• Créez un projet Google Cloud  :

  gcloud projects create PROJECT_ID

  Remplacez PROJECT_ID par le nom du projet Google Cloud que vous créez.

• Sélectionnez le projet Google Cloud que vous avez créé :

  gcloud config set project PROJECT_ID

  Remplacez PROJECT_ID par le nom de votre projet Google Cloud .
Créez des identifiants d'authentification locaux pour votre compte utilisateur :
gcloud auth application-default login
Si une erreur d'authentification est renvoyée et que vous utilisez un fournisseur d'identité (IdP) externe, vérifiez que vous vous êtes connecté à la gcloud CLI avec votre identité fédérée.
Attribuez des rôles à votre compte utilisateur. Exécutez la commande suivante une fois pour chacun des rôles IAM suivants : roles/pubsub.admin
gcloud projects add-iam-policy-binding PROJECT_ID --member="user:USER_IDENTIFIER" --role=ROLE
Remplacez les éléments suivants :
• PROJECT_ID : ID du projet.
• USER_IDENTIFIER : identifiant de votre compte utilisateur . Par exemple, [e-mail].
• ROLE : rôle IAM que vous attribuez à votre compte utilisateur.

Installez la Google Cloud CLI.

Si vous utilisez un fournisseur d'identité (IdP) externe, vous devez d'abord vous connecter à la gcloud CLI avec votre identité fédérée.

Pour initialiser la gcloud CLI, exécutez la commande suivante :

gcloud init

Créez ou sélectionnez un projet Google Cloud .

Rôles requis pour sélectionner ou créer un projet

• Sélectionner un projet : la sélection d'un projet ne nécessite pas de rôle IAM spécifique. Vous pouvez sélectionner n'importe quel projet pour lequel un rôle vous a été attribué.
• Créer un projet : pour créer un projet, vous devez disposer du rôle Créateur de projet (roles/resourcemanager.projectCreator), qui contient l'autorisation resourcemanager.projects.create. Découvrez comment attribuer des rôles.

Remarque : Si vous ne comptez pas conserver les ressources créées dans cette procédure, créez un projet au lieu d'en sélectionner un existant. Après avoir suivi ces étapes, vous pouvez supprimer le projet. Cela entraîne la suppression de toutes les ressources qui lui sont associées.

• Créez un projet Google Cloud  :

  gcloud projects create PROJECT_ID

  Remplacez PROJECT_ID par le nom du projet Google Cloud que vous créez.

• Sélectionnez le projet Google Cloud que vous avez créé :

  gcloud config set project PROJECT_ID

  Remplacez PROJECT_ID par le nom de votre projet Google Cloud .

Créez des identifiants d'authentification locaux pour votre compte utilisateur :

gcloud auth application-default login

Si une erreur d'authentification est renvoyée et que vous utilisez un fournisseur d'identité (IdP) externe, vérifiez que vous vous êtes connecté à la gcloud CLI avec votre identité fédérée.

Attribuez des rôles à votre compte utilisateur. Exécutez la commande suivante une fois pour chacun des rôles IAM suivants : roles/pubsub.admin

gcloud projects add-iam-policy-binding PROJECT_ID --member="user:USER_IDENTIFIER" --role=ROLE

Remplacez les éléments suivants :

• PROJECT_ID : ID du projet.
• USER_IDENTIFIER : identifiant de votre compte utilisateur . Par exemple, [e-mail].
• ROLE : rôle IAM que vous attribuez à votre compte utilisateur.

Remarque : Pour les charges de travail de production, nous vous recommandons d'utiliser la fédération d'identité de charge de travail pour l'authentification.

Télécharger le fichier JAR du connecteur

Téléchargez le fichier JAR du connecteur sur votre ordinateur local. Pour en savoir plus, consultez Obtenir le connecteur dans le fichier README de GitHub.

Copier les fichiers de configuration du connecteur

1. Clonez ou téléchargez le dépôt GitHub du connecteur.

   git clone https://github.com/googleapis/java-pubsub-group-kafka-connector.git
   cd java-pubsub-group-kafka-connector
   

2. Copiez le contenu du répertoire config dans le sous-répertoire config de votre installation Kafka.

   cp config/* [path to Kafka installation]/config/
   

Ces fichiers contiennent les paramètres de configuration du connecteur.

Mettre à jour votre configuration Kafka Connect

1. Accédez au répertoire contenant le binaire Kafka Connect que vous avez téléchargé.
2. Dans le répertoire binaire Kafka Connect, ouvrez le fichier nommé config/connect-standalone.properties dans un éditeur de texte.
3. Si plugin.path property est mis en commentaire, annulez la mise en commentaire.

4. Mettez à jour plugin.path property pour inclure le chemin d'accès au fichier JAR du connecteur.

   Exemple :

   plugin.path=/home/PubSubKafkaConnector/pubsub-group-kafka-connector-1.0.0.jar
   

5. Définissez la propriété offset.storage.file.filename sur un nom de fichier local. En mode autonome, Kafka utilise ce fichier pour stocker les données de décalage.

   Exemple :

   offset.storage.file.filename=/tmp/connect.offsets
   

Transférer des événements de Kafka vers Pub/Sub

Cette section explique comment démarrer le connecteur récepteur, publier des événements sur Kafka, puis lire les messages transférés depuis Pub/Sub.

1. Utilisez la Google Cloud CLI pour créer un sujet Pub/Sub avec un abonnement.

   gcloud pubsub topics create PUBSUB_TOPIC
   gcloud pubsub subscriptions create PUBSUB_SUBSCRIPTION --topic=PUBSUB_TOPIC

   Remplacez les éléments suivants :

   • PUBSUB_TOPIC : nom d'un sujet Pub/Sub permettant de recevoir les messages de Kafka.
   • PUBSUB_SUBSCRIPTION : nom d'un abonnement Pub/Sub pour le sujet.

2. Ouvrez le fichier /config/cps-sink-connector.properties dans un éditeur de texte. Ajoutez des valeurs pour les propriétés suivantes, qui sont marquées "TODO" dans les commentaires :

   topics=KAFKA_TOPICS
   cps.project=PROJECT_ID
   cps.topic=PUBSUB_TOPIC

   Remplacez les éléments suivants :

   • KAFKA_TOPICS : liste de sujets Kafka à lire, séparés par une virgule.
   • PROJECT_ID : projet Google Cloud contenant votre sujet Pub/Sub.
   • PUBSUB_TOPIC : sujet Pub/Sub permettant de recevoir les messages de Kafka.

3. Dans le répertoire Kafka, exécutez la commande suivante :

   bin/connect-standalone.sh \
     config/connect-standalone.properties \
     config/cps-sink-connector.properties
   

4. Suivez les étapes du guide de démarrage rapide Apache Kafka pour écrire des événements dans votre sujet Kafka.

5. Utilisez la gcloud CLI pour lire les événements à partir de Pub/Sub.

   gcloud pubsub subscriptions pull PUBSUB_SUBSCRIPTION --auto-ack

Transférer des messages de Pub/Sub vers Kafka

Cette section explique comment démarrer le connecteur source, publier des messages sur Pub/Sub et lire les messages transférés depuis Kafka.

1. Utilisez la gcloud CLI pour créer un sujet Pub/Sub avec un abonnement.

   gcloud pubsub topics create PUBSUB_TOPIC
   gcloud pubsub subscriptions create PUBSUB_SUBSCRIPTION --topic=PUBSUB_TOPIC

   Remplacez les éléments suivants :

   • PUBSUB_TOPIC : nom d'un sujet Pub/Sub.
   • PUBSUB_SUBSCRIPTION : nom d'un abonnement Pub/Sub.

2. Ouvrez le fichier nommé /config/cps-source-connector.properties dans un éditeur de texte. Ajoutez des valeurs pour les propriétés suivantes, qui sont marquées "TODO" dans les commentaires :

   kafka.topic=KAFKA_TOPIC
   cps.project=PROJECT_ID
   cps.subscription=PUBSUB_SUBSCRIPTION

   Remplacez les éléments suivants :

   • KAFKA_TOPIC : sujet Kafka pour recevoir les messages Pub/Sub.
   • PROJECT_ID : projet Google Cloud contenant votre sujet Pub/Sub.
   • PUBSUB_TOPIC : sujet Pub/Sub.

3. Dans le répertoire Kafka, exécutez la commande suivante :

   bin/connect-standalone.sh \
     config/connect-standalone.properties \
     config/cps-source-connector.properties
   

4. Utilisez la gcloud CLI pour publier un message dans Pub/Sub.

   gcloud pubsub topics publish PUBSUB_TOPIC --message="message 1"

5. Lire le message de Kafka. Suivez les étapes du guide de démarrage rapide Apache Kafka pour lire les messages du sujet Kafka.

Conversion par message

Un enregistrement Kafka contient une clé et une valeur, qui sont des tableaux d'octets de longueur variable. Un enregistrement Kafka peut également comporter des en-têtes (paires clé/valeur). Un message Pub/Sub comporte deux parties principales : le corps du message et zéro ou plusieurs attributs clé-valeur.

Kafka Connect utilise des convertisseurs pour sérialiser les clés et les valeurs vers et depuis Kafka. Pour contrôler la sérialisation, définissez les propriétés suivantes dans les fichiers de configuration du connecteur :

• key.converter : convertisseur utilisé pour sérialiser les clés d'enregistrement.
• value.converter : convertisseur utilisé pour sérialiser les valeurs d'enregistrement.

Le corps d'un message Pub/Sub est un objet ByteString. La conversion la plus efficace consiste donc à copier directement la charge utile. Pour cette raison, nous vous recommandons d'utiliser un convertisseur qui produit des types de données primitifs (schéma d'entier, de float, de chaîne ou d'octets) dans la mesure du possible, afin d'éviter de désérialiser et de resérialiser le même corps du message.

Conversion de Kafka vers Pub/Sub

Le connecteur récepteur convertit les enregistrements Kafka en messages Pub/Sub comme suit :

• La clé d'enregistrement Kafka est stockée en tant qu'attribut nommé "key" dans le message Pub/Sub.
• Par défaut, le connecteur supprime tous les en-têtes de l'enregistrement Kafka. Toutefois, si vous définissez l'option de configuration headers.publish sur true, le connecteur écrit les en-têtes en tant qu'attributs Pub/Sub. Le connecteur ignore les en-têtes qui dépassent les limites de Pub/Sub concernant les attributs de message.
• Pour les schémas d'entiers, de floats, de chaînes et d'octets, le connecteur transmet les octets de la valeur d'enregistrement Kafka directement dans le corps du message Pub/Sub.
• Pour les schémas de structure, le connecteur écrit chaque champ en tant qu'attribut du message Pub/Sub. Par exemple, si le champ est { "id"=123 }, le message Pub/Sub obtenu comporte un attribut "id"="123". La valeur du champ est toujours convertie en chaîne. Les types de cartes et de structures ne sont pas acceptés en tant que types de champs dans une structure.
• Pour les schémas de mappage, le connecteur écrit chaque paire clé/valeur en tant qu'attribut du message Pub/Sub. Par exemple, si le mappage est {"alice"=1,"bob"=2}, le message Pub/Sub obtenu comporte deux attributs, "alice"="1" et "bob"="2". Les clés et les valeurs sont converties en chaînes.

Les schémas de structure et de carte présentent des comportements supplémentaires :

• Vous pouvez également spécifier un champ structuré ou une clé de mappage en particulier pour qu'il s'agisse du corps du message, en définissant la propriété de configuration messageBodyName. La valeur du champ ou de la clé est stockée sous la forme d'un ByteString dans le corps du message. Si vous ne définissez pas messageBodyName, le corps du message est vide pour les schémas de structure et de carte.

• Pour les valeurs de tableau, le connecteur n'accepte que les types de tableau primitifs. La séquence de valeurs du tableau est concaténée dans un seul objet ByteString.

Conversion de Pub/Sub vers Kafka

Le connecteur source convertit les messages Pub/Sub en enregistrements Kafka comme suit :

• Clé d'enregistrement Kafka : par défaut, la clé est définie sur null. Vous pouvez également spécifier un attribut de message Pub/Sub à utiliser comme clé en définissant l'option de configuration kafka.key.attribute. Dans ce cas, le connecteur recherche un attribut portant ce nom et définit la clé d'enregistrement sur la valeur de l'attribut. Si l'attribut spécifié n'est pas présent, la clé d'enregistrement est définie sur null.

• Valeur de l'enregistrement Kafka. Le connecteur écrit la valeur de l'enregistrement comme suit :

  • Si le message Pub/Sub ne comporte aucun attribut personnalisé, le connecteur écrit le corps du message Pub/Sub directement dans la valeur de l'enregistrement Kafka en tant que type byte[], à l'aide du convertisseur spécifié par value.converter.

  • Si le message Pub/Sub comporte des attributs personnalisés et que kafka.record.headers est défini sur false, le connecteur écrit un struct dans la valeur de l'enregistrement. La structure contient un champ pour chaque attribut et un champ nommé "message" dont la valeur est le corps du message Pub/Sub (stocké sous forme d'octets) :

    {
      "message": "<Pub/Sub message body>",
      "<attribute-1>": "<value-1>",
      "<attribute-2>": "<value-2>",
      ....
    }
    

    Dans ce cas, vous devez utiliser un value.converter compatible avec les schémas struct, tel que org.apache.kafka.connect.json.JsonConverter.

  • Si le message Pub/Sub comporte des attributs personnalisés et que kafka.record.headers est défini sur true, le connecteur écrit les attributs en tant qu'en-têtes d'enregistrement Kafka. Il écrit le corps du message Pub/Sub directement dans la valeur de l'enregistrement Kafka en tant que type byte[], à l'aide du convertisseur spécifié par value.converter.

• En-têtes d'enregistrement Kafka. Par défaut, les en-têtes sont vides, sauf si vous définissez kafka.record.headers sur true.

Remarque : Dans la configuration de votre broker Kafka, définissez message.max.bytes sur une valeur suffisamment élevée pour gérer vos messages Pub/Sub. Pub/Sub impose une taille maximale de message.

Options de configuration

En plus des configurations fournies par l'API Kafka Connect, le connecteur Kafka du groupe Pub/Sub est compatible avec la configuration du récepteur et de la source, comme décrit dans Configurations du connecteur Pub/Sub.

Assistance

Si vous avez besoin d'aide, créez une demande d'assistance. Pour les questions et discussions générales, créez un problème dans le dépôt GitHub.

Étapes suivantes

• Comprendre les différences entre Kafka et Pub/Sub
• En savoir plus sur le connecteur Kafka du groupe Pub/Sub
• Consultez le dépôt GitHub du connecteur Kafka pour les groupes Pub/Sub.