Crea un cluster di connessione

Un cluster Connect fornisce un ambiente per i connettori che aiuta a spostare i dati dai deployment Kafka esistenti a un cluster Google Cloud Managed Service per Apache Kafka o a spostare i dati dal cluster Managed Service per Apache Kafka a un altro servizio Google Cloud o a un altro cluster Kafka. Il cluster Kafka secondario può essere un altro cluster Google Cloud Managed Service per Apache Kafka, un cluster autogestito o un cluster on-premise.

Prima di iniziare

Assicurati di aver già creato un cluster Managed Service per Apache Kafka. Devi conoscere il nome del cluster Managed Service per Apache Kafka a cui verrà collegato il cluster di connessione.

Ogni cluster Connect è associato a un cluster Managed Service per Apache Kafka. Questo cluster memorizza lo stato dei connettori in esecuzione sul cluster Connect.

Ruoli e autorizzazioni richiesti per creare un cluster di connessione

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

Autorizzazioni obbligatorie

Per creare un cluster Connect sono necessarie le seguenti autorizzazioni:

  • Concedi l'autorizzazione per creare un cluster di connessione nella località specificata: managedkafka.connectClusters.create

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

Entità ACL obbligatorie

Per impostazione predefinita, i cluster Managed Service per Apache Kafka consentono al cluster di connessione di accedere alle risorse se non sono configurati ACL. Per farlo, imposta allow.everyone.if.no.acl.found su true, che è l'impostazione predefinita.

Tuttavia, se nel cluster Managed Service per Apache Kafka sono configurati ACL, il cluster di connessione non dispone automaticamente delle autorizzazioni di lettura e scrittura per le risorse. Devi concederle manualmente.

Il account di servizio del cluster Connect utilizzato come entità nelle ACL segue questo formato: User:service-{consumer project number}@gcp-sa-managedkafka.iam.gserviceaccount.com.

Se hai configurato gli ACL nel cluster Kafka, concedi al cluster Connect le autorizzazioni di lettura e scrittura per gli argomenti e le autorizzazioni di lettura per i gruppi di consumer utilizzando i seguenti comandi:

/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 *

Per saperne di più su questi comandi, consulta Configura le ACL di Apache Kafka per un controllo degli accessi granulare.

Crea un cluster di connessione in un altro progetto

Managed Service per Apache Kafka utilizza un agente di servizio per accedere alle risorseGoogle Cloud . L'agente di servizio è associato al progetto in cui crei il cluster.

Se crei un cluster Connect in un progetto diverso da quello del cluster Managed Service per Apache Kafka, il cluster Connect e il cluster Kafka utilizzano gli agenti di servizio associati ai rispettivi progetti. In questo caso, l'agente di servizio per il cluster di connessione deve disporre dell'autorizzazione per accedere alle risorse Google Cloud nel progetto del cluster Kafka.

Per concedere le autorizzazioni necessarie, assegna il ruolo Agente di servizio Managed Kafka all'agente di servizio del cluster Connect nel progetto del cluster Kafka. Ad esempio, se crei un cluster Kafka nel progetto kafka-project e un cluster Connect nel progetto connect-project, concedi il ruolo Agente di servizio Managed Kafka su kafka-project all'agente di servizio associato a connect-project.

L'indirizzo email dell'agente di servizio ha il seguente formato: service-PROJECT_NUMBER@gcp-sa-managedkafka.iam.gserviceaccount.com, dove PROJECT_NUMBER è il numero del progetto. Per saperne di più su come concedere il ruolo, consulta Crea e concedi ruoli agli agenti di servizio.

Proprietà di un cluster Connect

Questa sezione descrive le proprietà di un cluster connesso.

Nome cluster di connessione

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

Cluster Kafka principale

Il cluster Managed Service per Apache Kafka associato al tuo cluster Connect. Questo cluster associato (cluster principale) memorizza lo stato dei connettori in esecuzione sul cluster di connessione. In genere, il cluster Managed Service per Apache Kafka principale funge anche da destinazione per tutti i connettori di origine e da input per tutti i connettori sink in esecuzione nel cluster Connect.

