Google utilizza la tecnologia AI per tradurre i contenuti nella tua lingua preferita. Le traduzioni generate dall'AI potrebbero contenere errori.

Crea un cluster Google Cloud Managed Service per Apache Kafka

Un cluster Managed Service per Apache Kafka fornisce un ambiente per l'archiviazione e l'elaborazione di flussi di messaggi organizzati in argomenti.

Per creare un cluster, puoi utilizzare la console Google Cloud , Google Cloud CLI, la libreria client o l'API Managed Kafka. Non puoi utilizzare l'API Apache Kafka open source per creare un cluster.

Prima di iniziare

Verifica di avere familiarità con quanto segue:

Ruoli e autorizzazioni richiesti

Per ottenere l'autorizzazione necessaria per creare un cluster, chiedi all'amministratore di concederti il ruolo IAM Managed Kafka Cluster Editor (roles/managedkafka.clusterEditor) nel progetto. Per saperne di più sulla concessione dei ruoli, consulta Gestisci l'accesso a progetti, cartelle e organizzazioni.

Questo ruolo predefinito contiene l'autorizzazione managedkafka.clusters.create necessaria per creare un cluster.

Potresti anche ottenere questa autorizzazione con ruoli personalizzati o altri ruoli predefiniti.

Autorizzazioni subnet del cluster

Quando crei un cluster Managed Service per Apache Kafka, specifichi una o più subnet Virtual Private Cloud (VPC) in cui il cluster è accessibile. Durante l'operazione di creazione del cluster, l'agente di servizio Managed Kafka collega queste subnet al cluster.

L'autorizzazione managedkafka.clusters.create consente indirettamente a un'entità di collegare qualsiasi subnet che l'agente di servizio è autorizzato a utilizzare, anche se l'entità non dispone dell'autorizzazione diretta per utilizzare la subnet.

Proprietà di un cluster Managed Service per Apache Kafka

Quando crei o aggiorni un cluster Managed Service per Apache Kafka, devi specificare le seguenti proprietà.

Nome del cluster

Il nome o l'ID del cluster Managed Service per Apache Kafka che stai creando. Per maggiori informazioni, consulta le linee guida per assegnare un nome a una risorsa Managed Service per Apache Kafka. Il nome di un cluster è immutabile.

Località

La località in cui stai creando il cluster. La località deve essere una delle regioni Google Cloud supportate. La località di un cluster non può essere modificata in un secondo momento. Per un elenco di località disponibili, consulta le località di Managed Service per Apache Kafka.

Configurazione della capacità

La capacità di calcolo, memoria e archiviazione del cluster Kafka. Quando crei un cluster, devi specificare i seguenti valori:

vCPUs: il numero di vCPU nel cluster. Sono necessarie almeno 3 vCPU per cluster.
Memoria: la quantità di memoria assegnata al cluster. Devi eseguire il provisioning di una quantità di memoria compresa tra 1 GiB e 8 GiB per vCPU.

Ad esempio, se crei un cluster con 6 vCPU, la memoria minima che puoi allocare al cluster è 6 GiB (1 GiB per vCPU) e il massimo è 48 GiB (8 GiB per vCPU).

Il servizio esegue automaticamente il provisioning dei broker, inclusi i volumi Persistent Disk per i broker. Facoltativamente, puoi attivare l'opzione Spazio di archiviazione locale personalizzato, che ti consente di specificare la capacità del disco per ogni broker. Per saperne di più, consulta Configura le dimensioni del disco del broker.

Per linee guida su come stimare la capacità necessaria per il tuo cluster, consulta Pianifica le dimensioni del cluster Kafka.

Configurazione di rete

La configurazione di rete è un elenco di subnet VPC in cui il cluster è accessibile. Per produrre o utilizzare i messaggi, i client devono essere in grado di raggiungere una di queste subnet.

Di seguito sono riportate alcune linee guida per la configurazione di rete:

Per un cluster è richiesta almeno una subnet. Il valore massimo è 10.
Per ogni cluster è consentita esattamente una subnet per rete.
Ogni subnet deve trovarsi nella stessa regione del cluster. Il progetto e la rete possono essere diversi.
Gli indirizzi IP dei broker e del server bootstrap sono allocati automaticamente in ogni subnet. Inoltre, le voci DNS per questi indirizzi IP sono create nelle corrispondenti reti VPC.
Se aggiungi una subnet da un progetto diverso, devi concedere le autorizzazioni al account di servizio gestito da Google associato al cluster. Per saperne di più, consulta Connetti un cluster tra progetti.

