Créer un sujet d'importation Confluent Cloud

Un sujet d'importation Confluent Cloud vous permet d'ingérer en continu des données depuis Confluent Cloud en tant que source externe et dans Pub/Sub. Vous pouvez ensuite diffuser les données vers l'une des destinations compatibles avec Pub/Sub.

Ce document vous explique comment créer et gérer des sujets d'importation Confluent Cloud. Pour créer un sujet standard, consultez Créer un sujet standard.

Pour en savoir plus sur les thèmes d'importation, consultez À propos des thèmes d'importation.

Avant de commencer

• En savoir plus sur le processus de publication Pub/Sub

• Configurez les rôles et autorisations requis pour gérer les thèmes d'importation Confluent Cloud, y compris les suivants :

  • Ajouter le rôle d'éditeur Pub/Sub au compte de service Pub/Sub

  • Ajoutez le rôle d'agent de service Pub/Sub au compte de service Pub/Sub, s'il n'est pas déjà attribué.

  • Ajouter le rôle "Utilisateur du compte de service" au compte de service

• Configurez la fédération d'identité de charge de travail pour queTrusted Cloud puisse accéder au service de streaming externe.

Rôles et autorisations requis

Pour obtenir les autorisations nécessaires pour créer et gérer des sujets d'importation Confluent Cloud, demandez à votre administrateur de vous accorder le rôle IAM Éditeur Pub/Sub (roles/pubsub.editor) sur votre sujet ou votre projet. 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.

Ce rôle prédéfini contient les autorisations requises pour créer et gérer des sujets d'importation Confluent Cloud. Pour connaître les autorisations exactes requises, développez la section Autorisations requises :

Autorisations requises

Vous devez disposer des autorisations suivantes pour créer et gérer des sujets d'importation Confluent Cloud :

• Créez un sujet d'importation : pubsub.topics.create
• Supprimez un thème d'importation : pubsub.topics.delete
• Obtenir un sujet d'importation : pubsub.topics.get
• Lister un sujet d'importation : pubsub.topics.list
• Publier dans un sujet d'importation : pubsub.topics.publish and pubsub.serviceAgent
• Mettre à jour un thème d'importation : pubsub.topics.update
• Obtenez la stratégie IAM pour un sujet d'importation : pubsub.topics.getIamPolicy
• Configurez la stratégie IAM pour un sujet d'importation : pubsub.topics.setIamPolicy

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

Vous pouvez configurer le contrôle des accès au niveau du projet et au niveau de chaque ressource.

Configurer l'identité fédérée pour accéder à Confluent Cloud

La fédération d'identité de charge de travail permet aux services Trusted Cloud d'accéder aux charges de travail exécutées en dehors de Trusted Cloud. Grâce à la fédération d'identité, vous n'avez pas besoin de gérer ni de transmettre des identifiants à Trusted Cloud pour accéder à vos ressources dans d'autres clouds. À la place, vous pouvez utiliser les identités des charges de travail elles-mêmes pour vous authentifier auprès de Trusted Cloud et accéder aux ressources.

Créer un compte de service dans Trusted Cloud

Il s'agit d'une étape facultative. Si vous disposez déjà d'un compte de service, vous pouvez l'utiliser dans cette procédure au lieu d'en créer un autre. Si vous utilisez un compte de service existant, accédez à Enregistrer l'ID unique du compte de service pour l'étape suivante.

Pour les sujets d'importation Confluent Cloud, Pub/Sub utilise le compte de service comme identité pour accéder aux ressources de Confluent Cloud.

Pour en savoir plus sur la création d'un compte de service, y compris sur les prérequis, les rôles et autorisations requis, et les consignes de dénomination, consultez Créer des comptes de service. Une fois le compte de service créé, vous devrez peut-être attendre au moins 60 secondes avant de l'utiliser. Ce comportement se produit car les opérations de lecture sont cohérentes à terme. Il peut s'écouler un certain temps avant que le nouveau compte de service devienne visible.

Enregistrez l'ID unique du compte de service.

