Crea un tema de importación de Confluent Cloud

Un tema de importación de Confluent Cloud te permite transferir datos de forma continua desde Confluent Cloud como una fuente externa a Pub/Sub. Luego, puedes transmitir los datos a cualquiera de los destinos que admite Pub/Sub.

En este documento, se muestra cómo crear y administrar temas de importación de Confluent Cloud. Para crear un tema estándar, consulta Crea un tema estándar.

Para obtener más información sobre los temas de importación, consulta Acerca de los temas de importación.

Antes de comenzar

Roles y permisos requeridos

Para obtener los permisos que necesitas para crear y administrar temas de importación de Confluent Cloud, pídele a tu administrador que te otorgue el rol de IAM de editor de Pub/Sub (roles/pubsub.editor) en tu tema o proyecto. Para obtener más información sobre cómo otorgar roles, consulta Administra el acceso a proyectos, carpetas y organizaciones.

Este rol predefinido contiene los permisos necesarios para crear y administrar temas de importación de Confluent Cloud. Para ver los permisos exactos que son necesarios, expande la sección Permisos requeridos:

Permisos necesarios

Se requieren los siguientes permisos para crear y administrar temas de importación de Confluent Cloud:

  • Crea un tema de importación: pubsub.topics.create
  • Borrar un tema de importación: pubsub.topics.delete
  • Obtén un tema de importación: pubsub.topics.get
  • Enumera un tema de importación: pubsub.topics.list
  • Publicar en un tema de importación: pubsub.topics.publish
  • Actualiza un tema de importación: pubsub.topics.update
  • Obtén la política de IAM de un tema de importación: pubsub.topics.getIamPolicy
  • Configura la política de IAM para un tema de importación: pubsub.topics.setIamPolicy

También puedes obtener estos permisos con roles personalizados o con otros roles predefinidos.

Puedes configurar el control de acceso a nivel del proyecto y a nivel de los recursos individuales.

Configura la identidad federada para acceder a Confluent Cloud

La federación de Workload Identity permite que los servicios de Google Cloud accedan a las cargas de trabajo que se ejecutan fuera de Google Cloud. Con la federación de identidades, no es necesario que mantengas ni pases credenciales a Google Cloud para acceder a tus recursos en otras nubes. En su lugar, puedes usar las identidades de las cargas de trabajo para autenticarte en Google Cloud y acceder a los recursos.

Crea una cuenta de servicio en Google Cloud

Este paso es opcional. Si ya tienes una cuenta de servicio, puedes usarla en este procedimiento en lugar de crear una nueva. Si usas una cuenta de servicio existente, ve a Registra el ID único de la cuenta de servicio para ver el paso siguiente.

En el caso de los temas de importación de Confluent Cloud, Pub/Sub usa la cuenta de servicio como la identidad para acceder a los recursos de Confluent Cloud.

Para obtener más información sobre cómo crear una cuenta de servicio, incluidos los requisitos previos, los roles y los permisos necesarios, y los lineamientos de nombres, consulta Cómo crear cuentas de servicio. Después de crear una cuenta de servicio, es posible que debas esperar 60 segundos o más antes de usarla. Este comportamiento sucede porque las operaciones de lectura tienen coherencia eventual. Es posible que la nueva cuenta de servicio tarde un poco más en ser visible.

Registra el ID único de la cuenta de servicio

Necesitas un ID único de la cuenta de servicio para configurar el proveedor de identidad y el grupo en la consola de Confluent Cloud.

  1. En la consola de Google Cloud, ve a la página de detalles de la cuenta de servicio.

    Ir a la cuenta de servicio

  2. Haz clic en la cuenta de servicio que acabas de crear o en la que planeas usar.

  3. En la página Detalles de la cuenta de servicio, registra el número de ID único.

    Necesitas el ID como parte del flujo de trabajo para configurar el proveedor y el grupo de identidad en la consola de Confluent Cloud.

