Crea un clúster de Connect

Un clúster de Connect proporciona un entorno para los conectores que ayuda a mover datos de las implementaciones de Kafka existentes a un clúster de Google Cloud Managed Service para Apache Kafka o mover datos de un clúster de Managed Service para Apache Kafka a otro servicio de Google Cloud o a otro clúster de Kafka. El clúster de Kafka secundario puede ser otro clúster de Google Cloud Managed Service para Apache Kafka, uno autoadministrado o uno local.

Antes de comenzar

Asegúrate de haber creado un clúster de Managed Service para Apache Kafka. Necesitas el nombre del clúster de Managed Service para Apache Kafka al que se adjuntará el clúster de Connect.

Cada clúster de Connect está asociado con un clúster de Managed Service para Apache Kafka. Este clúster almacena el estado de los conectores que se ejecutan en el clúster de Connect.

Roles y permisos necesarios para crear un clúster de Connect

Para obtener los permisos que necesitas para crear un clúster de Connect, pídele a tu administrador que te otorgue el rol de IAM de editor de clústeres de Kafka Connect administrado (roles/managedkafka.connectClusterEditor) en tu 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 un clúster de Connect. Para ver los permisos exactos que son necesarios, expande la sección Permisos requeridos:

Permisos necesarios

Se requieren los siguientes permisos para crear un clúster de Connect:

  • Otorga el permiso para crear un clúster de Connect en la ubicación especificada: managedkafka.connectClusters.create

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

Principales de ACL obligatorias

De forma predeterminada, los clústeres de Managed Service para Apache Kafka permiten que el clúster de Connect acceda a los recursos si no se configuran ACL. Para ello, se debe configurar allow.everyone.if.no.acl.found como true, que es el parámetro de configuración predeterminado.

Sin embargo, si el clúster de Managed Service para Apache Kafka tiene ACL configuradas, el clúster de Connect no tiene automáticamente los permisos de lectura y escritura en los recursos. Debes otorgarlos de forma manual.

La cuenta de servicio del clúster de Connect que se usa como principal en las LCA tiene el siguiente formato: User:service-{consumer project number}@gcp-sa-managedkafka.iam.gserviceaccount.com.

Si configuraste ACL en tu clúster de Kafka, otorga al clúster de Connect permisos de lectura y escritura para los temas y permisos de lectura para los grupos de consumidores con los siguientes comandos:

/bin/kafka-acls.sh \
    --bootstrap-server BOOTSTRAP_ADDR \
    --command-config PATH_TO_CLIENT_PROPERTIES \
    --add \
    --allow-principal User:service-{consumer project number}@gcp-sa-managedkafka.iam.gserviceaccount.com \
    --operation READ --operation WRITE --topic *

/bin/kafka-acls.sh \
    --bootstrap-server BOOTSTRAP_ADDR \
    --command-config PATH_TO_CLIENT_PROPERTIES \
    --add \
    --allow-principal User:service-{consumer project number}@gcp-sa-managedkafka.iam.gserviceaccount.com \
    --operation READ --group *

Para obtener más información sobre estos comandos, consulta Configura las ACL de Apache Kafka para el control de acceso detallado.

Crea un clúster de Connect en otro proyecto

Managed Service para Apache Kafka usa un agente de servicio para acceder a los recursos de Google Cloud . El agente de servicio se asocia con el proyecto en el que creas el clúster.

Si creas un clúster de Connect en un proyecto diferente del clúster de Managed Service para Apache Kafka, el clúster de Connect y el clúster de Kafka usarán los agentes de servicio asociados con sus respectivos proyectos. En ese caso, el agente de servicio del clúster de Connect necesita permiso para acceder a los recursos de Google Cloud en el proyecto del clúster de Kafka.

Para otorgar los permisos necesarios, asigna el rol de agente de servicio de Kafka administrado al agente de servicio del clúster de Connect en el proyecto del clúster de Kafka. Por ejemplo, si creas un clúster de Kafka en el proyecto kafka-project y un clúster de Connect en el proyecto connect-project, otorga el rol de agente de servicio de Kafka administrado en kafka-project al agente de servicio asociado con connect-project.

La dirección de correo electrónico del agente de servicio tiene el siguiente formato: service-PROJECT_NUMBER@gcp-sa-managedkafka.iam.gserviceaccount.com, en el que PROJECT_NUMBER es el número del proyecto. Para obtener más información sobre cómo otorgar el rol, consulta Crea y otorga roles a los agentes de servicio.

Propiedades de un clúster de Connect

En esta sección, se describen las propiedades de un clúster de Connect.

