Crea un clúster de Google Cloud Managed Service para Apache Kafka

Un clúster de Managed Service para Apache Kafka proporciona un entorno para almacenar y procesar flujos de mensajes organizados en temas.

Para crear un clúster, puedes usar la consola de Google Cloud , Google Cloud CLI, la biblioteca cliente o la API de Managed Kafka. No puedes usar la API de Apache Kafka de código abierto para crear un clúster.

Antes de comenzar

Verifica que conozcas los siguientes conceptos:

• Cómo funciona Managed Service para Apache Kafka.

• Los conceptos de redes relacionados con Managed Service para Apache Kafka.

Roles y permisos requeridos

Para obtener el permiso que necesitas para crear un clúster, pídele a tu administrador que te otorgue el rol de IAM de editor de clústeres de Kafka administrados (roles/managedkafka.clusterEditor) 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 el permiso managedkafka.clusters.create, que se requiere para crear un clúster.

También puedes obtener este permiso con roles personalizados o con otros roles predefinidos.

Permisos de la subred del clúster

Cuando creas un clúster de Managed Service para Apache Kafka, especificas una o más subredes de nube privada virtual (VPC) en las que se puede acceder al clúster. Durante la operación de creación del clúster, el agente de servicio de Kafka administrado adjunta estas subredes al clúster.

El permiso managedkafka.clusters.create permite que una principal adjunte de forma indirecta cualquier subred que el agente de servicio tenga permiso para usar, incluso si la principal no tiene permiso directo para usar la subred.

Propiedades de un clúster de Managed Service para Apache Kafka

Cuando creas o actualizas un clúster de Managed Service para Apache Kafka, debes especificar las siguientes propiedades.

Nombre del clúster

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

Ubicación

Es la ubicación donde se creará el clúster. La ubicación debe ser una de las regiones de Google Cloud admitidas. No podrás cambiarla más adelante. Para ver la lista de ubicaciones disponibles, consulta las ubicaciones de Managed Service para Apache Kafka.

Configuración de capacidad

La capacidad de procesamiento, memoria y almacenamiento del clúster de Kafka. Cuando creas un clúster, debes especificar los siguientes valores:

• vCPUs: Es la cantidad de CPUs virtuales en el clúster. Se requieren al menos 3 CPUs virtuales por clúster.

• Memoria: Es la cantidad de memoria que se asigna al clúster. Debes aprovisionar entre 1 GiB y 8 GiB por CPU virtual.

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

El servicio aprovisiona automáticamente agentes, incluidos los volúmenes de Persistent Disk para los agentes. De manera opcional, puedes habilitar el Almacenamiento local personalizado, que te permite especificar la capacidad del disco para cada agente. Para obtener más información, consulta Configura el tamaño del disco del agente.

Para obtener lineamientos sobre cómo estimar la capacidad que necesita tu clúster, consulta Planifica el tamaño de tu clúster de Kafka.

Configuración de red

La configuración de red es una lista de subredes de VPC en las que se puede acceder al clúster. Para producir o consumir mensajes, los clientes deben poder acceder a una de estas subredes.

A continuación, se presentan algunos lineamientos para la configuración de tu red:

• Se requiere un mínimo de 1 subred para un clúster. La cantidad máxima es 10.

• Se permite exactamente 1 subred por red para cualquier clúster determinado.

• Cada subred debe estar en la misma región que el clúster. El proyecto y la red pueden ser diferentes.

• Las direcciones IP de los agentes y el servidor de arranque se asignan automáticamente en cada subred. Además, las entradas del DNS para estas direcciones IP se crean en las redes de VPC correspondientes.

• Si agregas una subred de un proyecto diferente, debes otorgar permisos a la cuenta de servicio administrada por Google que está asociada con el clúster. Para obtener más información, consulta Conecta un clúster entre proyectos.

Después de crear el clúster, puedes actualizar la lista de subredes. Para obtener más información sobre las redes, consulta Configura las redes para Managed Service para Apache Kafka.

El agente de servicio de Kafka administrado usa las subredes del clúster. El agente tiene permiso para usar las subredes a través del rol de IAM de agente de servicio de Kafka administrado (roles/managedkafka.serviceAgent).

Clústeres públicos

De manera opcional, puedes configurar tu clúster como público para permitir que las aplicaciones cliente se conecten a él y ejecuten operaciones de Kafka (incluido el envío y la recepción de mensajes) a través de la Internet pública. Para habilitar el acceso público, debes especificar uno o más rangos de IP de origen permitidos.