Vous avez besoin d'un ID unique de compte de service pour configurer le fournisseur d'identité et le pool dans la console Confluent Cloud.

1. Dans la console Trusted Cloud , accédez à la page d'informations Compte de service.

   Accéder à la page "Comptes de service"

2. Cliquez sur le compte de service que vous venez de créer ou sur celui que vous prévoyez d'utiliser.

3. Sur la page Détails du compte de service, notez le numéro d'ID unique.

   Vous aurez besoin de cet ID pour configurer le fournisseur d'identité et le pool dans la console Confluent Cloud.

Ajouter le rôle de créateur de jetons du compte de service au compte de service Pub/Sub

Le rôle de créateur de jetons du compte de service (roles/iam.serviceAccountTokenCreator) permet aux comptes principaux de créer des identifiants éphémères pour un compte de service. Ces jetons ou identifiants sont utilisés pour emprunter l'identité du compte de service.

Pour plus d'informations sur l'emprunt de l'identité d'un compte de service, consultez la page Emprunter l'identité d'un compte de service.

Vous pouvez également ajouter le rôle Éditeur Pub/Sub (roles/pubsub.publisher) au cours de cette procédure. Pour en savoir plus sur le rôle et la raison pour laquelle vous l'ajoutez, consultez Ajouter le rôle d'éditeur Pub/Sub au compte de service Pub/Sub.

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

   Accéder à IAM

2. Cochez la case Inclure les attributions de rôles fournies par S3NS.

3. Recherchez le compte de service au format service-{PROJECT_NUMBER}@gcp-sa-pubsub.s3ns-system.iam.gserviceaccount.com.

4. Pour ce compte de service, cliquez sur le bouton Modifier le compte principal.

5. Si nécessaire, cliquez sur Ajouter un autre rôle.

6. Recherchez et cliquez sur le rôle Créateur de jetons du compte de service (roles/iam.serviceAccountTokenCreator).

7. Cliquez sur Enregistrer.

Créer un fournisseur d'identité dans Confluent Cloud

Pour s'authentifier auprès de Confluent Cloud, le compte de service Google Cloud a besoin d'un pool d'identités. Vous devez d'abord créer un fournisseur d'identité dans Confluent Cloud.

Pour savoir comment créer un fournisseur d'identité dans Confluent Cloud, consultez la page Ajouter un fournisseur d'identité OAuth/OIDC.

1. Connectez-vous à la console Confluent Cloud.

2. Dans le menu, cliquez sur Comptes et accès.

3. Cliquez sur Identités de charge de travail.

4. Cliquez sur Ajouter un fournisseur.

5. Cliquez sur OAuth/OIDC, puis sur Next (Suivant).

6. Cliquez sur Autre fournisseur OIDC, puis sur Suivant.

7. Indiquez un nom et une description de l'objectif du fournisseur d'identité.

8. Cliquez sur Afficher la configuration avancée.

9. Dans le champ URI de l'émetteur, saisissez https://accounts.google.com.

10. Dans le champ URI JWKS, saisissez https://www.googleapis.com/oauth2/v3/certs.

11. Cliquez sur Valider et enregistrer.

Créez un pool d'identités et accordez les rôles appropriés dans Confluent Cloud.

Vous devez créer un pool d'identités sous votre profil d'identité et accorder les rôles nécessaires pour permettre au compte de service Pub/Sub de s'authentifier et de lire les sujets Kafka de Confluent Cloud.

Assurez-vous que votre cluster est créé dans Confluent Cloud avant de créer un pool d'identités.

Pour savoir comment créer un pool d'identités, consultez la page Utiliser des pools d'identités avec votre fournisseur d'identité OAuth/OIDC.

1. Connectez-vous à la console Confluent Cloud.

2. Dans le menu, cliquez sur Comptes et accès.

3. Cliquez sur Identités de charge de travail.

4. Cliquez sur le fournisseur d'identité que vous avez créé dans Créer un fournisseur d'identité dans Confluent Cloud.

5. Cliquez sur Add pool (Ajouter un pool).

6. Saisissez un nom et une description pour votre pool d'identités.