Nombre del clúster de Connect

Es el nombre del clúster de Connect que estás creando. Si necesitas ayuda para asignarle un nombre a un clúster de Connect, consulta los Lineamientos para asignarles nombres a los recursos de Managed Service para Apache Kafka. Los nombres de clústeres son inmutables.

Clúster principal de Kafka

Es el clúster de Managed Service para Apache Kafka asociado con tu clúster de Connect. Este clúster asociado (clúster principal) almacena el estado de los conectores que se ejecutan en el clúster de Connect. Por lo general, el clúster principal de Managed Service para Apache Kafka también funciona como el destino de todos los conectores de origen y la entrada de todos los conectores receptores que se ejecutan en el clúster de Connect.

Un solo clúster de Managed Service para Apache Kafka puede tener varios clústeres de Connect. Si eliges un clúster de Managed Service para Apache Kafka en un proyecto diferente, asegúrate de que se configuren los permisos adecuados.

No puedes actualizar a un clúster de Kafka diferente después de crear el clúster de Connect.

Beneficios de la coubicación de regiones para la latencia y los costos de red

La ubicación conjunta de tus clústeres de Managed Service para Apache Kafka y Connect en la misma región reduce la latencia y los costos de red. Por ejemplo, supongamos que tu clúster de Managed Service para Apache Kafka está en region-a y usas un conector receptor para escribir datos de este clúster de Managed Service para Apache Kafka (origen) en una tabla de BigQuery (receptor) que también está en region-a. Si implementas tu clúster de Connect en region-a, esta opción de implementación minimiza la latencia de la operación de escritura de BigQuery y elimina los costos de transferencia de red entre regiones entre el clúster de Managed Service para Apache Kafka y el clúster de Connect.

Consideraciones de latencia y costo de varios sistemas

Kafka Connect usa conectores para mover datos entre sistemas. Un lado del conector siempre interactúa con un clúster de Managed Service para Apache Kafka. Un solo clúster de Kafka Connect puede ejecutar varios conectores, cada uno de los cuales actúa como una fuente (extrae datos de un sistema) o un receptor (envía datos a un sistema).

Si bien un clúster de Connect en la misma región que el clúster de Managed Service para Apache Kafka se beneficia de una menor latencia de comunicación entre ellos, cada conector también interactúa con otro sistema, como una tabla de BigQuery o un clúster de Kafka diferente. Incluso si el clúster de Connect y el clúster de Managed Service para Apache Kafka se encuentran en la misma ubicación, ese otro sistema podría estar en una región diferente. Esto genera una mayor latencia y un costo más alto. La latencia general de la canalización depende de las ubicaciones de los tres sistemas: el clúster de Managed Service para Apache Kafka, el clúster de Connect y el sistema de origen o receptor.

Por ejemplo, si tu clúster de Managed Service para Apache Kafka está en region-a, tu clúster de Connect en region-b y usas un conector de Cloud Storage para un bucket en region-c, se te cobrarán dos saltos de red (de region-a a region-b y, luego, de region-b a region-c, o al revés, según la dirección del conector).

Considera cuidadosamente todas las regiones involucradas cuando planifiques la ubicación de tu clúster de Connect para optimizar la latencia y el costo.

Configuración de capacidad

La configuración de capacidad requiere que establezcas la cantidad de CPU virtuales y de memoria para cada CPU virtual de tu clúster de Connect. Puedes actualizar la capacidad de un clúster de Connect después de crearlo. Las siguientes son las propiedades de la configuración de capacidad:

  • CPU virtuales: Es la cantidad de CPU virtuales asignadas a un clúster de Connect. El valor mínimo es de 3 CPU virtuales.

  • Memoria: Es la cantidad de memoria que se asigna a cada CPU virtual. Debes aprovisionar entre 1 GiB y 8 GiB por CPU virtual. La cantidad de memoria se puede aumentar o disminuir dentro de estos límites después de crear el clúster.

    Por ejemplo, si creas un clúster con 6 CPU virtuales, la memoria mínima que puedes asignar al clúster es de 6 GiB (1 GiB por CPU virtual) y el máximo es de 48 GiB (8 GiB por CPU virtual).

La CPU virtual y la memoria asignadas a cada trabajador en un clúster de Connect tienen un impacto significativo en el rendimiento, la capacidad y el costo del clúster. Aquí tienes un desglose de cómo la CPU virtual y la memoria afectan a un clúster de Connect.