Cuando configuras los rangos de IP de origen permitidos, se aplican las siguientes reglas:

• Todos los rangos de IP de origen permitidos deben especificarse en la notación CIDR de IPv4 (por ejemplo, 203.0.113.0/24 o 198.51.100.5/32).

• Cada tamaño de subred CIDR debe estar entre /16 y /32.

• Los rangos CIDR no deben superponerse.

• Todos los rangos de IP de origen permitidos deben ser enrutables de forma pública. Se rechazan los rangos de IP privados, como 10.0.0.0/8 (RFC 1918).

• Puedes especificar un máximo de 500 rangos de IP de origen permitidos.

• No se admiten las direcciones IPv6.

Precaución: Para proteger tu clúster, permite el acceso solo desde rangos de IP externos de confianza. No expongas tu clúster a rangos no confiables.

Antes de habilitar el acceso público para un clúster, te recomendamos que leas Conecta clientes a un clúster público para aprender cómo se implementa la función de acceso público.

Etiquetas

Las etiquetas son pares clave-valor que te ayudan con la organización y la identificación. Las etiquetas te permiten categorizar los recursos según el entorno. Por ejemplo, "env:production" y "owner:data-engineering".

Puedes filtrar y buscar recursos según sus etiquetas. Por ejemplo, supongamos que tienes varios clústeres de Managed Service para Apache Kafka para diferentes departamentos. Puedes configurar y buscar clústeres con la etiqueta "department:marketing" para encontrar el relevante.

Configuración de rebalanceo

Este parámetro de configuración determina si el servicio reequilibra automáticamente las réplicas de partición entre los agentes.

Los modos disponibles son los siguientes:

• Reequilibrio automático en el escalamiento vertical: Cuando esta opción está habilitada, el servicio activa automáticamente un reequilibrio de las réplicas cuando escalas verticalmente el clúster. Este modo ayuda a mantener una distribución uniforme de la carga, pero puede afectar temporalmente el rendimiento durante la operación de rebalanceo.

• Sin rebalanceo: Cuando esta opción está habilitada, el servicio no rebalancea las réplicas de forma automática.

Para obtener más información, consulta Reequilibra las particiones de los agentes.

Encriptación

Managed Service para Apache Kafka puede encriptar mensajes con Google-owned and Google-managed encryption keys (predeterminado) o claves de encriptación administradas por el cliente (CMEK). Todos los mensajes se encriptan en reposo y en tránsito. El tipo de encriptación de un clúster es inmutable.

De forma predeterminada, se usan Google-owned and Google-managed encryption keys. Google Cloud crea, administra y almacena estas claves por completo dentro de su infraestructura.

Las CMEK son claves de encriptación que administras con Cloud Key Management Service. Esta función te permite tener un mayor control sobre las claves que se usan para encriptar los datos en reposo en los servicios de Google Cloud compatibles. El uso de CMEK genera costos adicionales relacionados con Cloud Key Management Service. Para usar CMEK, el llavero de claves debe estar en la misma ubicación que los recursos con los que lo usas. Para obtener más información, consulta Configura la encriptación de mensajes.

Configuración de mTLS

De manera opcional, puedes configurar mTLS como un método de autenticación alternativo que use certificados de cliente. La configuración incluye lo siguiente:

• Grupos de AC: Es una lista de 1 a 10 grupos de Certificate Authority Service (CAS) en los que confía el clúster para la autenticación de clientes.

• Reglas de asignación de principales de SSL: Es una propiedad opcional, pero recomendada, del agente de ssl.principal.mapping.rules para simplificar los nombres de principales de certificados largos y usarlos en las LCA de Kafka.

Para obtener más información sobre mTLS, consulta Configura la autenticación mTLS.

Crea un clúster

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

Por lo general, la creación de un clúster tarda entre 20 y 30 minutos.

Para crear un clúster, sigue estos pasos:

Console

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

   Ir a los clústeres

2. Selecciona Crear.

   Se abrirá la página Crear clúster de Kafka.

3. En Nombre del clúster, ingresa una cadena.

   Para obtener más información sobre cómo asignar un nombre a un clúster, consulta Lineamientos para asignar un nombre a un recurso de Managed Service para Apache Kafka.

4. En Ubicación, ingresa una ubicación compatible.

   Para obtener más información sobre las ubicaciones admitidas, consulta Ubicaciones admitidas de Managed Service para Apache Kafka.