7. Définissez Revendication d'identité sur claims.

8. Sous Définir des filtres, cliquez sur l'onglet Avancé. Saisissez le code ci-dessous :

   claims.iss=='https://accounts.google.com' && claims.sub=='<SERVICE_ACCOUNT_UNIQUE_ID>'
   

   Remplacez <SERVICE_ACCOUNT_UNIQUE_ID> par l'ID unique de votre compte de service, que vous trouverez dans Enregistrer l'ID unique du compte de service.

9. Cliquez sur Suivant.

10. Cliquez sur Ajouter une autorisation. Cliquez ensuite sur Suivant.

11. Dans le cluster concerné, cliquez sur Ajouter une attribution de rôle.

12. Cliquez sur le rôle Opérateur, puis sur Ajouter.

    Ce rôle accorde Pub/Sub. Accès du compte de service au cluster contenant le sujet Confluent Kafka que vous souhaitez ingérer dans Pub/Sub.

13. Sous le cluster, cliquez sur Thèmes. Cliquez ensuite sur Ajouter une attribution de rôle.

14. Sélectionnez le rôle DeveloperRead.

15. Cliquez sur l'option appropriée et spécifiez le thème ou le préfixe. Par exemple, Thème spécifique, Règle de préfixe ou Tous les thèmes.

16. Cliquez sur Ajouter.

17. Cliquez sur Suivant.

18. Cliquez sur Valider et enregistrer.

Ajouter le rôle d'éditeur Pub/Sub au principal Pub/Sub

Pour activer la publication, vous devez attribuer un rôle d'éditeur au compte de service Pub/Sub afin que Pub/Sub puisse publier dans le sujet d'importation Confluent Cloud.

Ajouter le rôle d'agent de service Pub/Sub au compte de service Pub/Sub

Pour permettre à Pub/Sub d'utiliser le quota de publication du projet de votre sujet d'importation, l'agent de service Pub/Sub doit disposer de l'autorisation serviceusage.services.use sur le projet de votre sujet d'importation.

Pour accorder cette autorisation, nous vous recommandons d'ajouter le rôle d'agent de service Pub/Sub au compte de service Pub/Sub.

Si le compte de service Pub/Sub ne dispose pas du rôle d'agent de service Pub/Sub, vous pouvez l'attribuer comme suit :

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

   Accéder à IAM

2. Cochez la case Inclure les attributions de rôles fournies par S3NS.

3. Recherchez le compte de service au format service-{PROJECT_NUMBER}@gcp-sa-pubsub.s3ns-system.iam.gserviceaccount.com.

4. Pour ce compte de service, cliquez sur le bouton Modifier le compte principal.

5. Si nécessaire, cliquez sur Ajouter un autre rôle.

6. Recherchez et cliquez sur le rôle Agent de service Pub/Sub (roles/pubsub.serviceAgent).

7. Cliquez sur Enregistrer.

Activer la publication à partir de tous les thèmes

Utilisez cette méthode si vous n'avez créé aucun sujet d'importation Confluent Cloud.

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

   Accéder à IAM

2. Cochez la case Inclure les attributions de rôles fournies par S3NS.

3. Recherchez le compte de service au format service-{PROJECT_NUMBER}@gcp-sa-pubsub.s3ns-system.iam.gserviceaccount.com.

4. Pour ce compte de service, cliquez sur le bouton Modifier le compte principal.

5. Si nécessaire, cliquez sur Ajouter un autre rôle.

6. Recherchez et cliquez sur le rôle Éditeur Pub/Sub (roles/pubsub.publisher).

7. Cliquez sur Enregistrer.

Activer la publication à partir d'un seul sujet

N'utilisez cette méthode que si le sujet d'importation Confluent Cloud existe déjà.

1. In the Trusted Cloud console, activate Cloud Shell.

   Activate Cloud Shell

   At the bottom of the Trusted Cloud console, a Cloud Shell session starts and displays a command-line prompt. Cloud Shell is a shell environment with the Google Cloud CLI already installed and with values already set for your current project. It can take a few seconds for the session to initialize.