Agrega el rol de creador de tokens de cuenta de servicio a la cuenta de servicio de Pub/Sub

El rol de creador de tokens de cuenta de servicio (roles/iam.serviceAccountTokenCreator) permite que las principales creen credenciales de corta duración para una cuenta de servicio. Estos tokens o credenciales se usan para suplantar la identidad de la cuenta de servicio.

Para obtener más información sobre el uso de la identidad de cuentas de servicio, consulta Uso de la identidad de cuentas de servicio.

También puedes agregar el rol de publicador de Pub/Sub (roles/pubsub.publisher) durante este procedimiento. Para obtener más información sobre el rol y por qué lo agregas, consulta Cómo agregar el rol de publicador de Pub/Sub a la cuenta de servicio de Pub/Sub.

  1. En la consola de Google Cloud, ve a la página IAM.

    Ir a IAM

  2. Haz clic en la casilla de verificación Incluir asignaciones de roles proporcionadas por Google.

  3. Busca la cuenta de servicio que tiene el formato service-{PROJECT_NUMBER}@gcp-sa-pubsub.iam.gserviceaccount.com.

  4. En esta cuenta de servicio, haz clic en el botón Editar principal.

  5. Si es necesario, haz clic en Agregar otro rol.

  6. Busca y haz clic en el rol de creador de tokens de cuenta de servicio (roles/iam.serviceAccountTokenCreator).

  7. Haz clic en Guardar.

Crea un proveedor de identidad en Confluent Cloud

Para autenticarse en Confluent Cloud, la cuenta de servicio de Google Cloud necesita un grupo de identidades. Primero, debes crear un proveedor de identidad en Confluent Cloud.

Para obtener más información sobre cómo crear un proveedor de identidad en Confluent Cloud, consulta la página Cómo agregar un proveedor de identidad de OAuth/OIDC.

  1. Accede a la consola de Confluent Cloud.

  2. En el menú, haz clic en Cuentas y acceso.

  3. Haz clic en Identidades de carga de trabajo.

  4. Haz clic en Agregar proveedor.

  5. Haz clic en OAuth/OIDC y, luego, en Siguiente.

  6. Haz clic en Otro proveedor de OIDC y, luego, en Siguiente.

  7. Proporciona un nombre y una descripción del propósito del proveedor de identidad.

  8. Haz clic en Mostrar configuración avanzada.

  9. En el campo URI del emisor, ingresa https://accounts.google.com.

  10. En el campo JWKS URI, ingresa https://www.googleapis.com/oauth2/v3/certs.

  11. Haz clic en Validar y guardar.

Crea un grupo de identidades y otorga los roles adecuados en Confluent Cloud

Debes crear un grupo de identidades en tu perfil de identidad y otorgar los roles necesarios para permitir que la cuenta de servicio de Pub/Sub autentique y lea los temas de Kafka de Confluent Cloud.

Asegúrate de que tu clúster se haya creado en Confluent Cloud antes de continuar con la creación de un grupo de identidades.

Para obtener más información sobre cómo crear un grupo de identidades, consulta la página Cómo usar grupos de identidades con tu proveedor de identidad de OAuth/OIDC.

  1. Accede a la consola de Confluent Cloud.

  2. En el menú, haz clic en Cuentas y acceso.

  3. Haz clic en Identidades de carga de trabajo.

  4. Haz clic en el proveedor de identidad que creaste en Crea un proveedor de identidad en Confluent Cloud.

  5. Haz clic en Agregar grupo.

  6. Proporciona un nombre y una descripción para tu grupo de identidad.

  7. Establece el derecho de identidad en claims.

  8. En Establecer filtros, haz clic en la pestaña Configuración avanzada. Ingresa este código:

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

    Reemplaza <SERVICE_ACCOUNT_UNIQUE_ID> por el ID único de tu cuenta de servicio que se encuentra en Cómo registrar el ID único de la cuenta de servicio.

  9. Haz clic en Siguiente.

  10. Haz clic en Agregar permiso nuevo. Luego, haz clic en Siguiente.

  11. En el clúster correspondiente, haz clic en Agregar asignación de roles.

  12. Haz clic en el rol Operador y, luego, en Agregar.

    Este rol otorga Pub/Sub. Acceso de la cuenta de servicio al clúster que contiene el tema de Confluent Kafka que deseas transferir a Pub/Sub

  13. En el clúster, haz clic en Temas. Luego, haz clic en Agregar asignación de rol.

  14. Selecciona el rol DeveloperRead.

  15. Haz clic en la opción adecuada y especifica el tema o el prefijo. Por ejemplo, Tema específico, Regla de prefijo o Todos los temas.

  16. Haz clic en Agregar.

  17. Haz clic en Siguiente.

  18. Haz clic en Validar y guardar.