5. En Configuración de capacidad, ingresa los valores de Memoria y CPU virtuales.

   Para obtener más información sobre cómo ajustar el tamaño de un clúster, consulta Planifica el tamaño de tu clúster de Kafka.

6. Es opcional. Para especificar la capacidad del disco del agente, haz lo siguiente:

   1. Haz clic en el botón de activación Habilitar el tamaño de almacenamiento local personalizado para llevarlo a la posición activada.

      Importante: Después de hacer clic en Guardar, no podrás volver a inhabilitar este parámetro de configuración para este clúster.

   2. En el campo Almacenamiento local personalizado, ingresa el tamaño del disco por agente, en GiB.

   Si dejas el botón de activación Habilitar el tamaño de almacenamiento local personalizado en la posición de desactivado, se aprovisionará la capacidad de disco predeterminada. Para obtener más información, consulta Configura el tamaño del disco del agente.

7. En Configuración de red, ingresa los siguientes detalles:

   1. Proyecto: Es el proyecto en el que se encuentra la subred. La subred debe estar ubicada en la misma región que el clúster, pero el proyecto puede ser diferente.
   2. Red: Es la red a la que está conectada la subred.
   3. Subred: Es el nombre de la subred.
   4. Ruta de acceso del URI de la subred: Este campo se completa automáticamente. También puedes ingresar la ruta de la subred aquí. El nombre de la subred debe tener el siguiente formato: projects/PROJECT_ID/regions/REGION/subnetworks/SUBNET_ID.
   5. Haz clic en Listo.

8. Opcional: Para agregar subredes adicionales, haz clic en Agregar una subred conectada.

   Puedes agregar subredes adicionales, hasta un valor máximo de 10.

9. Opcional: Habilita el acceso público a tu clúster.

   1. En la sección Configuración del clúster público, en Se puede acceder al clúster, selecciona En una red privada y con Internet pública.

   2. En el campo Lista separada por comas de rangos IPv4 de origen permitidos en notación CIDR, ingresa los rangos de IP de tus aplicaciones cliente en notación CIDR IPv4 (por ejemplo, 203.0.113.0/24). No se admiten las direcciones IPv6.

10. Opcional: Configura una clave de encriptación administrada por el cliente (CMEK).

    1. En Encriptación, selecciona Clave de Cloud KMS.

    2. En Tipo de clave, selecciona Cloud KMS.

    3. En Selecciona una clave administrada por el cliente, selecciona o ingresa la CMEK.

       Si no ves esta opción, asegúrate de haber habilitado la API de Cloud KMS para el proyecto.

    Para obtener más información, consulta Configura la encriptación de mensajes.