Dopo aver creato il cluster, puoi aggiornare l'elenco delle subnet. Per saperne di più sul networking, consulta Configura il networking per Managed Service per Apache Kafka.

L'agente di servizio Managed Kafka utilizza le subnet del cluster. L'agente dispone dell'autorizzazione per utilizzare le subnet tramite il ruolo IAM Managed Kafka Service Agent (roles/managedkafka.serviceAgent).

Cluster pubblici

Se vuoi, puoi configurare il tuo cluster come pubblico per consentire alle applicazioni client di connettersi al cluster ed eseguire operazioni Kafka (tra cui l'invio e la ricezione di messaggi) su internet pubblico. Per abilitare l'accesso pubblico, devi specificare uno o più intervalli IP di origine consentiti.

Quando configuri gli intervalli IP di origine consentiti, si applicano le seguenti regole:

Tutti gli intervalli IP di origine consentiti devono essere specificati nella notazione CIDR IPv4 (ad esempio, 203.0.113.0/24 o 198.51.100.5/32).
Ogni dimensione della subnet CIDR deve essere compresa tra /16 e /32.
Gli intervalli CIDR non devono sovrapporsi.
Tutti gli intervalli IP di origine consentiti devono essere instradabili pubblicamente. Gli intervalli IP privati, come 10.0.0.0/8 (RFC 1918), vengono rifiutati.
Puoi specificare un massimo di 500 intervalli IP di origine consentiti.
Gli indirizzi IPv6 non sono supportati.

Prima di abilitare l'accesso pubblico per un cluster, ti consigliamo di leggere Connetti i client a un cluster pubblico per scoprire come viene implementata la funzionalità di accesso pubblico.

Etichette

Le etichette sono coppie chiave-valore che ti aiutano nell'organizzazione e nell'identificazione. Le etichette ti consentono di categorizzare le risorse in base all'ambiente. Esempi: "env:production" e "owner:data-engineering".

Puoi filtrare e cercare le risorse in base alle relative etichette. Ad esempio, supponiamo che tu abbia più cluster Managed Service per Apache Kafka per diversi reparti. Puoi configurare e cercare i cluster con l'etichetta "department:marketing" per trovare quello pertinente.

Configurazione del ribilanciamento

Questa impostazione determina se il servizio ribilancia automaticamente le repliche delle partizioni tra i broker.

Le modalità disponibili sono:

Bilancia automaticamente in caso di scale up: quando questa opzione è abilitata, il servizio attiva automaticamente un bilanciamento delle repliche quando esegui lo scale up del cluster. Questa modalità aiuta a mantenere una distribuzione uniforme del carico, ma potrebbe influire temporaneamente sulle prestazioni durante l'operazione di ribilanciamento.
Nessun ribilanciamento: quando questa opzione è abilitata, il servizio non ribilancia automaticamente le repliche.

Per saperne di più, consulta Bilanciamento delle partizioni del broker.

Crittografia

Managed Service per Apache Kafka può criptare i messaggi con Google-owned and Google-managed encryption keys (predefinito) o con chiavi di crittografia gestite dal cliente (CMEK). Ogni messaggio è crittografato at-rest e in transito. Il tipo di crittografia per un cluster è immutabile.

Google-owned and Google-managed encryption keys vengono utilizzati per impostazione predefinita. Queste chiavi vengono create, gestite e archiviate interamente da Google Cloud all'interno della sua infrastruttura.

Le CMEK sono chiavi di crittografia che gestisci utilizzando Cloud Key Management Service. Questa funzionalità ti consente di avere un maggiore controllo sulle chiavi utilizzate per criptare i dati at-rest all'interno dei servizi Google Cloud supportati. L'utilizzo di CMEK comporta costi aggiuntivi relativi a Cloud Key Management Service. Per l'utilizzo di CMEK, le chiavi automatizzate devono trovarsi nella stessa località delle risorse con cui le utilizzi. Per saperne di più, consulta Configura la crittografia dei messaggi.

Configurazione mTLS

Facoltativamente, puoi configurare mTLS come metodo di autenticazione alternativo che utilizza i certificati client. La configurazione include quanto segue:

Pool CA: un elenco di 1-10 pool di Certificate Authority Service (CAS) considerati attendibili dal cluster per l'autenticazione client.
Regole di mappatura delle entità SSL: una proprietà facoltativa, ma consigliata, del broker ssl.principal.mapping.rules per semplificare i nomi lunghi delle entità dei certificati da utilizzare negli ACL di Kafka.

Per saperne di più su mTLS, consulta Configura l'autenticazione mTLS.

Crea un cluster

Prima di creare un cluster, leggi la documentazione relativa alle proprietà del cluster.

La creazione di un cluster richiede in genere 20-30 minuti.

Per creare un cluster, segui questi passaggi:

Console

Nella console Google Cloud , vai alla pagina Cluster.

Vai a Cluster
Seleziona Crea.

Viene visualizzata la pagina Crea cluster Kafka.
In Nome cluster, inserisci una stringa.

Per saperne di più su come assegnare un nome a un cluster, consulta le linee guida per assegnare un nome a una risorsa Managed Service per Apache Kafka.
In Località, inserisci una località supportata.

Per saperne di più sulle località supportate, consulta Località supportate di Managed Service per Apache Kafka.
In Configurazione della capacità, inserisci i valori per Memoria e vCPU.

Per saperne di più su come dimensionare un cluster, consulta Pianifica le dimensioni del cluster Kafka.
Facoltativo. Per specificare la capacità del disco del broker, segui questi passaggi:
1. Fai clic sul pulsante di attivazione/disattivazione Abilita la dimensione dello spazio di archiviazione locale personalizzata in modo che sia impostato su On.
  
  Importante: dopo aver fatto clic su Salva, non puoi disattivare nuovamente questa impostazione per questo cluster.
2. Nel campo Spazio di archiviazione locale personalizzato, inserisci la dimensione del disco per broker, in GiB.
Se lasci l'opzione Abilita la dimensione dello spazio di archiviazione locale personalizzata in posizione di disattivazione, viene eseguito il provisioning della capacità del disco predefinita. Per saperne di più, consulta Configura le dimensioni del disco del broker.
In Configurazione di rete, inserisci i seguenti dettagli:
1. Progetto: il progetto in cui si trova la subnet. La subnet deve trovarsi nella stessa regione del cluster, ma il progetto potrebbe essere diverso.
2. Rete: la rete a cui è connessa la subnet.
3. Subnetwork: il nome della subnet.
4. Percorso URI subnet: questo campo viene compilato automaticamente. In alternativa, puoi inserire il percorso della subnet qui. Il nome della subnet deve essere nel formato: projects/PROJECT_ID/regions/REGION/subnetworks/SUBNET_ID.
5. Fai clic su Fine.
(Facoltativo) Per aggiungere altre subnet, fai clic su Aggiungi una subnet connessa.

Puoi aggiungere altre subnet, fino a un massimo di 10.
(Facoltativo) Abilita l'accesso pubblico al cluster.
1. Nella sezione Configurazione cluster pubblico, in Il cluster è raggiungibile, seleziona In una rete privata e utilizzando internet pubblico.
2. Nel campo Elenco separato da virgole di intervalli IPv4 di origine consentiti in notazione CIDR, inserisci gli intervalli IP delle tue applicazioni client in notazione CIDR IPv4 (ad esempio, 203.0.113.0/24). Gli indirizzi IPv6 non sono supportati.
(Facoltativo) Imposta una chiave di crittografia gestita dal cliente (CMEK).
1. In Crittografia, seleziona Chiave Cloud KMS.
2. In Tipo di chiave, seleziona Cloud KMS.
3. In Seleziona una chiave gestita dal cliente, seleziona o inserisci la CMEK.
  
  Se non vedi questa opzione, assicurati di aver abilitato l'API Cloud KMS per il progetto.
Per saperne di più, consulta Configura la crittografia dei messaggi.
Fai clic su Crea.

gcloud

Nella console Google Cloud , attiva Cloud Shell.

Attiva Cloud Shell

Nella parte inferiore della console Google Cloud , viene avviata una sessione di Cloud Shell e viene visualizzato un prompt della riga di comando. Cloud Shell è un ambiente shell con Google Cloud CLI già installata e con valori già impostati per il progetto corrente. L'inizializzazione della sessione può richiedere alcuni secondi.

Prima di utilizzare i dati dei comandi riportati di seguito, effettua le seguenti sostituzioni:

PROJECT_ID: l'ID progetto
LOCATION: la località del cluster.
CLUSTER_ID: l'ID del cluster.
CPU_COUNT: il numero di vCPU per il cluster.
MEMORY: la quantità di memoria per il cluster. Esempio: 3GiB.
SUBNET_ID: l'ID della subnet a cui connettersi. Esempio: default.
ENCRYPTION_KEY: l'ID della chiave di crittografia gestita dal cliente da utilizzare per il cluster. Questa opzione è facoltativa.
LABELS: le etichette da associare al cluster.
ALLOWED_SOURCE_IP_RANGES: gli intervalli IP di origine consentiti per il cluster pubblico. Questa opzione è facoltativa.

Esegui questo comando:

Linux, macOS o Cloud Shell

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)

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)

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