2. Exécutez la commande gcloud pubsub topics add-iam-policy-binding :

   gcloud pubsub topics add-iam-policy-binding TOPIC_ID \
      --member="serviceAccount:service-PROJECT_NUMBER@gcp-sa-pubsub.s3ns-system.iam.gserviceaccount.com" \
      --role="roles/pubsub.publisher"

   Remplacez les éléments suivants :

   • TOPIC_ID : ID du thème d'importation Confluent Cloud.

   • PROJECT_NUMBER : numéro du projet. Pour afficher le numéro de projet, consultez Identifier des projets.

Ajouter le rôle "Utilisateur du compte de service" au compte de service

Le rôle Utilisateur de compte de service (roles/iam.serviceAccountUser) inclut l'autorisation iam.serviceAccounts.actAs, qui permet à un compte principal d'associer un compte de service aux paramètres d'ingestion du thème d'importation Confluent Cloud et d'utiliser ce compte de service pour l'identité fédérée.

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

   Accéder à IAM

2. Pour le compte principal qui émet les appels de création ou de mise à jour de sujets, cliquez sur le bouton Modifier le compte principal.

3. Si nécessaire, cliquez sur Ajouter un autre rôle.

4. Recherchez et cliquez sur le rôle Utilisateur du compte de service (roles/iam.serviceAccountUser).

5. Cliquez sur Enregistrer.

Utiliser les thèmes d'importation Confluent Cloud

Vous pouvez créer un sujet d'importation ou en modifier un existant.

Remarques

• Créer le sujet et l'abonnement séparément, même si vous le faites rapidement, peut entraîner une perte de données. Il existe une courte période pendant laquelle le thème est disponible sans abonnement. Si des données sont envoyées à la rubrique pendant cette période, elles sont perdues. En créant d'abord le sujet, puis l'abonnement, et en convertissant ensuite le sujet en sujet d'importation, vous vous assurez de ne manquer aucun message pendant le processus d'importation.

• Si vous devez recréer le sujet Kafka d'un sujet d'importation existant portant le même nom, vous ne pouvez pas simplement supprimer le sujet Kafka et le recréer. Cette action peut invalider la gestion des décalages de Pub/Sub, ce qui peut entraîner une perte de données. Pour limiter ce risque, procédez comme suit :

  • Supprimez le sujet d'importation Pub/Sub.
  • Supprimez le sujet Kafka.
  • Créez le sujet Kafka.
  • Créez le sujet d'importation Pub/Sub.

• Les données d'un sujet Confluent Cloud Kafka sont toujours lues à partir du décalage le plus ancien.

Créer un sujet d'importation Confluent Cloud

Pour en savoir plus sur les propriétés associées à un thème, consultez Propriétés d'un thème.

Assurez-vous d'avoir effectué les procédures suivantes :

• Configurer l'identité fédérée pour accéder à Confluent Cloud

• Ajouter le rôle d'éditeur Pub/Sub au compte de service Pub/Sub

• Ajouter le rôle "Utilisateur du compte de service" au compte de service

Pour créer un sujet d'importation Confluent Cloud, procédez comme suit :

Console

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

   Accéder aux sujets

2. Cliquez sur Create topic (Créer un sujet).
3. Dans le champ ID du sujet, saisissez un ID pour votre sujet d'importation. Pour en savoir plus sur la dénomination des thèmes, consultez les consignes de dénomination.
4. Sélectionnez Ajouter un abonnement par défaut.
5. Sélectionnez Activer l'ingestion.
6. Pour la source d'ingestion, sélectionnez Confluent Cloud.
7. Saisissez les informations suivantes :
   1. Serveur d'amorçage : serveur d'amorçage de votre cluster contenant le sujet Kafka que vous ingérez dans Pub/Sub. Le format est le suivant : hostname:port.
   2. ID du cluster : ID du cluster contenant le sujet Kafka que vous ingérez dans Pub/Sub.
   3. Sujet : nom du sujet Kafka que vous ingérez dans Pub/Sub.
   4. ID du pool d'identités : ID du pool d'identités utilisé pour l'authentification avec Confluent Cloud.
   5. Compte de service : compte de service que vous avez créé dans Créer un compte de service dans Google Cloud.