Recuento de CPU virtuales

  • Kafka Connect divide el trabajo de un conector en tareas. Cada tarea puede procesar datos en paralelo. Más CPUs virtuales implican que se pueden ejecutar más tareas de forma simultánea, lo que genera una mayor capacidad de procesamiento.

  • Más CPU virtuales aumentan los costos de tu clúster de Connect.

Memoria

  • Kafka Connect usa la memoria para almacenar datos en búfer a medida que fluyen entre los conectores y Managed Service para Apache Kafka. Una memoria más grande permite búferes más grandes. Una memoria grande puede mejorar la capacidad de procesamiento, en especial para los flujos de datos de gran volumen. Los conectores que manejan mensajes o registros muy grandes requieren memoria suficiente para procesarlos sin encontrarse con excepciones OutOfMemoryError.

  • Más memoria aumenta el costo de tu clúster de Connect.

  • Si usas una lógica de transformación pesada, necesitas más asignación de memoria.

Tu objetivo es elegir la configuración de capacidad adecuada para tu clúster de Connect. Para ello, debes comprender la capacidad de procesamiento que puede controlar tu clúster de Connect.

Subred de trabajadores (principal)

La subred de trabajadores, también conocida como subred principal, conecta tu red de VPC al clúster de Connect. Esta subred permite que los trabajadores del clúster lleguen a los extremos de las fuentes y los receptores en la red del consumidor, como los clústeres de Managed Service para Apache Kafka o los clústeres de Kafka autohospedados.

Estos son algunos requisitos para configurar la subred de trabajadores:

  • La subred de trabajadores es obligatoria.

  • La subred debe estar ubicada en la misma región que el clúster de Connect.

  • La subred debe estar en la misma VPC superior que una de las subredes conectadas de la lista del clúster principal de Kafka.

  • El rango CIDR de la subred debe tener un tamaño mínimo de /22 (1,024 direcciones).

A los trabajadores del clúster se les asignan direcciones IP en la subred de trabajadores con una interfaz de Private Service Connect. Los trabajadores pueden llegar a cualquier destino de red al que se pueda acceder desde la red de VPC de la subred, con los siguientes requisitos:

  • El extremo no debe estar en el rango CIDR 172.16.0.0/14. Este rango está reservado para el uso interno de Managed Service para Apache Kafka Connect.
  • Las reglas de firewall deben permitir el tráfico. Consulta Configura la seguridad para los adjuntos de red.
  • Para el tráfico de Internet, debes configurar una Cloud NAT. Por ejemplo, se requiere una Cloud NAT para que un conector de MirrorMaker replique datos de un clúster de Kafka al que se puede acceder a través de Internet.
  • Para acceder a los extremos de Private Service Connect que se encuentran en una VPC diferente de la VPC de la subred de trabajadores, debes asegurarte de usar una configuración de consumidor compatible (por ejemplo, NCC). Para obtener más información, consulta Información sobre el acceso a servicios publicados a través de extremos.

Dominios del DNS que se pueden resolver

Los dominios de DNS que se pueden resolver, también conocidos como nombres de dominio de DNS, permiten que las direcciones DNS de la red de VPC del consumidor estén disponibles para la VPC del usuario. Esto permite que el clúster de Connect resuelva los nombres de DNS en direcciones IP, lo que facilita la comunicación con otros servicios, incluidos otros clústeres de Kafka para los conectores de MirrorMaker.

En el caso de los dominios de DNS que se pueden resolver, puedes seleccionar un clúster de Managed Service para Apache Kafka. No es necesario que configures el nombre de dominio de DNS para el clúster principal de Managed Service para Apache Kafka, ya que su dirección de arranque se incluye automáticamente en la lista de dominios de DNS que se pueden resolver.

Sin embargo, también puedes especificar un dominio de DNS de forma manual, lo que es necesario si seleccionas un clúster de Kafka externo. El dominio DNS del clúster principal de Managed Service para Apache Kafka se incluye automáticamente. Otros clústeres de Kafka aún requieren configurar dominios de DNS.

Recursos de Secret Manager

Algunos conectores requieren datos sensibles, como contraseñas, como parte de su configuración. Para administrar este tipo de datos de forma segura, puedes almacenarlos en Secret Manager y otorgar al clúster de Connect acceso al Secret.

Para usar los secretos de Secret Manager con Kafka Connect, haz lo siguiente:

  1. Otorga los siguientes roles de IAM a la cuenta de servicio administrada de Kafka. Estos roles permiten que tu clúster de Connect vea los secretos y acceda a sus cargas útiles.

    • Visualizador de Secret Manager (roles/secretmanager.viewer)
    • Descriptor de acceso de secretos de Secret Manager (roles/secretmanager.secretAccessor)

  2. Crea un Secret que contenga los datos sensibles. Para obtener más información, consulta Crea un secreto.

  3. Cuando crees o actualices tu clúster de Connect, especifica los secretos a los que tiene acceso. Puedes especificar hasta 32 secretos por clúster de Connect.