11. 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. Antes de usar cualquiera de los datos de comando a continuación, haz los siguientes reemplazos:

   • PROJECT_ID: Es el ID del proyecto.
   • LOCATION: Es la ubicación del clúster.
   • CLUSTER_ID: Es el ID del clúster.
   • CPU_COUNT: Es la cantidad de CPU virtuales para el clúster.
   • MEMORY: Es la cantidad de memoria para el clúster. Ejemplo: 3GiB.
   • SUBNET_ID: Es el ID de la subred a la que se conectará. Ejemplo: default.
   • ENCRYPTION_KEY: Es el ID de la clave de encriptación administrada por el cliente que se usará para el clúster. Esto es opcional.
   • LABELS: Son las etiquetas que se asociarán con el clúster.
   • ALLOWED_SOURCE_IP_RANGES: los rangos de IP de origen permitidos para el clúster público. Esto es opcional.

   Ejecuta el siguiente comando:

   Linux, macOS o Cloud Shell

   Nota: Asegúrate de haber inicializado Google Cloud CLI con autenticación y un proyecto mediante la ejecución de gcloud init o gcloud auth login y gcloud config set project.

   gcloud managed-kafka clusters create CLUSTER_ID \
       --location=LOCATION \
       --cpu=CPU_COUNT \
       --memory=MEMORY \
       --subnets=projects/PROJECT_ID/regions/LOCATION/subnetworks/SUBNET_ID \
       --auto-rebalance \
       --encryption-key=ENCRYPTION_KEY \
       --async \
       --labels=LABELS \
       --public-cluster \
       --allowed-source-ip-ranges=ALLOWED_SOURCE_IP_RANGES

   Windows (PowerShell)

   Nota: Asegúrate de haber inicializado Google Cloud CLI con autenticación y un proyecto mediante la ejecución de gcloud init o gcloud auth login y gcloud config set project.

   gcloud managed-kafka clusters create CLUSTER_ID `
       --location=LOCATION `
       --cpu=CPU_COUNT `
       --memory=MEMORY `
       --subnets=projects/PROJECT_ID/regions/LOCATION/subnetworks/SUBNET_ID `
       --auto-rebalance `
       --encryption-key=ENCRYPTION_KEY `
       --async `
       --labels=LABELS `
       --public-cluster `
       --allowed-source-ip-ranges=ALLOWED_SOURCE_IP_RANGES

   Windows (cmd.exe)

   Nota: Asegúrate de haber inicializado Google Cloud CLI con autenticación y un proyecto mediante la ejecución de gcloud init, o gcloud auth login y gcloud config set project.

   gcloud managed-kafka clusters create CLUSTER_ID ^
       --location=LOCATION ^
       --cpu=CPU_COUNT ^
       --memory=MEMORY ^
       --subnets=projects/PROJECT_ID/regions/LOCATION/subnetworks/SUBNET_ID ^
       --auto-rebalance ^
       --encryption-key=ENCRYPTION_KEY ^
       --async ^
       --labels=LABELS ^
       --public-cluster ^
       --allowed-source-ip-ranges=ALLOWED_SOURCE_IP_RANGES

   Deberías recibir una respuesta similar a la que figura a continuación:

   done: false
   metadata:
     '@type': type.googleapis.com/google.cloud.managedkafka.v1.OperationMetadata
     apiVersion: v1
     createTime: 'CREATE_TIME'
     requestedCancellation: false
     target: projects/PROJECT_ID/locations/LOCATION/clusters/CLUSTER_ID
     verb: create
   name: projects/PROJECT_ID/locations/LOCATION/operations/OPERATION_ID
   

   Almacena el OPERATION_ID para hacer un seguimiento de progress.

REST

Antes de usar cualquiera de los datos de solicitud a continuación, realiza los siguientes reemplazos:

• PROJECT_ID: Es el ID del proyecto de Google Cloud .
• LOCATION: Es la ubicación del clúster.
• CLUSTER_ID: Es el ID del clúster.
• CPU_COUNT: Es la cantidad de CPU virtuales para el clúster.
• MEMORY: Es la cantidad de memoria para el clúster, en bytes. Ejemplo: 3221225472.
• SUBNET_ID: Es el ID de la subred a la que se conectará. Ejemplo: default.

Método HTTP y URL:

POST https://managedkafka.googleapis.com/v1/projects/PROJECT_ID/locations/LOCATION/clusters?clusterId=CLUSTER_ID

Cuerpo JSON de la solicitud:

{
  "capacityConfig": {
    "vcpuCount": CPU_COUNT,
    "memoryBytes": MEMORY
  },
  "gcpConfig": {
    "accessConfig": {
      "networkConfigs": [
        {
          "subnet": "projects/PROJECT_ID/regions/LOCATION/subnetworks/SUBNET_ID"
        }
      ]
    }
  }
}

Para enviar tu solicitud, expande una de estas opciones:

curl (Linux, macOS o Cloud Shell)

Nota: Con el siguiente comando, se supone que accediste a gcloud CLI con tu cuenta de usuario a través de la ejecución de gcloud init o gcloud auth login , o a través del uso de Cloud Shell, que accede de forma automática a gcloud CLI . Para comprobar la cuenta activa actual, ejecuta gcloud auth list.

Guarda el cuerpo de la solicitud en un archivo llamado request.json y ejecuta el siguiente comando:

curl -X POST \
     -H "Authorization: Bearer $(gcloud auth print-access-token)" \
     -H "Content-Type: application/json; charset=utf-8" \
     -d @request.json \
     "https://managedkafka.googleapis.com/v1/projects/PROJECT_ID/locations/LOCATION/clusters?clusterId=CLUSTER_ID"

PowerShell (Windows)

Nota: El siguiente comando supone que accediste a gcloud CLI con tu cuenta de usuario a través de la ejecución de gcloud init o gcloud auth login . Para comprobar la cuenta activa actual, ejecuta gcloud auth list.

Guarda el cuerpo de la solicitud en un archivo llamado request.json y ejecuta el siguiente comando:

$cred = gcloud auth print-access-token
$headers = @{ "Authorization" = "Bearer $cred" }