Agrega el rol de publicador de Pub/Sub al principal de Pub/Sub

Para habilitar la publicación, debes asignar un rol de publicador a la cuenta de servicio de Pub/Sub para que Pub/Sub pueda publicar en el tema de importación de Confluent Cloud.

Habilita la publicación desde todos los temas

Usa este método si no creaste ningún tema de importación de Confluent Cloud.

  1. En la consola de Google Cloud, ve a la página IAM.

    Ir a IAM

  2. Haz clic en la casilla de verificación Incluir asignaciones de roles proporcionadas por Google.

  3. Busca la cuenta de servicio que tiene el formato service-{PROJECT_NUMBER}@gcp-sa-pubsub.iam.gserviceaccount.com.

  4. En esta cuenta de servicio, haz clic en el botón Editar principal.

  5. Si es necesario, haz clic en Agregar otro rol.

  6. Busca y haz clic en el rol de publicador de Pub/Sub (roles/pubsub.publisher).

  7. Haz clic en Guardar.

Habilita la publicación desde un solo tema

Usa este método solo si ya existe el tema de importación de Confluent Cloud.

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

    Activate Cloud Shell

    At the bottom of the Google 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. Ejecuta el comando gcloud pubsub topics add-iam-policy-binding:

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

    Reemplaza lo siguiente:

    • TOPIC_ID: Es el ID del tema de importación de Confluent Cloud.

    • PROJECT_NUMBER: Número del proyecto Para ver el número del proyecto, consulta Identifica proyectos.

Agrega el rol de usuario de la cuenta de servicio a la cuenta de servicio

El rol de usuario de cuenta de servicio (roles/iam.serviceAccountUser) incluye el permiso iam.serviceAccounts.actAs que permite que una principal adjunte una cuenta de servicio a la configuración de transferencia del tema de importación de Confluent Cloud y use esa cuenta de servicio para la identidad federada.

  1. En la consola de Google Cloud, ve a la página IAM.

    Ir a IAM

  2. En el principal que emite las llamadas para crear o actualizar temas, haz clic en el botón Editar principal.

  3. Si es necesario, haz clic en Agregar otro rol.

  4. Busca y haz clic en el Rol de usuario de cuenta de servicio (roles/iam.serviceAccountUser).

  5. Haz clic en Guardar.

Usa temas de importación de Confluent Cloud

Puedes crear un tema de importación nuevo o editar uno existente.

Consideraciones

  • Crear el tema y la suscripción por separado, incluso si se hace de forma rápida, puede provocar la pérdida de datos. Hay un período breve en el que el tema existe sin una suscripción. Si se envían datos al tema durante este tiempo, se pierden. Si primero creas el tema, creas la suscripción y, luego, lo conviertes en un tema de importación, garantizas que no se pierdan mensajes durante el proceso de importación.

  • Si necesitas volver a crear el tema de Kafka de un tema de importación existente con el mismo nombre, no puedes solo borrar el tema de Kafka y volver a crearlo. Esta acción puede invalidar la administración de compensaciones de Pub/Sub, lo que puede provocar la pérdida de datos. Para mitigar este problema, sigue estos pasos:

    • Borra el tema de importación de Pub/Sub.
    • Borra el tema de Kafka.
    • Crea el tema de Kafka.
    • Crea el tema de importación de Pub/Sub.

  • Los datos de un tema de Kafka de Confluent Cloud siempre se leen desde el desplazamiento más antiguo.