Un singolo cluster Managed Service per Apache Kafka può avere più cluster Connect. Se scegli un cluster Managed Service per Apache Kafka in un progetto diverso, assicurati che siano configurate le autorizzazioni appropriate.

Non puoi eseguire l'aggiornamento a un cluster Kafka diverso dopo aver creato il cluster Connect.

Vantaggi della collocazione nella stessa regione per la latenza e i costi di rete

La co-localizzazione dei cluster Managed Service per Apache Kafka e Connect nella stessa regione riduce la latenza e i costi di rete. Ad esempio, supponiamo che il tuo cluster Managed Service per Apache Kafka si trovi in region-a e che tu stia utilizzando un connettore sink per scrivere dati da questo cluster Managed Service per Apache Kafka (origine) in una tabella BigQuery (sink) che si trova anche in region-a. Se esegui il deployment del cluster Connect in region-a, questa scelta di deployment riduce al minimo la latenza per l'operazione di scrittura di BigQuery ed elimina i costi di trasferimento di rete tra regioni tra il cluster Managed Service per Apache Kafka e il cluster Connect.

Considerazioni su latenza e costi di più sistemi

Kafka Connect utilizza i connettori per spostare i dati tra i sistemi. Un lato del connettore interagisce sempre con un cluster Managed Service per Apache Kafka. Un singolo cluster Kafka Connect può eseguire più connettori, ognuno dei quali funge da origine (estrazione dei dati da un sistema) o da sink (invio dei dati a un sistema).

Sebbene un cluster Connect nella stessa regione del cluster Managed Service per Apache Kafka tragga vantaggio da una latenza di comunicazione inferiore tra i due, ogni connettore interagisce anche con un altro sistema, ad esempio una tabella BigQuery o un altro cluster Kafka. Anche se il cluster Connect e il cluster Managed Service per Apache Kafka si trovano nella stessa posizione, l'altro sistema potrebbe trovarsi in una regione diversa. Ciò comporta una latenza e un costo maggiori. La latenza complessiva della pipeline dipende dalle località di tutti e tre i sistemi: il cluster Managed Service per Apache Kafka, il cluster Connect e il sistema di origine o sink.

Ad esempio, se il tuo cluster Managed Service per Apache Kafka si trova in region-a, il tuo cluster Connect in region-b e utilizzi un connettore Cloud Storage per un bucket in region-c, ti verranno addebitati due hop di rete (da region-a a region-b e poi da region-b a region-c o viceversa, a seconda della direzione del connettore).

Valuta attentamente tutte le regioni coinvolte quando pianifichi il posizionamento del cluster Connect per ottimizzare sia la latenza sia i costi.

Configurazione della capacità

La configurazione della capacità richiede di impostare il numero di vCPU e la quantità di memoria per ogni vCPU del cluster di connessione. Puoi aggiornare la capacità di un cluster di connessione dopo averlo creato. Di seguito sono riportate le proprietà per la configurazione della capacità:

  • vCPU: il numero di vCPU assegnate a un cluster Connect. Il valore minimo è 3 vCPU.

  • Memoria: la quantità di memoria assegnata per ogni vCPU. Devi eseguire il provisioning di una quantità di memoria compresa tra 1 GiB e 8 GiB per vCPU. La quantità di memoria può essere aumentata o diminuita entro questi limiti dopo la creazione del cluster.

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

La vCPU e la memoria allocate a ogni worker in un cluster Connect hanno un impatto significativo sulle prestazioni, sulla capacità e sul costo del cluster. Ecco un'analisi dettagliata di come la vCPU e la memoria influiscono su un cluster Connect.

Numero di vCPU

  • Kafka Connect divide il lavoro di un connettore in attività. Ogni attività può elaborare i dati in parallelo. Più vCPU significano che è possibile eseguire più attività contemporaneamente, il che si traduce in un throughput più elevato.

  • Un numero maggiore di vCPU aumenta i costi del cluster Connect.

Memoria

  • Kafka Connect utilizza la memoria per eseguire il buffering dei dati mentre fluiscono tra i connettori e Managed Service per Apache Kafka. Una memoria più grande consente di avere buffer più grandi. Una memoria di grandi dimensioni può migliorare il throughput, soprattutto per i flussi di dati ad alto volume. I connettori che gestiscono messaggi o record di grandi dimensioni richiedono memoria sufficiente per elaborarli senza incorrere in eccezioni OutOfMemoryError.

  • Una maggiore quantità di memoria aumenta il costo del cluster Connect.

  • Se utilizzi una logica di trasformazione complessa, devi allocare più memoria.