8. Cliquez sur Create topic (Créer un sujet).

gcloud

1. In the Trusted Cloud console, activate Cloud Shell.

   Activate Cloud Shell

   At the bottom of the Trusted Cloud console, a Cloud Shell session starts and displays a command-line prompt. Cloud Shell is a shell environment with the Google Cloud CLI already installed and with values already set for your current project. It can take a few seconds for the session to initialize.

2. Exécutez la commande gcloud pubsub topics create :

   gcloud pubsub topics create TOPIC_ID 
   
       --confluent-cloud-ingestion-bootstrap-server CONFLUENT_BOOTSTRAP_SERVER 
   
       --confluent-cloud-ingestion-cluster-id CONFLUENT_CLUSTER_ID 
   
       --confluent-cloud-ingestion-topic CONFLUENT_TOPIC 
   
       --confluent-cloud-ingestion-identity-pool-id CONFLUENT_IDENTITY_POOL_ID 
   
       --confluent-cloud-ingestion-service-account PUBSUB_SERVICE_ACCOUNT

   Remplacez les éléments suivants :

   • TOPIC_ID : nom ou ID de votre sujet Pub/Sub.
   • CONFLUENT_BOOTSTRAP_SERVER : serveur d'amorçage de votre cluster contenant le sujet Kafka que vous ingérez dans Pub/Sub. Le format est le suivant : hostname:port.
   • CONFLUENT_CLUSTER_ID : ID de votre cluster contenant le sujet Kafka que vous ingérez dans Pub/Sub.
   • CONFLUENT_TOPIC : nom du sujet Kafka que vous ingérez dans Pub/Sub.
   • CONFLUENT_IDENTITY_POOL_ID : ID du pool d'identités utilisé pour l'authentification avec Confluent Cloud.
   • PUBSUB_SERVICE_ACCOUNT : compte de service que vous avez créé dans Créer un compte de service dans Google Cloud.

C++

Avant d'essayer cet exemple, suivez les instructions d'installation dans le langage C++ qui se trouvent sur la page Démarrage rapide : utiliser des bibliothèques clientes. Pour en savoir plus, consultez la documentation de référence sur l'API Pub/Sub pour C++.

namespace pubsub = ::google::cloud::pubsub;
namespace pubsub_admin = ::google::cloud::pubsub_admin;
[](pubsub_admin::TopicAdminClient client, std::string project_id,
   std::string topic_id, std::string const& bootstrap_server,
   std::string const& cluster_id, std::string const& confluent_topic,
   std::string const& identity_pool_id,
   std::string const& gcp_service_account) {
  google::pubsub::v1::Topic request;
  request.set_name(
      pubsub::Topic(std::move(project_id), std::move(topic_id)).FullName());
  auto* confluent_cloud = request.mutable_ingestion_data_source_settings()
                              ->mutable_confluent_cloud();
  confluent_cloud->set_bootstrap_server(bootstrap_server);
  confluent_cloud->set_cluster_id(cluster_id);
  confluent_cloud->set_topic(confluent_topic);
  confluent_cloud->set_identity_pool_id(identity_pool_id);
  confluent_cloud->set_gcp_service_account(gcp_service_account);

  auto topic = client.CreateTopic(request);
  // Note that kAlreadyExists is a possible error when the library retries.
  if (topic.status().code() == google::cloud::StatusCode::kAlreadyExists) {
    std::cout << "The topic already exists\n";
    return;
  }
  if (!topic) throw std::move(topic).status();

  std::cout << "The topic was successfully created: " << topic->DebugString()
            << "\n";
}

Go

L'exemple suivant utilise la version majeure de la bibliothèque cliente Go Pub/Sub (v2). Si vous utilisez toujours la bibliothèque v1, consultez le guide de migration vers la v2. Pour consulter la liste des exemples de code de la version 1, consultez les exemples de code obsolètes.