Crea un tema de importación de Confluent Cloud

Para obtener más información sobre las propiedades asociadas con un tema, consulta Propiedades de un tema.

Asegúrate de haber completado los siguientes procedimientos:

Para crear un tema de importación de Confluent Cloud, sigue estos pasos:

Console

  1. En la consola de Google Cloud, ve a la página Temas.

    Ir a temas

  2. Haz clic en Crear un tema.

  3. En el campo ID de tema, ingresa un ID para tu tema de importación.

    Para obtener más información sobre cómo asignar nombres a los temas, consulta los lineamientos de nombres.

  4. Selecciona Agregar una suscripción predeterminada.

  5. Selecciona Habilitar transferencia.

  6. En Fuente de transferencia, selecciona Confluent Cloud.

  7. Ingresa los siguientes detalles:

    1. Servidor de arranque: Es el servidor de arranque de tu clúster que contiene el tema de Kafka que transfieres a Pub/Sub. El formato es el siguiente: hostname:port.

    2. ID del clúster: Es el ID de tu clúster que contiene el tema de Kafka que transfieres a Pub/Sub.

    3. Tema: Es el nombre del tema de Kafka que transfieres a Pub/Sub.

    4. ID del grupo de identidades: Es el ID del grupo de identidades que se usa para autenticarse con Confluent Cloud.

    5. Cuenta de servicio: Es la cuenta de servicio que creaste en Crea una cuenta de servicio en Google Cloud.

  8. Haz clic en Crear un tema.

gcloud

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

    Activate Cloud Shell

    At the bottom of the Google 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. Ejecuta el comando 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

    Reemplaza lo siguiente:

    • TOPIC_ID: El nombre o el ID de tu tema de Pub/Sub.
    • CONFLUENT_BOOTSTRAP_SERVER: Es el servidor de arranque de tu clúster que contiene el tema de Kafka que transfieres a Pub/Sub. El formato es el siguiente: hostname:port.
    • CONFLUENT_CLUSTER_ID: Es el ID de tu clúster que contiene el tema de Kafka que transfieres a Pub/Sub.
    • CONFLUENT_TOPIC: Es el nombre del tema de Kafka que transfieres a Pub/Sub.
    • CONFLUENT_IDENTITY_POOL_ID: El ID del grupo de identidades que se usa para autenticarse con Confluent Cloud.
    • PUBSUB_SERVICE_ACCOUNT: Es la cuenta de servicio que creaste en Crea una cuenta de servicio en Google Cloud.

Go

Antes de probar esta muestra, sigue las instrucciones de configuración de Go que encontrarás en la guía de inicio rápido de Pub/Sub con bibliotecas cliente. Si quieres obtener más información, consulta la documentación de referencia de la API de Go de Pub/Sub.

Para autenticarte en Pub/Sub, configura las credenciales predeterminadas de la aplicación. Si deseas obtener más información, consulta Configura la autenticación para un entorno de desarrollo local. 

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

Antes de probar esta muestra, sigue las instrucciones de configuración de Java que encontrarás en la guía de inicio rápido de Pub/Sub con bibliotecas cliente. Si quieres obtener más información, consulta la documentación de referencia de la API de Java de Pub/Sub.

Para autenticarte en Pub/Sub, configura las credenciales predeterminadas de la aplicación. Si deseas obtener más información, consulta Configura la autenticación para un entorno de desarrollo local. 


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

Antes de probar esta muestra, sigue las instrucciones de configuración de Node.js que encontrarás en la guía de inicio rápido de Pub/Sub con bibliotecas cliente. Si quieres obtener más información, consulta la documentación de referencia de la API de Node.js de Pub/Sub.