Dovresti ricevere una risposta simile alla seguente:

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

Memorizza OPERATION_ID per monitorare progress.

REST

Prima di utilizzare i dati della richiesta, apporta le sostituzioni seguenti:

PROJECT_ID: il tuo ID progetto Google Cloud
LOCATION: la località del cluster.
CLUSTER_ID: l'ID del cluster.
CPU_COUNT: il numero di vCPU per il cluster.
MEMORY: la quantità di memoria per il cluster, in byte. Esempio: 3221225472.
SUBNET_ID: l'ID della subnet a cui connettersi. Esempio: default.

Metodo HTTP e URL:

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

Corpo JSON della richiesta:

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

Per inviare la richiesta, espandi una di queste opzioni:

curl (Linux, macOS o Cloud Shell)

Nota: il seguente comando presuppone che tu abbia eseguito l'accesso all'interfaccia a riga di comando gcloud con il tuo account utente eseguendo gcloud init o gcloud auth login oppure utilizzando Cloud Shell, che consente di accedere automaticamente all'interfaccia a riga di comando gcloud. Puoi controllare l'account attualmente attivo eseguendo gcloud auth list.

Salva il corpo della richiesta in un file denominato request.json, quindi esegui il comando seguente:

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: il comando seguente presuppone che tu abbia eseguito l'accesso all'interfaccia a riga di comando gcloud con il tuo account utente eseguendo gcloud init o gcloud auth login . Puoi controllare l'account attualmente attivo eseguendo gcloud auth list.