Il tuo obiettivo è scegliere la configurazione di capacità corretta per il tuo cluster Connect. Per farlo, devi comprendere la velocità effettiva che il tuo cluster Connect può gestire.

Subnet worker (principale)

La subnet worker, nota anche come subnet principale, connette la tua rete VPC al cluster di connessione. Questa subnet consente ai worker del cluster di raggiungere gli endpoint delle origini e dei sink nella rete consumer, ad esempio i cluster Managed Service per Apache Kafka o i cluster Kafka in hosting autonomo.

Ecco alcuni requisiti per la configurazione della subnet dei worker:

  • La subnet worker è obbligatoria.

  • La subnet deve trovarsi nella stessa regione del cluster di connessione.

  • La subnet deve trovarsi nello stesso VPC principale di una delle subnet connesse dell'elenco del cluster Kafka principale.

  • L'intervallo CIDR della subnet deve avere una dimensione minima di /22 (1024 indirizzi).

Ai worker del cluster vengono assegnati indirizzi IP nella subnet dei worker, utilizzando un'interfaccia Private Service Connect. I worker possono raggiungere qualsiasi destinazione di rete accessibile dalla rete VPC della subnet, con i seguenti requisiti:

  • L'endpoint non deve trovarsi nell'intervallo CIDR 172.16.0.0/14. Questo intervallo è riservato all'uso interno di Managed Service per Apache Kafka Connect.
  • Le regole firewall devono consentire il traffico. Consulta Configura la sicurezza per i collegamenti di rete.
  • Per il traffico internet, devi configurare un Cloud NAT. Ad esempio, è necessario un Cloud NAT per un connettore MirrorMaker per replicare i dati da un cluster Kafka accessibile su internet.
  • Per accedere agli endpoint Private Service Connect che si trovano in un VPC diverso dal VPC della subnet worker, devi assicurarti di utilizzare una configurazione consumer supportata (ad esempio NCC). Per saperne di più, consulta Informazioni sull'accesso ai servizi pubblicati tramite endpoint.

Domini DNS risolvibili

I domini DNS risolvibili, noti anche come nomi di dominio DNS, consentono di rendere disponibili gli indirizzi DNS nella rete VPC consumer per il VPC tenant. In questo modo, il cluster Connect può risolvere i nomi DNS in indirizzi IP, facilitando la comunicazione con altri servizi, inclusi altri cluster Kafka per i connettori MirrorMaker.

Per i domini DNS risolvibili, puoi selezionare un cluster Managed Service per Apache Kafka. Non è necessario configurare il nome di dominio DNS per il cluster Managed Service per Apache Kafka principale, poiché il suo indirizzo di bootstrap viene incluso automaticamente nell'elenco dei domini DNS risolvibili.

Tuttavia, puoi anche specificare un dominio DNS manualmente, il che è necessario se selezioni un cluster Kafka esterno. Il dominio DNS del cluster Managed Service per Apache Kafka principale viene incluso automaticamente. Altri cluster Kafka richiedono comunque la configurazione dei domini DNS.

Risorse di Secret Manager

Alcuni connettori richiedono dati sensibili, come le password, come parte della loro configurazione. Per gestire questo tipo di dati in modo sicuro, puoi archiviarli in Secret Manager e concedere al cluster Connect l'accesso al secret.

Per utilizzare i secret di Secret Manager con Kafka Connect, segui questi passaggi:

  1. Concedi i seguenti ruoli IAM al service account Managed Kafka. Questi ruoli consentono al cluster di connessione di visualizzare i secret e accedere ai relativi payload.

    • Secret Manager Viewer (roles/secretmanager.viewer)
    • Secret Manager Secret Accessor (roles/secretmanager.secretAccessor)

  2. Crea un secret contenente i dati sensibili. Per saperne di più, consulta Crea un secret.

  3. Quando crei o aggiorni il cluster di connessione, specifica i secret a cui il cluster ha accesso. Puoi specificare fino a 32 secret per cluster di connessione.