Avant d'essayer cet exemple, suivez les instructions de configuration pour Go du guide de démarrage rapide : Utiliser les bibliothèques clientes. Pour en savoir plus, consultez la documentation de référence sur l'API Pub/Sub pour Go.

import (
	"context"
	"fmt"
	"io"

	"cloud.google.com/go/pubsub"
)

func createTopicWithConfluentCloudIngestion(w io.Writer, projectID, topicID, bootstrapServer, clusterID, confluentTopic, poolID, gcpSA string) error {
	// projectID := "my-project-id"
	// topicID := "my-topic"

	// // Confluent Cloud ingestion settings.
	// bootstrapServer := "bootstrap-server"
	// clusterID := "cluster-id"
	// confluentTopic := "confluent-topic"
	// poolID := "identity-pool-id"
	// gcpSA := "gcp-service-account"

	ctx := context.Background()
	client, err := pubsub.NewClient(ctx, projectID)
	if err != nil {
		return fmt.Errorf("pubsub.NewClient: %w", err)
	}
	defer client.Close()

	cfg := &pubsub.TopicConfig{
		IngestionDataSourceSettings: &pubsub.IngestionDataSourceSettings{
			Source: &pubsub.IngestionDataSourceConfluentCloud{
				BootstrapServer:   bootstrapServer,
				ClusterID:         clusterID,
				Topic:             confluentTopic,
				IdentityPoolID:    poolID,
				GCPServiceAccount: gcpSA,
			},
		},
	}
	t, err := client.CreateTopicWithConfig(ctx, topicID, cfg)
	if err != nil {
		return fmt.Errorf("CreateTopic: %w", err)
	}
	fmt.Fprintf(w, "Created topic with Confluent Cloud ingestion: %v\n", t)
	return nil
}

Java

Avant d'essayer cet exemple, suivez les instructions d'installation dans le langage Java qui se trouvent sur la page Démarrage rapide : utiliser des bibliothèques clientes. Pour en savoir plus, consultez la documentation de référence sur l'API Pub/Sub pour Java.

import com.google.cloud.pubsub.v1.TopicAdminClient;
import com.google.pubsub.v1.IngestionDataSourceSettings;
import com.google.pubsub.v1.Topic;
import com.google.pubsub.v1.TopicName;
import java.io.IOException;

public class CreateTopicWithConfluentCloudIngestionExample {
  public static void main(String... args) throws Exception {
    // TODO(developer): Replace these variables before running the sample.
    String projectId = "your-project-id";
    String topicId = "your-topic-id";
    // Confluent Cloud ingestion settings.
    String bootstrapServer = "bootstrap-server";
    String clusterId = "cluster-id";
    String confluentTopic = "confluent-topic";
    String identityPoolId = "identity-pool-id";
    String gcpServiceAccount = "gcp-service-account";

    createTopicWithConfluentCloudIngestionExample(
        projectId,
        topicId,
        bootstrapServer,
        clusterId,
        confluentTopic,
        identityPoolId,
        gcpServiceAccount);
  }

  public static void createTopicWithConfluentCloudIngestionExample(
      String projectId,
      String topicId,
      String bootstrapServer,
      String clusterId,
      String confluentTopic,
      String identityPoolId,
      String gcpServiceAccount)
      throws IOException {
    try (TopicAdminClient topicAdminClient = TopicAdminClient.create()) {
      TopicName topicName = TopicName.of(projectId, topicId);

      IngestionDataSourceSettings.ConfluentCloud confluentCloud =
          IngestionDataSourceSettings.ConfluentCloud.newBuilder()
              .setBootstrapServer(bootstrapServer)
              .setClusterId(clusterId)
              .setTopic(confluentTopic)
              .setIdentityPoolId(identityPoolId)
              .setGcpServiceAccount(gcpServiceAccount)
              .build();
      IngestionDataSourceSettings ingestionDataSourceSettings =
          IngestionDataSourceSettings.newBuilder().setConfluentCloud(confluentCloud).build();

      Topic topic =
          topicAdminClient.createTopic(
              Topic.newBuilder()
                  .setName(topicName.toString())
                  .setIngestionDataSourceSettings(ingestionDataSourceSettings)
                  .build());

      System.out.println(
          "Created topic with Confluent Cloud ingestion settings: " + topic.getAllFields());
    }
  }
}