Los Secrets se activan como archivos en los trabajadores del clúster. Los conectores tienen acceso de solo lectura a estos archivos. Cuando creas un conector, las propiedades de configuración del conector pueden hacer referencia a los secretos.

  • Para hacer referencia a la ruta de acceso a un archivo de Secret, usa el siguiente formato:

    /var/secrets/PROJECT_NAME-SECRET_NAME-SECRET_VERSION

    Ejemplo: ssl.truststore.location=/var/secrets/project1-truststore-1

  • Para usar el valor de un Secret como un valor de configuración (por ejemplo, una contraseña), usa el siguiente formato:

    ${directory:/var/secrets:PROJECT_NAME-SECRET_NAME-SECRET_VERSION}

    Ejemplo: password=${directory:/var/secrets:project1-database_password-3}

Reemplaza lo siguiente:

  • PROJECT_NAME: Es el nombre del proyecto de Google Cloud .
  • SECRET_NAME: Es el nombre del secreto.
  • SECRET_VERSION: Es la versión del Secret.

Etiquetas

Las etiquetas son pares clave-valor que te ayudan con la organización y la identificación. Te ayudan a organizar los clústeres de Connect. Puedes adjuntar una etiqueta a cada clúster de Connect y, luego, filtrar los recursos según las etiquetas. Algunos ejemplos de etiquetas son environment:prod y application:web-app.

Crea un clúster de Connect

Antes de crear un clúster, revisa la documentación sobre las propiedades del clúster de Connect.

La creación de un clúster de Connect tarda entre 20 y 30 minutos.

Console

  1. En la consola de Google Cloud , ve a la página Conectar clústeres.

    Ir a Conectar clústeres

  2. Haga clic en Crear.

  3. En el campo Nombre del clúster de Connect, ingresa un nombre para el clúster de Connect. Para obtener más información, consulta Lineamientos para asignarles nombres a los recursos de Managed Service para Apache Kafka.

  4. En la lista Clúster principal de Kafka, selecciona un clúster de Managed Service para Apache Kafka. Para obtener más información, consulta Clúster de Kafka principal.

  5. En la lista Región, selecciona una ubicación para el clúster de Connect. Para obtener más información sobre cómo seleccionar una ubicación, consulta Clúster principal de Kafka.

  6. En la sección Configuración de la capacidad, ingresa los valores para los siguientes campos o mantén los valores predeterminados.

    • En el campo vCPUs, ingresa la cantidad de CPU virtuales para el clúster.

    • En el campo Memoria, ingresa la cantidad de memoria por CPU, en GiB. El valor no puede superar los 8 GiB por CPU.

    Para obtener más información, consulta Configuración de la capacidad.

  7. En la sección Configuración de red, selecciona una red de VPC de la lista Red o deja el valor predeterminado. Esta lista se completa cuando seleccionas el clúster principal de Kafka.

  8. En la sección Subred de trabajador, selecciona una subred de la lista Subred o deja el valor predeterminado. Para obtener más información, consulta Subred de trabajador. El campo Ruta de URI de la subred se completa automáticamente cuando seleccionas la subred.

  9. Opcional: Agrega un dominio de DNS que se pueda resolver. El dominio del DNS del clúster principal de Kafka se agrega automáticamente como un dominio DNS que se puede resolver. Para especificar dominios de DNS adicionales, haz lo siguiente:

    1. Expande la sección Dominios del DNS que se pueden resolver.

    2. Haz clic en Agregar dominio de DNS.

    3. Para agregar el dominio de DNS de un clúster existente de Managed Service para Apache Kafka, selecciona el clúster de la lista Clúster de Kafka. De lo contrario, ingresa el dominio de DNS en el campo Dominio de DNS.

    4. Haz clic en Listo.

  10. Opcional: Para agregar recursos de Secret Manager, haz lo siguiente:

    1. Expande la sección Recursos de Secret Manager.

    2. Haz clic en Agregar recurso de Secret.

    3. En la lista Secreto, selecciona un secreto.

    4. En la lista Versión del Secret, selecciona una versión del Secret.

    5. Haz clic en Listo.

  11. Opcional: Agrega etiquetas para organizar tu proyecto. Para agregar una etiqueta, haz lo siguiente:

    1. Expande la sección Etiquetas.

    2. Haz clic en Agregar etiqueta.

    3. En el campo Clave, ingresa la clave de la etiqueta.

    4. En el campo Valor, ingresa el valor de la etiqueta.

  12. Haz clic en Crear.