I secret vengono montati come file nei worker del cluster. I connettori hanno accesso di sola lettura a questi file. Quando crei un connettore, le sue proprietà di configurazione possono fare riferimento ai secret.

  • Per fare riferimento al percorso di un file secret, utilizza il seguente formato:

    /var/secrets/PROJECT_NAME-SECRET_NAME-SECRET_VERSION

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

  • Per utilizzare il valore di un secret come valore di configurazione (ad esempio, una password), utilizza il seguente formato:

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

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

Sostituisci quanto segue:

  • PROJECT_NAME: il nome del progetto Google Cloud .
  • SECRET_NAME: il nome del secret.
  • SECRET_VERSION: la versione del secret.

Etichette

Le etichette sono coppie chiave-valore che ti aiutano nell'organizzazione e nell'identificazione. Ti aiutano a organizzare i cluster Connect. Puoi collegare un'etichetta a ogni cluster Connect, quindi filtrare le risorse in base alle relative etichette. Esempi di etichette sono environment:prod, application:web-app.

Crea un cluster di connessione

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

La creazione di un cluster di connessione richiede dai 20 ai 30 minuti.

Console

  1. Nella console Google Cloud , vai alla pagina Connetti i cluster.

    Vai a Cluster connessi

  2. Fai clic su Crea.

  3. Nel campo Nome cluster di connessione, inserisci un nome per il cluster di connessione. Per saperne di più, consulta le linee guida per assegnare un nome a una risorsa Managed Service per Apache Kafka.

  4. Nell'elenco Cluster Kafka principale, seleziona un cluster Managed Service per Apache Kafka. Per saperne di più, consulta Cluster Kafka principale.

  5. Nell'elenco Regione, seleziona una località per il cluster di connessione. Per maggiori informazioni su come selezionare una località, consulta Cluster Kafka principale.

  6. Nella sezione Configurazione della capacità, inserisci i valori per i seguenti campi o mantieni i valori predefiniti.

    • Nel campo vCPUs, inserisci il numero di CPU virtuali per il cluster.

    • Nel campo Memoria, inserisci la quantità di memoria per CPU, in GiB. Il valore non può superare gli 8 GiB per CPU.

    Per saperne di più, consulta Configurazione della capacità.

  7. Nella sezione Configurazione di rete, seleziona una rete VPC dall'elenco Rete o lascia il valore predefinito. Questo elenco viene compilato quando selezioni il cluster Kafka principale.

  8. Nella sezione Subnet worker, seleziona una subnet dall'elenco Subnet o lascia il valore predefinito. Per saperne di più, consulta Subnet worker. Il campo Percorso URI subnet viene compilato automaticamente quando selezioni la subnet.

  9. (Facoltativo) Aggiungi un dominio DNS risolvibile. Il dominio DNS del cluster Kafka principale viene aggiunto automaticamente come dominio DNS risolvibile. Per specificare altri domini DNS, segui questi passaggi:

    1. Espandi la sezione Domini DNS risolvibili.

    2. Fai clic su Aggiungi dominio DNS.

    3. Per aggiungere il dominio DNS di un cluster Managed Service per Apache Kafka esistente, seleziona il cluster dall'elenco Cluster Kafka. In caso contrario, inserisci il dominio DNS nel campo Dominio DNS.

    4. Fai clic su Fine.

  10. (Facoltativo) Per aggiungere risorse di Secret Manager, segui questi passaggi:

    1. Espandi la sezione Risorse di Secret Manager.

    2. Fai clic su Aggiungi risorsa secret.

    3. Nell'elenco Secret, seleziona un secret.

    4. Nell'elenco Versione secret, seleziona una versione del secret.

    5. Fai clic su Fine.

  11. (Facoltativo) Aggiungi etichette per organizzare il progetto. Per aggiungere un'etichetta:

    1. Espandi la sezione Etichette.

    2. Fai clic su Aggiungi etichetta.

    3. Nel campo Chiave, inserisci la chiave dell'etichetta.

    4. Nel campo Valore, inserisci il valore dell'etichetta.

  12. Fai clic su Crea.

