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 archiviare ed elaborare 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 conoscere quanto segue:

Ruoli e autorizzazioni richiesti per creare un cluster

Per ottenere le autorizzazioni necessarie per creare un cluster, chiedi all'amministratore di concederti il ruolo IAM Editor cluster Kafka gestito (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 le autorizzazioni necessarie per creare un cluster. Per vedere quali sono esattamente le autorizzazioni richieste, espandi la sezione Autorizzazioni obbligatorie:

Autorizzazioni obbligatorie

Per creare un cluster sono necessarie le seguenti autorizzazioni:

Crea un cluster: managedkafka.clusters.create

Potresti anche ottenere queste autorizzazioni con ruoli personalizzati o altri ruoli predefiniti.

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 linee guida su come assegnare un nome a un cluster, consulta 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 richieste almeno 3 vCPU per cluster.
Memoria: la quantità di memoria assegnata al cluster. Devi eseguire il provisioning 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 la massima è 48 GiB (8 GiB per vCPU).

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

Per linee guida sulla stima della capacità necessaria per il cluster, consulta Pianificare 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.
È consentita esattamente una subnet per rete per qualsiasi cluster.
Ogni subnet deve trovarsi nella stessa regione del cluster. Il progetto e la rete possono essere diversi.
Gli indirizzi IP dei server dell'intermediario e di bootstrap sono allocati automaticamente in ogni subnet. Inoltre, le voci DNS per questi indirizzi IP vengono create nelle corrispondenti reti VPC.
Se aggiungi una subnet di un altro progetto, devi concedere le autorizzazioni al account di servizio gestito da Google associato al cluster. Per maggiori informazioni, consulta Connettere un cluster a più progetti.

Dopo aver creato il cluster, puoi aggiornare l'elenco delle subnet. Per maggiori informazioni sul networking, vedi Configura il networking per Managed Service per Apache Kafka.

Cluster pubblici

Se vuoi, puoi configurare il cluster come cluster pubblico per consentire alle applicazioni client di connettersi al cluster ed eseguire operazioni Kafka (incluso l'invio e la ricezione di messaggi) tramite 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, vengono applicate le seguenti regole:

Tutti gli intervalli IP di origine consentiti devono essere specificati in notazione CIDR IPv4 (ad esempio, 203.0.113.0/24 o 198.51.100.5/32).
La dimensione di ogni 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 attivare l'accesso pubblico per un cluster, ti consigliamo di leggere Connettere 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 con l'organizzazione e l'identificazione. Le etichette ti consentono di classificare le risorse in base all'ambiente. Esempi sono "env:production" e "owner:data-engineering".

Puoi filtrare e cercare le risorse in base alle etichette. Ad esempio, supponiamo che tu abbia più cluster Managed Service per Apache Kafka per dipartimenti diversi. 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:

Bilanciamento automatico in caso di scalabilità verticale: se questa opzione è attivata, il servizio attiva automaticamente un ribilanciamento delle repliche quando fai lo scale up del cluster. Questa modalità consente di mantenere una distribuzione uniforme del carico, ma potrebbe influire temporaneamente sulle prestazioni durante l'operazione di ribilanciamento.
Nessun ribilanciamento: se questa opzione è abilitata, il servizio non ribilancia automaticamente le repliche.

Per ulteriori informazioni, vedi Ribilanciamento delle partizioni del broker.

Crittografia

Managed Service per Apache Kafka può criptare i messaggi con Google-owned and Google-managed encryption keys (impostazione predefinita) o chiavi di crittografia gestite dal cliente (CMEK). Ogni messaggio è criptato sia at-rest che 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 inattivi all'interno dei servizi Google Cloud supportati. L'utilizzo di CMEK comporta costi aggiuntivi correlati a Cloud Key Management Service. Per l'utilizzo di CMEK, il portachiavi deve trovarsi nella stessa posizione delle risorse con cui lo utilizzi. Per saperne di più, consulta Configura la crittografia dei messaggi.

Configurazione mTLS

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

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

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

Crea un cluster

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

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

Per creare un cluster:

Console

Nella console Google Cloud , vai alla pagina Cluster.

Vai a Cluster
Seleziona Crea.

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

Per maggiori informazioni 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 Posizione, inserisci una posizione supportata.

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

Per ulteriori informazioni su come dimensionare un cluster, consulta Pianificare le dimensioni del cluster Kafka.
Facoltativo. Per specificare la capacità del disco del broker:
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 di nuovo questa impostazione per questo cluster.
2. Nel campo Spazio di archiviazione locale personalizzato, inserisci le dimensioni del disco per broker, in GiB.
Se lasci il pulsante di attivazione/disattivazione Abilita la dimensione dello spazio di archiviazione locale personalizzata in posizione off, viene eseguito il provisioning della capacità del disco predefinita. Per saperne di più, consulta Configurare le dimensioni del disco del broker.
Per 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. Subnet: il nome della subnet.
4. Percorso URI subnet: questo campo viene compilato automaticamente. In alternativa, puoi inserire qui il percorso della subnet. 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 può essere raggiunto, 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. Per Tipo di chiave, seleziona Cloud KMS.
3. Per 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 posizione 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 subnet 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 Google Cloud ID progetto
LOCATION: la posizione 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 subnet 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 Installare 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ù, vedi 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 in Installare 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 Installare le librerie client. Per saperne di più, consulta la documentazione di riferimento dell'API Python 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ù, vedi Configura ADC per un ambiente di sviluppo locale.

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 seguente comando 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 i problemi 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 affiliate negli Stati Uniti e/o in altri paesi.