gcloud

  1. En la consola de Google Cloud , activa Cloud Shell.

    Activa Cloud Shell

    En la parte inferior de la consola de Google Cloud , se inicia una sesión de Cloud Shell en la que se muestra una ventana de línea de comandos. Cloud Shell es un entorno de shell con Google Cloud CLI ya instalada y con valores ya establecidos para el proyecto actual. La sesión puede tardar unos segundos en inicializarse.

  2. Ejecuta el comando gcloud managed-kafka connect-clusters create:

    gcloud managed-kafka connect-clusters create CONNECT_CLUSTER_ID \
    --location=LOCATION \
    --cpu=CPU \
    --memory=MEMORY \
    --primary-subnet=WORKER_SUBNET \
    --kafka-cluster=KAFKA_CLUSTER \
    [--project=PROJECT_ID] \
    [--secret=SECRET] \
    [--dns-name=DNS_DOMAIN_NAME] \
    [--config-file=CONFIG_FILE] \
    [--labels=LABELS]
    [--async]

    Reemplaza lo siguiente:

    • CONNECT_CLUSTER_ID: Es el ID o el nombre del clúster de Connect. Si necesitas ayuda para asignarle un nombre a un clúster de Connect, consulta los Lineamientos para asignarles nombres a los recursos de Managed Service para Apache Kafka. El nombre de un clúster de Connect es inmutable.

    • LOCATION: Es la ubicación en la que creas el clúster de Connect. Debe ser una región de Google Cloudcompatible. No puedes cambiar la ubicación de un clúster de Connect después de crearlo. Para ver la lista de ubicaciones disponibles, consulta las ubicaciones de Managed Service para Apache Kafka. Para obtener más información sobre las recomendaciones de ubicación, consulta Clúster de Kafka principal.

    • CPU: Es la cantidad de CPU virtuales para el clúster de Connect. El valor mínimo es 3 CPUs virtuales. Consulta Recuento de CPU virtuales.

    • MEMORY: Es la cantidad de memoria para el clúster de Connect. Usa las unidades "MB", "MiB", "GB", "GiB", "TB" o "TiB". Por ejemplo, "3GiB". Debes aprovisionar entre 1 GiB y 8 GiB por CPU virtual. Consulta Memoria.

    • WORKER_SUBNET: Es la subred de trabajador para el clúster de Connect.

      El formato de la subred es projects/PROJECT_ID/regions/REGION/subnetworks/SUBNET_ID.

      La subred de trabajadores debe estar en la misma región que el clúster de Connect.

    • PROJECT_ID: Es el ID del proyecto de Google Cloud (opcional). Si no se proporciona, se usa el proyecto actual.

    • KAFKA_CLUSTER: Es el ID o el nombre completamente calificado del clúster principal de Managed Service para Apache Kafka asociado con el clúster de Connect. Consulta Clúster de Kafka. El formato del clúster de Kafka es projects/PROJECT_ID/locations/LOCATION/clusters/CLUSTER_ID.

      No puedes actualizar a un clúster de Kafka diferente después de crear el clúster de Connect.

    • SECRET: (Opcional) Son los secretos que se cargarán en los trabajadores. Se deben proporcionar las versiones exactas de los secretos de Secret Manager, ya que no se admiten alias. Se pueden cargar hasta 32 Secrets en un clúster. Formato: projects/PROJECT_ID/secrets/SECRET_NAME/versions/VERSION_ID

    • DNS_DOMAIN_NAME: Son los nombres de dominio de DNS de la subred que se harán visibles para el clúster de Connect (opcional). El clúster de Connect puede acceder a los recursos con nombres de dominio en lugar de depender de direcciones IP. Consulta Intercambio de tráfico de DNS.

    • LABELS: Son las etiquetas que se asociarán con el clúster (opcional). Para obtener más información sobre el formato de las etiquetas, consulta Etiquetas. Es la lista de pares KEY=VALUE de etiquetas que se agregarán. Las claves deben comenzar con un carácter en minúscula y contener solo guiones (-), guiones bajos (_), caracteres en minúscula y números. Los valores deben contener solo guiones (-), guiones bajos (_), caracteres en minúscula y números.

    • CONFIG_FILE: (Opcional) Es la ruta de acceso al archivo JSON o YAML que contiene la configuración que se anula de los valores predeterminados del clúster o del conector. Este archivo también admite JSON o YAML intercalados.

    • --async: (Opcional) Devuelve de inmediato, sin esperar a que se complete la operación en curso. Con la marca --async, puedes continuar con otras tareas mientras se crea el clúster en segundo plano. Si no usas la marca, el sistema espera a que se complete la operación antes de devolver una respuesta. Debes esperar hasta que el clúster se actualice por completo antes de poder continuar con otras tareas.

    Recibirás una respuesta similar a la siguiente:

    Create request issued for: [sample-connectcluster]