gcloud

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

  2. Esegui il 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]

    Sostituisci quanto segue:

    • CONNECT_CLUSTER_ID: l'ID o il nome del cluster Connect. Per maggiori informazioni su come assegnare un nome a un cluster di connessione, consulta le linee guida per assegnare un nome a una risorsa Managed Service per Apache Kafka. Il nome di un cluster di connessione è immutabile.

    • LOCATION: la località in cui crei il cluster Connect. Deve essere una regione Google Cloudsupportata. Non puoi modificare la località di un cluster Connect dopo la creazione. Per un elenco di località disponibili, consulta le località di Managed Service per Apache Kafka. Per saperne di più sui suggerimenti per le località, consulta Cluster Kafka principale.

    • CPU: il numero di vCPU per il cluster Connect. Il valore minimo è 3 vCPU. Consulta Numero di vCPU.

    • MEMORY: la quantità di memoria per il cluster Connect. Utilizza le unità "MB", "MiB", "GB", "GiB", "TB" o "TiB". Ad esempio, "3GiB". Devi eseguire il provisioning di una quantità di memoria compresa tra 1 GiB e 8 GiB per vCPU. Consulta la sezione Memoria.

    • WORKER_SUBNET: La sottorete di lavoro per il cluster Connect.

      Il formato della subnet è projects/PROJECT_ID/regions/REGION/subnetworks/SUBNET_ID.

      La subnet worker deve trovarsi nella stessa regione del cluster di connessione.

    • PROJECT_ID: (Facoltativo) L'ID del progettoGoogle Cloud . Se non viene fornito, viene utilizzato il progetto corrente.

    • KAFKA_CLUSTER: l'ID o il nome completo del cluster Managed Service per Apache Kafka principale associato al cluster Connect. Vedi Cluster Kafka. Il formato del cluster Kafka è projects/PROJECT_ID/locations/LOCATION/clusters/CLUSTER_ID.

      Non puoi eseguire l'aggiornamento a un cluster Kafka diverso dopo aver creato il cluster Connect.

    • SECRET: (facoltativo) secret da caricare nei worker. Devono essere fornite le versioni esatte dei secret di Secret Manager, gli alias non sono supportati. In un cluster possono essere caricati fino a 32 secret. Formato: projects/PROJECT_ID/secrets/SECRET_NAME/versions/VERSION_ID

    • DNS_DOMAIN_NAME: (facoltativo) nomi di dominio DNS della subnet da rendere visibili al cluster connesso. Il cluster Connect può accedere alle risorse utilizzando i nomi di dominio anziché fare affidamento sugli indirizzi IP. Consulta Peering DNS.

    • LABELS: (facoltativo) etichette da associare al cluster. Per saperne di più sul formato delle etichette, consulta Etichette. Elenco di coppie CHIAVE=VALORE di etichette da aggiungere. Le chiavi devono iniziare con un carattere minuscolo e contenere solo trattini (-), trattini bassi (_), caratteri minuscoli e numeri. I valori devono contenere solo trattini (-), trattini bassi (_), lettere minuscole e numeri.

    • CONFIG_FILE: (facoltativo) il percorso del file JSON o YAML contenente la configurazione che viene sostituita dai valori predefiniti del cluster o del connettore. Questo file supporta anche JSON o YAML in linea.

    • --async: (facoltativo) termina immediatamente, senza attendere il completamento dell'operazione in corso. Con il flag --async, puoi continuare con altre attività mentre la creazione del cluster avviene in background. Se non utilizzi il flag, il sistema attende il completamento dell'operazione prima di restituire una risposta. Devi attendere il completamento dell'aggiornamento del cluster prima di poter continuare con altre attività.

    Ricevi una risposta simile alla seguente:

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

    Memorizza il valore OPERATION_ID per monitorare l'avanzamento. Ad esempio, il valore qui è operation-1753590328249-63ae19098cc06-64300a0a-06512d02.

Terraform

Puoi utilizzare una risorsa Terraform per creare un cluster 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
      }
    }
  }
}

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

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

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.

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.

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

Monitora l'operazione di creazione del cluster

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

  • La creazione di un cluster Connect richiede in genere 20-30 minuti. Per monitorare l'avanzamento della creazione del cluster, il comando gcloud managed-kafka connect-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.

Passaggi successivi

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