Salva il corpo della richiesta in un file denominato request.json, quindi esegui il comando seguente:

$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

Dovresti ricevere una risposta JSON simile alla seguente:

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

Puoi utilizzare una risorsa Terraform per creare un cluster.

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

Per scoprire come applicare o rimuovere una configurazione Terraform, consulta Comandi Terraform di base.

Vai

Prima di provare questo esempio, segui le istruzioni di configurazione di Go in Installa le librerie client. Per saperne di più, consulta la documentazione di riferimento dell'API Go di Managed Service per Apache Kafka.

Per eseguire l'autenticazione in Managed Service per Apache Kafka, configura le Credenziali predefinite dell'applicazione (ADC). Per saperne di più, consulta Configura ADC per un ambiente di sviluppo locale.

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

Prima di provare questo esempio, segui le istruzioni di configurazione di Java riportate in Installa le librerie client. Per saperne di più, consulta la documentazione di riferimento dell'API Java di Managed Service per Apache Kafka.

Per eseguire l'autenticazione in Managed Service per Apache Kafka, configura le Credenziali predefinite dell'applicazione. Per saperne di più, consulta Configura ADC per un ambiente di sviluppo locale.


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

Prima di provare questo esempio, segui le istruzioni di configurazione di Python in Installa le librerie client. Per saperne di più, consulta la documentazione di riferimento dell'API Python di Managed Service per Apache Kafka.

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

Monitora l'operazione di creazione del cluster

Puoi eseguire il comando seguente solo se hai utilizzato gcloud CLI per creare il cluster.

La creazione di un cluster richiede in genere 20-30 minuti. Per monitorare l'avanzamento della creazione del cluster, il comando gcloud managed-kafka clusters create utilizza un'operazione a lunga esecuzione (LRO), che puoi monitorare utilizzando il seguente comando:
```
gcloud managed-kafka operations describe OPERATION_ID \
    --location=LOCATION
```
Sostituisci quanto segue:
- OPERATION_ID con il valore dell'ID operazione della sezione precedente.
- LOCATION con il valore della località della sezione precedente.

Risoluzione dei problemi

Per informazioni su come risolvere gli errori relativi alla creazione di un cluster, consulta Errori di creazione del cluster Kafka.

Passaggi successivi

Apache Kafka® è un marchio registrato di Apache Software Foundation o delle sue società affiliate negli Stati Uniti e/o in altri paesi.