Check operation [projects/test-project/locations/us-east1/operations/operation-1753590328249-63ae19098cc06-64300a0a-06512d02] for status.

    Almacena el OPERATION_ID para hacer un seguimiento del progreso. Por ejemplo, el valor aquí es operation-1753590328249-63ae19098cc06-64300a0a-06512d02.

Terraform

Puedes usar un recurso de Terraform para crear un clúster de Connect.

resource "google_managed_kafka_connect_cluster" "default" {
  provider           = google-beta
  project            = data.google_project.default.project_id
  connect_cluster_id = "my-connect-cluster-id"
  location           = "us-central1"
  kafka_cluster      = google_managed_kafka_cluster.default.id
  capacity_config {
    vcpu_count   = 12
    memory_bytes = 12884901888 # 12 GiB
  }
  gcp_config {
    access_config {
      network_configs {
        primary_subnet = google_compute_subnetwork.default.id
      }
    }
  }
}

Si deseas obtener más información para aplicar o quitar una configuración de Terraform, consulta los comandos básicos de Terraform.

Go

Antes de probar esta muestra, sigue las instrucciones de configuración de Go en Instala las bibliotecas cliente. Para obtener más información, consulta la documentación de referencia de la API de Go de Managed Service para Apache Kafka.

Para autenticarte en Managed Service para Apache Kafka, configura las credenciales predeterminadas de la aplicación(ADC). Para obtener más información, consulta Configura ADC para un entorno de desarrollo local.

import (
	"context"
	"fmt"
	"io"

	"cloud.google.com/go/managedkafka/apiv1/managedkafkapb"
	"google.golang.org/api/option"

	managedkafka "cloud.google.com/go/managedkafka/apiv1"
)

func createConnectCluster(w io.Writer, projectID, region, clusterID, kafkaCluster string, opts ...option.ClientOption) error {
	// projectID := "my-project-id"
	// region := "us-central1"
	// clusterID := "my-connect-cluster"
	// kafkaCluster := "projects/my-project-id/locations/us-central1/clusters/my-kafka-cluster"
	ctx := context.Background()
	client, err := managedkafka.NewManagedKafkaConnectClient(ctx, opts...)
	if err != nil {
		return fmt.Errorf("managedkafka.NewManagedKafkaConnectClient got err: %w", err)
	}
	defer client.Close()

	locationPath := fmt.Sprintf("projects/%s/locations/%s", projectID, region)
	clusterPath := fmt.Sprintf("%s/connectClusters/%s", locationPath, clusterID)

	// Capacity configuration with 12 vCPU and 12 GiB RAM
	capacityConfig := &managedkafkapb.CapacityConfig{
		VcpuCount:   12,
		MemoryBytes: 12884901888, // 12 GiB in bytes
	}

	// Optionally, you can also specify accessible subnets and resolvable DNS
	// domains as part of your network configuration. For example:
	// networkConfigs := []*managedkafkapb.ConnectNetworkConfig{
	// 	{
	// 		PrimarySubnet:      primarySubnet,
	// 		AdditionalSubnets:  []string{"subnet-1", "subnet-2"},
	// 		DnsDomainNames:     []string{"domain-1", "domain-2"},
	// 	},
	// }

	connectCluster := &managedkafkapb.ConnectCluster{
		Name:           clusterPath,
		KafkaCluster:   kafkaCluster,
		CapacityConfig: capacityConfig,
	}

	req := &managedkafkapb.CreateConnectClusterRequest{
		Parent:           locationPath,
		ConnectClusterId: clusterID,
		ConnectCluster:   connectCluster,
	}
	op, err := client.CreateConnectCluster(ctx, req)
	if err != nil {
		return fmt.Errorf("client.CreateConnectCluster got err: %w", err)
	}
	// The duration of this operation can vary considerably, typically taking 5-15 minutes.
	resp, err := op.Wait(ctx)
	if err != nil {
		return fmt.Errorf("op.Wait got err: %w", err)
	}
	fmt.Fprintf(w, "Created connect cluster: %s\n", resp.Name)
	return nil
}