Node.js

Avant d'essayer cet exemple, suivez les instructions d'installation dans le langage Node.js qui se trouvent sur la page Démarrage rapide : utiliser des bibliothèques clientes. Pour en savoir plus, consultez la documentation de référence sur l'API Pub/Sub pour Node.js.

/**
 * TODO(developer): Uncomment these variables before running the sample.
 */
// const topicNameOrId = 'YOUR_TOPIC_NAME_OR_ID';
// const bootstrapServer = 'url:port';
// const clusterId = 'YOUR_CLUSTER_ID';
// const confluentTopic = 'YOUR_CONFLUENT_TOPIC';
// const identityPoolId = 'pool-ID';
// const gcpServiceAccount = 'ingestion-account@...';

// Imports the Google Cloud client library
const {PubSub} = require('@google-cloud/pubsub');

// Creates a client; cache this for further use
const pubSubClient = new PubSub();

async function createTopicWithConfluentCloudIngestion(
  topicNameOrId,
  bootstrapServer,
  clusterId,
  confluentTopic,
  identityPoolId,
  gcpServiceAccount,
) {
  // Creates a new topic with Confluent Cloud ingestion.
  await pubSubClient.createTopic({
    name: topicNameOrId,
    ingestionDataSourceSettings: {
      confluentCloud: {
        bootstrapServer,
        clusterId,
        topic: confluentTopic,
        identityPoolId,
        gcpServiceAccount,
      },
    },
  });
  console.log(`Topic ${topicNameOrId} created with Confluent Cloud ingestion.`);
}

Node.ts

Avant d'essayer cet exemple, suivez les instructions d'installation dans le langage Node.js qui se trouvent sur la page Démarrage rapide : utiliser des bibliothèques clientes. Pour en savoir plus, consultez la documentation de référence sur l'API Pub/Sub pour Node.js.

/**
 * TODO(developer): Uncomment these variables before running the sample.
 */
// const topicNameOrId = 'YOUR_TOPIC_NAME_OR_ID';
// const bootstrapServer = 'url:port';
// const clusterId = 'YOUR_CLUSTER_ID';
// const confluentTopic = 'YOUR_CONFLUENT_TOPIC';
// const identityPoolId = 'pool-ID';
// const gcpServiceAccount = 'ingestion-account@...';

// Imports the Google Cloud client library
import {PubSub} from '@google-cloud/pubsub';

// Creates a client; cache this for further use
const pubSubClient = new PubSub();

async function createTopicWithConfluentCloudIngestion(
  topicNameOrId: string,
  bootstrapServer: string,
  clusterId: string,
  confluentTopic: string,
  identityPoolId: string,
  gcpServiceAccount: string,
) {
  // Creates a new topic with Confluent Cloud ingestion.
  await pubSubClient.createTopic({
    name: topicNameOrId,
    ingestionDataSourceSettings: {
      confluentCloud: {
        bootstrapServer,
        clusterId,
        topic: confluentTopic,
        identityPoolId,
        gcpServiceAccount,
      },
    },
  });
  console.log(`Topic ${topicNameOrId} created with Confluent Cloud ingestion.`);
}

Python

Avant d'essayer cet exemple, suivez les instructions d'installation dans le langage Python qui se trouvent sur la page Démarrage rapide : utiliser des bibliothèques clientes. Pour en savoir plus, consultez la documentation de référence sur l'API Pub/Sub pour Python.

from google.cloud import pubsub_v1
from google.pubsub_v1.types import Topic
from google.pubsub_v1.types import IngestionDataSourceSettings

# TODO(developer)
# project_id = "your-project-id"
# topic_id = "your-topic-id"
# bootstrap_server = "your-bootstrap-server"
# cluster_id = "your-cluster-id"
# confluent_topic = "your-confluent-topic"
# identity_pool_id = "your-identity-pool-id"
# gcp_service_account = "your-gcp-service-account"