Para autenticarte en Pub/Sub, configura las credenciales predeterminadas de la aplicación. Si deseas obtener más información, consulta Configura la autenticación para un entorno de desarrollo local. 

/**
 * 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.`);
}

Python

Antes de probar esta muestra, sigue las instrucciones de configuración de Python que encontrarás en la guía de inicio rápido de Pub/Sub con bibliotecas cliente. Si quieres obtener más información, consulta la documentación de referencia de la API de Python de Pub/Sub.

Para autenticarte en Pub/Sub, configura las credenciales predeterminadas de la aplicación. Si deseas obtener más información, consulta Configura la autenticación para un entorno de desarrollo local. 

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")

C++

Antes de probar esta muestra, sigue las instrucciones de configuración de C++ que encontrarás en la guía de inicio rápido de Pub/Sub con bibliotecas cliente. Si quieres obtener más información, consulta la documentación de referencia de la API de C++ de Pub/Sub.

Para autenticarte en Pub/Sub, configura las credenciales predeterminadas de la aplicación. Si deseas obtener más información, consulta Configura la autenticación para un entorno de desarrollo local. 

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";
}

Node.js (TypeScript)

Antes de probar esta muestra, sigue las instrucciones de configuración de Node.js en la guía de inicio rápido de Pub/Sub mediante bibliotecas cliente. Si quieres obtener más información, consulta la documentación de referencia de la API de Pub/Sub para Node.js.

Para autenticarte en Pub/Sub, configura las credenciales predeterminadas de la aplicación. Si deseas obtener más información, consulta Configura la autenticación para un entorno de desarrollo local. 

/**
 * 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.`);
}

Si tienes problemas, consulta Cómo solucionar problemas relacionados con un tema de importación de Confluent Cloud.

Cómo editar un tema de importación de Confluent Cloud Hubs

Para editar la configuración de la fuente de datos de transferencia de un tema de importación de Confluent Cloud, sigue estos pasos:

Console

  1. En la consola de Google Cloud, ve a la página Temas.

    Ir a temas

  2. Haz clic en el tema de importación de Confluent Cloud.

  3. En la página de detalles del tema, haz clic en Editar.

  4. Actualiza los campos que deseas cambiar.

  5. Haz clic en Actualizar.

gcloud

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

    Activate Cloud Shell

    At the bottom of the Google 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.

    Para evitar perder la configuración del tema de importación, asegúrate de incluirlos todos cada vez que lo actualices. Si omites algo, Pub/Sub restablece la configuración a su valor predeterminado original.

  2. Ejecuta el comando gcloud pubsub topics update con todas las marcas mencionadas en el siguiente ejemplo:

    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

    Reemplaza lo siguiente:

    • TOPIC_ID: El nombre o el ID de tu tema de Pub/Sub.
    • CONFLUENT_BOOTSTRAP_SERVER: Es el servidor de arranque de tu clúster que contiene el tema de Kafka que transfieres a Pub/Sub. El formato es el siguiente: hostname:port.
    • CONFLUENT_CLUSTER_ID: Es el ID de tu clúster que contiene el tema de Kafka que transfieres a Pub/Sub.
    • CONFLUENT_TOPIC: Es el nombre del tema de Kafka que transfieres a Pub/Sub.
    • CONFLUENT_IDENTITY_POOL_ID: El ID del grupo de identidades que se usa para autenticarse con Confluent Cloud.
    • CONFLUENT_IDENTITY_POOL_ID: Es la cuenta de servicio que creaste en Crea una cuenta de servicio en Google Cloud.

Cuotas y límites

La capacidad de procesamiento del publicador para los temas de importación está limitada por la cuota de publicación del tema. Para obtener más información, consulta Cuotas y límites de Pub/Sub.

¿Qué sigue?

Apache Kafka® es una marca registrada de The Apache Software Foundation o sus afiliados de Estados Unidos y otros países.