Java

Antes de probar esta muestra, sigue las instrucciones de configuración de Java en Instala las bibliotecas cliente. Para obtener más información, consulta la documentación de referencia de la API de Java de Managed Service para Apache Kafka.

Para autenticarte en Managed Service para Apache Kafka, configura las credenciales predeterminadas de la aplicación. Para obtener más información, consulta Configura ADC para un entorno de desarrollo local.


import com.google.api.gax.longrunning.OperationFuture;
import com.google.api.gax.longrunning.OperationSnapshot;
import com.google.api.gax.longrunning.OperationTimedPollAlgorithm;
import com.google.api.gax.retrying.RetrySettings;
import com.google.api.gax.retrying.RetryingFuture;
import com.google.api.gax.retrying.TimedRetryAlgorithm;
import com.google.cloud.managedkafka.v1.CapacityConfig;
import com.google.cloud.managedkafka.v1.ConnectAccessConfig;
import com.google.cloud.managedkafka.v1.ConnectCluster;
import com.google.cloud.managedkafka.v1.ConnectGcpConfig;
import com.google.cloud.managedkafka.v1.ConnectNetworkConfig;
import com.google.cloud.managedkafka.v1.CreateConnectClusterRequest;
import com.google.cloud.managedkafka.v1.LocationName;
import com.google.cloud.managedkafka.v1.ManagedKafkaConnectClient;
import com.google.cloud.managedkafka.v1.ManagedKafkaConnectSettings;
import com.google.cloud.managedkafka.v1.OperationMetadata;
import java.time.Duration;
import java.util.concurrent.ExecutionException;

public class CreateConnectCluster {

  public static void main(String[] args) throws Exception {
    // TODO(developer): Replace these variables before running the example.
    String projectId = "my-project-id";
    String region = "my-region"; // e.g. us-east1
    String clusterId = "my-connect-cluster";
    String subnet = "my-subnet"; // e.g. projects/my-project/regions/my-region/subnetworks/my-subnet
    String kafkaCluster = "my-kafka-cluster"; // The Kafka cluster to connect to
    int cpu = 12;
    long memoryBytes = 12884901888L; // 12 GiB
    createConnectCluster(projectId, region, clusterId, subnet, kafkaCluster, cpu, memoryBytes);
  }

  public static void createConnectCluster(
      String projectId,
      String region,
      String clusterId,
      String subnet,
      String kafkaCluster,
      int cpu,
      long memoryBytes)
      throws Exception {
    CapacityConfig capacityConfig = CapacityConfig.newBuilder().setVcpuCount(cpu)
        .setMemoryBytes(memoryBytes).build();
    ConnectNetworkConfig networkConfig = ConnectNetworkConfig.newBuilder()
        .setPrimarySubnet(subnet)
        .build();
    // Optionally, you can also specify additional accessible subnets and resolvable
    // DNS domains as part of your network configuration. For example:
    // .addAllAdditionalSubnets(List.of("subnet-1", "subnet-2"))
    // .addAllDnsDomainNames(List.of("dns-1", "dns-2"))
    ConnectGcpConfig gcpConfig = ConnectGcpConfig.newBuilder()
        .setAccessConfig(ConnectAccessConfig.newBuilder().addNetworkConfigs(networkConfig).build())
        .build();
    ConnectCluster connectCluster = ConnectCluster.newBuilder()
        .setCapacityConfig(capacityConfig)
        .setGcpConfig(gcpConfig)
        .setKafkaCluster(kafkaCluster)
        .build();

    // Create the settings to configure the timeout for polling operations
    ManagedKafkaConnectSettings.Builder settingsBuilder = ManagedKafkaConnectSettings.newBuilder();
    TimedRetryAlgorithm timedRetryAlgorithm = OperationTimedPollAlgorithm.create(
        RetrySettings.newBuilder()
            .setTotalTimeoutDuration(Duration.ofHours(1L))
            .build());
    settingsBuilder.createConnectClusterOperationSettings()
        .setPollingAlgorithm(timedRetryAlgorithm);

    try (ManagedKafkaConnectClient managedKafkaConnectClient = ManagedKafkaConnectClient
        .create(settingsBuilder.build())) {
      CreateConnectClusterRequest request = CreateConnectClusterRequest.newBuilder()
          .setParent(LocationName.of(projectId, region).toString())
          .setConnectClusterId(clusterId)
          .setConnectCluster(connectCluster)
          .build();

      // The duration of this operation can vary considerably, typically taking
      // between 10-30 minutes.
      OperationFuture<ConnectCluster, OperationMetadata> future = managedKafkaConnectClient
          .createConnectClusterOperationCallable().futureCall(request);

      // Get the initial LRO and print details.
      OperationSnapshot operation = future.getInitialFuture().get();
      System.out.printf(
          "Connect cluster creation started. Operation name: %s\nDone: %s\nMetadata: %s\n",
          operation.getName(), operation.isDone(), future.getMetadata().get().toString());

      while (!future.isDone()) {
        // The pollingFuture gives us the most recent status of the operation
        RetryingFuture<OperationSnapshot> pollingFuture = future.getPollingFuture();
        OperationSnapshot currentOp = pollingFuture.getAttemptResult().get();
        System.out.printf("Polling Operation:\nName: %s\n Done: %s\n",
            currentOp.getName(),
            currentOp.isDone());
      }

      // NOTE: future.get() blocks completion until the operation is complete (isDone
      // = True)
      ConnectCluster response = future.get();
      System.out.printf("Created connect cluster: %s\n", response.getName());
    } catch (ExecutionException e) {
      System.err.printf("managedKafkaConnectClient.createConnectCluster got err: %s\n", 
          e.getMessage());
      throw e;
    }
  }
}