publisher = pubsub_v1.PublisherClient()
topic_path = publisher.topic_path(project_id, topic_id)

request = Topic(
    name=topic_path,
    ingestion_data_source_settings=IngestionDataSourceSettings(
        confluent_cloud=IngestionDataSourceSettings.ConfluentCloud(
            bootstrap_server=bootstrap_server,
            cluster_id=cluster_id,
            topic=confluent_topic,
            identity_pool_id=identity_pool_id,
            gcp_service_account=gcp_service_account,
        )
    ),
)

topic = publisher.create_topic(request=request)

print(f"Created topic: {topic.name} with Confluent Cloud Ingestion Settings")

Si vous rencontrez des problèmes, consultez Dépanner un sujet d'importation Confluent Cloud.

Modifier un sujet d'importation de hubs Confluent Cloud

Pour modifier les paramètres de la source de données d'ingestion d'un sujet d'importation Confluent Cloud, procédez comme suit :

Console

1. Dans la console Trusted Cloud , accédez à la page Sujets.

   Accéder aux sujets

2. Cliquez sur le sujet d'importation Confluent Cloud.

3. Sur la page des détails de la rubrique, cliquez sur Modifier.

4. Modifiez les champs que vous souhaitez modifier.

5. Cliquez sur Mettre à jour.

gcloud

1. In the Trusted Cloud console, activate Cloud Shell.

   Activate Cloud Shell

   At the bottom of the Trusted Cloud console, a Cloud Shell session starts and displays a command-line prompt. Cloud Shell is a shell environment with the Google Cloud CLI already installed and with values already set for your current project. It can take a few seconds for the session to initialize.

   Pour éviter de perdre vos paramètres pour le thème d'importation, assurez-vous de tous les inclure chaque fois que vous mettez à jour le thème. Si vous oubliez quelque chose, Pub/Sub rétablit la valeur par défaut d'origine du paramètre.

2. Exécutez la commande gcloud pubsub topics update avec tous les indicateurs mentionnés dans l'exemple suivant :

   gcloud pubsub topics update TOPIC_ID \
      --confluent-cloud-ingestion-bootstrap-server CONFLUENT_BOOTSTRAP_SERVER \
      --confluent-cloud-ingestion-cluster-id CONFLUENT_CLUSTER_ID \
      --confluent-cloud-ingestion-topic CONFLUENT_TOPIC \
      --confluent-cloud-ingestion-identity-pool-id CONFLUENT_IDENTITY_POOL_ID \
      --confluent-cloud-ingestion-service-account PUBSUB_SERVICE_ACCOUNT

   Remplacez les éléments suivants :

   • TOPIC_ID : nom ou ID de votre sujet Pub/Sub.
   • CONFLUENT_BOOTSTRAP_SERVER : serveur d'amorçage de votre cluster contenant le sujet Kafka que vous ingérez dans Pub/Sub. Le format est le suivant : hostname:port.
   • CONFLUENT_CLUSTER_ID : ID de votre cluster contenant le sujet Kafka que vous ingérez dans Pub/Sub
   • CONFLUENT_TOPIC : nom du sujet Kafka que vous ingérez dans Pub/Sub.
   • CONFLUENT_IDENTITY_POOL_ID : ID du pool d'identités utilisé pour l'authentification avec Confluent Cloud.
   • CONFLUENT_IDENTITY_POOL_ID : compte de service que vous avez créé dans Créer un compte de service dans Google Cloud.

Quotas et limites

Le débit de publication pour les sujets d'importation est limité par le quota de publication du sujet. Pour en savoir plus, consultez la page Quotas et limites de Pub/Sub.

Étapes suivantes

• Surveiller un thème d'importation

• Choisissez le type d'abonnement pour votre sujet.

• Découvrez comment publier un message dans un sujet.

• Créez ou modifiez un sujet avec gcloud CLI, les API REST ou les bibliothèques clientes.

• Résolvez les problèmes liés à l'importation d'un sujet Confluent Cloud.

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.