Invoke-WebRequest `
    -Method POST `
    -Headers $headers `
    -ContentType: "application/json; charset=utf-8" `
    -InFile request.json `
    -Uri "https://managedkafka.googleapis.com/v1/projects/PROJECT_ID/locations/LOCATION/clusters?clusterId=CLUSTER_ID" | Select-Object -Expand Content

Deberías recibir una respuesta JSON similar a la que se muestra a continuación:

{
  "name": "projects/PROJECT_ID/locations/LOCATION/operations/OPERATION_ID",
  "metadata": {
    "@type": "type.googleapis.com/google.cloud.managedkafka.v1.OperationMetadata",
    "createTime": "CREATE_TIME",
    "target": "projects/PROJECT_ID/locations/LOCATION/clusters/CLUSTER_ID",
    "verb": "create",
    "requestedCancellation": false,
    "apiVersion": "v1"
  },
  "done": false
}

Terraform

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

resource "google_managed_kafka_cluster" "default" {
  project    = data.google_project.default.project_id # Replace this with your project ID in quotes
  cluster_id = "my-cluster-id"
  location   = "us-central1"
  capacity_config {
    vcpu_count   = 3
    memory_bytes = 3221225472
  }
  gcp_config {
    access_config {
      network_configs {
        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 createCluster(w io.Writer, projectID, region, clusterID, subnet string, cpu, memoryBytes int64, opts ...option.ClientOption) error {
	// projectID := "my-project-id"
	// region := "us-central1"
	// clusterID := "my-cluster"
	// subnet := "projects/my-project-id/regions/us-central1/subnetworks/default"
	// cpu := 3
	// memoryBytes := 3221225472
	ctx := context.Background()
	client, err := managedkafka.NewClient(ctx, opts...)
	if err != nil {
		return fmt.Errorf("managedkafka.NewClient got err: %w", err)
	}
	defer client.Close()

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

	// Memory must be between 1 GiB and 8 GiB per CPU.
	capacityConfig := &managedkafkapb.CapacityConfig{
		VcpuCount:   cpu,
		MemoryBytes: memoryBytes,
	}
	var networkConfig []*managedkafkapb.NetworkConfig
	networkConfig = append(networkConfig, &managedkafkapb.NetworkConfig{
		Subnet: subnet,
	})
	platformConfig := &managedkafkapb.Cluster_GcpConfig{
		GcpConfig: &managedkafkapb.GcpConfig{
			AccessConfig: &managedkafkapb.AccessConfig{
				NetworkConfigs: networkConfig,
			},
		},
	}
	rebalanceConfig := &managedkafkapb.RebalanceConfig{
		Mode: managedkafkapb.RebalanceConfig_AUTO_REBALANCE_ON_SCALE_UP,
	}
	cluster := &managedkafkapb.Cluster{
		Name:            clusterPath,
		CapacityConfig:  capacityConfig,
		PlatformConfig:  platformConfig,
		RebalanceConfig: rebalanceConfig,
	}

	req := &managedkafkapb.CreateClusterRequest{
		Parent:    locationPath,
		ClusterId: clusterID,
		Cluster:   cluster,
	}
	op, err := client.CreateCluster(ctx, req)
	if err != nil {
		return fmt.Errorf("client.CreateCluster got err: %w", err)
	}
	// The duration of this operation can vary considerably, typically taking 10-40 minutes.
	resp, err := op.Wait(ctx)
	if err != nil {
		return fmt.Errorf("op.Wait got err: %w", err)
	}
	fmt.Fprintf(w, "Created 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.AccessConfig;
import com.google.cloud.managedkafka.v1.CapacityConfig;
import com.google.cloud.managedkafka.v1.Cluster;
import com.google.cloud.managedkafka.v1.CreateClusterRequest;
import com.google.cloud.managedkafka.v1.GcpConfig;
import com.google.cloud.managedkafka.v1.LocationName;
import com.google.cloud.managedkafka.v1.ManagedKafkaClient;
import com.google.cloud.managedkafka.v1.ManagedKafkaSettings;
import com.google.cloud.managedkafka.v1.NetworkConfig;
import com.google.cloud.managedkafka.v1.OperationMetadata;
import com.google.cloud.managedkafka.v1.RebalanceConfig;
import java.time.Duration;
import java.util.concurrent.ExecutionException;

public class CreateCluster {

  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-cluster";
    String subnet = "my-subnet"; // e.g. projects/my-project/regions/my-region/subnetworks/my-subnet
    int cpu = 3;
    long memoryBytes = 3221225472L; // 3 GiB
    createCluster(projectId, region, clusterId, subnet, cpu, memoryBytes);
  }

  public static void createCluster(
      String projectId, String region, String clusterId, String subnet, int cpu, long memoryBytes)
      throws Exception {
    CapacityConfig capacityConfig =
        CapacityConfig.newBuilder().setVcpuCount(cpu).setMemoryBytes(memoryBytes).build();
    NetworkConfig networkConfig = NetworkConfig.newBuilder().setSubnet(subnet).build();
    GcpConfig gcpConfig =
        GcpConfig.newBuilder()
            .setAccessConfig(AccessConfig.newBuilder().addNetworkConfigs(networkConfig).build())
            .build();
    RebalanceConfig rebalanceConfig =
        RebalanceConfig.newBuilder()
            .setMode(RebalanceConfig.Mode.AUTO_REBALANCE_ON_SCALE_UP)
            .build();
    Cluster cluster =
        Cluster.newBuilder()
            .setCapacityConfig(capacityConfig)
            .setGcpConfig(gcpConfig)
            .setRebalanceConfig(rebalanceConfig)
            .build();

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

    try (ManagedKafkaClient managedKafkaClient = ManagedKafkaClient.create(
        settingsBuilder.build())) {

      CreateClusterRequest request =
          CreateClusterRequest.newBuilder()
              .setParent(LocationName.of(projectId, region).toString())
              .setClusterId(clusterId)
              .setCluster(cluster)
              .build();

      // The duration of this operation can vary considerably, typically taking between 10-40
      // minutes.
      OperationFuture<Cluster, OperationMetadata> future =
          managedKafkaClient.createClusterOperationCallable().futureCall(request);

      // Get the initial LRO and print details.
      OperationSnapshot operation = future.getInitialFuture().get();
      System.out.printf("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)
      Cluster response = future.get();
      System.out.printf("Created cluster: %s\n", response.getName());
    } catch (ExecutionException e) {
      System.err.printf("managedKafkaClient.createCluster got err: %s", e.getMessage());
    }
  }
}

Python

Antes de probar esta muestra, sigue las instrucciones de configuración para Python que se encuentran 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

# TODO(developer)
# project_id = "my-project-id"
# region = "us-central1"
# cluster_id = "my-cluster"
# subnet = "projects/my-project-id/regions/us-central1/subnetworks/default"
# cpu = 3
# memory_bytes = 3221225472

client = managedkafka_v1.ManagedKafkaClient()

cluster = managedkafka_v1.Cluster()
cluster.name = client.cluster_path(project_id, region, cluster_id)
cluster.capacity_config.vcpu_count = cpu
cluster.capacity_config.memory_bytes = memory_bytes
cluster.gcp_config.access_config.network_configs = [
    managedkafka_v1.NetworkConfig(subnet=subnet)
]
cluster.rebalance_config.mode = (
    managedkafka_v1.RebalanceConfig.Mode.AUTO_REBALANCE_ON_SCALE_UP
)

request = managedkafka_v1.CreateClusterRequest(
    parent=client.common_location_path(project_id, region),
    cluster_id=cluster_id,
    cluster=cluster,
)

try:
    operation = client.create_cluster(request=request)
    print(f"Waiting for operation {operation.operation.name} to complete...")
    # The duration of this operation can vary considerably, typically taking 10-40 minutes.
    # We can set a timeout of 3000s (50 minutes).
    response = operation.result(timeout=3000)
    print("Created cluster:", response)
except GoogleAPICallError as e:
    print(f"The operation failed with error: {e.message}")

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

Puedes ejecutar el siguiente comando solo si usaste gcloud CLI para crear el clúster.

• Por lo general, la creación de un clúster tarda entre 20 y 30 minutos. Para hacer un seguimiento del progreso de la creación del clúster, el comando gcloud managed-kafka 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.

Solucionar problemas

Para obtener información sobre cómo solucionar problemas relacionados con la creación de un clúster, consulta Errores de creación de clústeres de Kafka.

Próximos pasos

• Supervisa tu clúster.

• Actualiza tu clúster.

• Enumera todos tus clústeres.

• Visualiza tu clúster.

• Borrar un clúster.

• Crea un tema dentro de tu clúster.

• Configura las redes para Managed Service para Apache Kafka.

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