Python

Antes de probar este ejemplo, sigue las instrucciones de configuración para Python que encontrarás en Instala las bibliotecas cliente. Para obtener más información, consulta la documentación de referencia de la API de Python de Managed Service para Apache Kafka.

Para autenticarte en Managed Service para Apache Kafka, configura las credenciales predeterminadas de la aplicación. Para obtener más información, consulta Configura ADC para un entorno de desarrollo local.

from google.api_core.exceptions import GoogleAPICallError
from google.cloud import managedkafka_v1
from google.cloud.managedkafka_v1.services.managed_kafka_connect import ManagedKafkaConnectClient
from google.cloud.managedkafka_v1.types import ConnectCluster, CreateConnectClusterRequest, ConnectNetworkConfig

# TODO(developer): Update with your values.
# project_id = "my-project-id"
# region = "us-central1"
# connect_cluster_id = "my-connect-cluster"
# kafka_cluster_id = "my-kafka-cluster"
# primary_subnet = "projects/my-project-id/regions/us-central1/subnetworks/default"
# cpu = 12
# memory_bytes = 12884901888  # 12 GiB

connect_client = ManagedKafkaConnectClient()
kafka_client = managedkafka_v1.ManagedKafkaClient()

parent = connect_client.common_location_path(project_id, region)
kafka_cluster_path = kafka_client.cluster_path(project_id, region, kafka_cluster_id)

connect_cluster = ConnectCluster()
connect_cluster.name = connect_client.connect_cluster_path(project_id, region, connect_cluster_id)
connect_cluster.kafka_cluster = kafka_cluster_path
connect_cluster.capacity_config.vcpu_count = cpu
connect_cluster.capacity_config.memory_bytes = memory_bytes
connect_cluster.gcp_config.access_config.network_configs = [ConnectNetworkConfig(primary_subnet=primary_subnet)]
# Optionally, you can also specify accessible subnets and resolvable DNS domains as part of your network configuration.
# For example:
# connect_cluster.gcp_config.access_config.network_configs = [
#     ConnectNetworkConfig(
#         primary_subnet=primary_subnet,
#         additional_subnets=additional_subnets,
#         dns_domain_names=dns_domain_names,
#     )
# ]

request = CreateConnectClusterRequest(
    parent=parent,
    connect_cluster_id=connect_cluster_id,
    connect_cluster=connect_cluster,
)

try:
    operation = connect_client.create_connect_cluster(request=request)
    print(f"Waiting for operation {operation.operation.name} to complete...")
    # Creating a Connect cluster can take 10-40 minutes.
    response = operation.result(timeout=3000)
    print("Created Connect cluster:", response)
except GoogleAPICallError as e:
    print(f"The operation failed with error: {e}")

Supervisa la operación de creación del clúster

Puedes ejecutar el siguiente comando solo si ejecutaste gcloud CLI para crear el clúster de Connect.

  • Por lo general, crear un clúster de Connect tarda entre 20 y 30 minutos. Para hacer un seguimiento del progreso de la creación del clúster, el comando gcloud managed-kafka connect-clusters create usa una operación de larga duración (LRO), que puedes supervisar con el siguiente comando:

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

    Reemplaza lo siguiente:

    • OPERATION_ID por el valor del ID de la operación de la sección anterior.
    • LOCATION por el valor de la ubicación de la sección anterior.

Próximos pasos

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