Crea un cluster di connessione

Un cluster di connessione fornisce un ambiente per i connettori che consente di spostare i dati dalle implementazioni Kafka esistenti in un cluster Google Cloud Managed Service per Apache Kafka o di spostare i dati dal cluster Managed Service per Apache Kafka a un altro Google Cloud servizio o un altro cluster Kafka. Il cluster Kafka secondario può essere un altro cluster Google Cloud Managed Service per Apache Kafka, un cluster autogestito o 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 di connessione è 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.

Per saperne di più su questo ruolo, consulta Ruoli predefiniti di Managed Service per Apache Kafka.

Entità ACL obbligatorie

Per impostazione predefinita, i cluster Managed Service per Apache Kafka consentono l'accesso del cluster di connessione alle risorse se non sono configurate ACL. A questo scopo, imposta allow.everyone.if.no.acl.found su true, che è l'impostazione predefinita.

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

Il account di servizio del cluster Connect utilizzato come principale negli elenchi di controllo degli accessi segue questo formato: User:service-{consumer project number}@gcp-sa-managedkafka.iam.gserviceaccount.com.

Se hai configurato gli ACL sul cluster Kafka, concedi al cluster Connect le autorizzazioni di lettura e scrittura per gli argomenti e le autorizzazioni di lettura per i gruppi di consumatori 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 ulteriori informazioni su questi comandi, consulta Configura gli elenchi di controllo dell'accesso di Apache Kafka per un controllo dell'accesso granulare.

Crea un cluster di connessione in un altro progetto

Quando crei un cluster di connessione, questo condivide lo stesso service agent con il cluster Managed Service per Apache Kafka che si trova nello stesso progetto. Se questo cluster Managed Service per Apache Kafka è designato come cluster Kafka principale collegato al cluster di connessione, non sono necessarie autorizzazioni aggiuntive.

L'agente di servizio ha il formato service-<project_number>@gcp-sa-managedkafka.iam.gserviceaccount.com. Il numero di progetto è quello del progetto contenente il cluster di connessione e il cluster Managed Service per Apache Kafka.

Se il cluster di connessione si trova nel progetto A e il cluster Managed Service per Apache Kafka associato si trova nel progetto B, segui questi passaggi:

  1. Assicurati che l'API Managed Kafka sia abilitata sia per il progetto A sia per il progetto B.

    Abilitare l'API

  2. Identifica il service agent del cluster Connect nel progetto A.

    L'agente di servizio ha il formato service-<project_number>@gcp-sa-managedkafka.iam.gserviceaccount.com.

  3. Nel progetto B, concedi al account di servizio del cluster Connect il ruolo Managed Kafka Client (roles/managedkafka.client).

    Questo ruolo concede le autorizzazioni necessarie per connettersi al cluster Managed Service per Apache Kafka ed eseguire operazioni come la lettura e la scrittura dei dati.

    Per saperne di più su come concedere il ruolo, consulta Creare e concedere ruoli agli agenti di servizio.

Segui sempre il principio del privilegio minimo quando concedi le autorizzazioni. Concedi solo le autorizzazioni necessarie per garantire la sicurezza e impedire l'accesso non autorizzato.

Proprietà di un cluster di connessione

Questa sezione descrive le proprietà di un cluster Connect.

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 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 di sink in esecuzione sul cluster di connessione.

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.

Dopo aver creato il cluster Connect, non puoi eseguire l'aggiornamento a un cluster Kafka diverso.

Vantaggi della colocation delle regioni per la latenza e i costi di rete

La collocazione 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 di sink per scrivere i dati da questo cluster Managed Service per Apache Kafka (origine) a 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 BigQuery ed elimina i costi di trasferimento di rete tra regioni tra il servizio gestito per il cluster 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 (estrae i dati da un sistema) o da sink (inserisce i dati in un sistema).

Anche se un cluster Connect nella stessa regione del cluster Managed Service per Apache Kafka beneficia di una latenza di comunicazione inferiore, 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 posizioni 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 di connessione 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 per il 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 di connessione. Il valore minimo è 3 vCPU.

  • Memoria: la quantità di memoria assegnata a ogni vCPU. Devi eseguire il provisioning 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 la massima è 48 GiB (8 GiB per vCPU).

La vCPU e la memoria allocate a ogni worker in un cluster di connessione hanno un impatto significativo su prestazioni, capacità e costi del cluster. Ecco una suddivisione di come vCPU e memoria influiscono su un cluster di connessione.

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 porta a una velocità effettiva maggiore.

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

Memoria

  • Kafka Connect utilizza la memoria per memorizzare i dati nel buffer mentre scorrono tra i connettori e Managed Service per Apache Kafka. Una memoria più grande consente buffer più grandi. Una memoria di grandi dimensioni può migliorare il throughput, soprattutto per i flussi di dati di volumi elevati. I connettori che gestiscono messaggi o record molto grandi richiedono memoria sufficiente per elaborarli senza generare eccezioni OutOfMemoryError.

  • Più memoria aumenta il costo del cluster Connect.

  • Se utilizzi una logica di trasformazione complessa, è necessaria una maggiore allocazione di memoria.

Il tuo obiettivo è scegliere la configurazione della capacità corretta per il 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 self-hosted.

Di seguito sono riportati alcuni requisiti per la configurazione della subnet 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 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 rientrare nell'intervallo CIDR 172.16.0.0/14. Questo intervallo è riservato per l'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 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 da quello della subnet worker, devi assicurarti di utilizzare una configurazione consumer supportata (ad esempio, NCC). Per saperne di più, vedi 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 al VPC tenant. In questo modo, il cluster di connessione 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 devi configurare il nome di dominio DNS per il cluster Managed Service per Apache Kafka principale, poiché il relativo indirizzo di bootstrap viene incluso automaticamente nell'elenco dei domini DNS risolvibili.

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

Risorse di Secret Manager

Specifica Secret Manager da caricare nei worker. Questi secret vengono archiviati in modo sicuro in Secret Manager e resi disponibili per il cluster di connessione.

Puoi utilizzare Secret Manager nelle configurazioni dei connettori, se vuoi. Ad esempio, puoi caricare un file di chiavi nel cluster Connect e fare in modo che il connettore lo legga. I Secret Manager vengono montati come file nei worker.

I cluster di connessione si integrano direttamente con Secret Manager. Devi utilizzare Secret Manager per archiviare e gestire i secret.

Il formato per specificare un secret è: projects/{PROJECT_ID}/secrets/{SECRET_NAME}/versions/{VERSION_ID}

  • PROJECT_ID: l'ID del progetto in cui si trova il secret di Secret Manager.

  • SECRET_NAME: il nome del secret in Secret Manager.

  • VERSION_ID: il numero di versione specifico del secret. Si tratta di un numero come "1", "2", "3".

Puoi caricare fino a 32 secret in un singolo cluster Connect.

Assicurati che l'agente di servizio che esegue i worker di Connect disponga del ruolo secretmanager.secretAccessor (Funzione di accesso di Secret Manager) sui secret che vuoi utilizzare. Questo ruolo consente al cluster di connessione di recuperare i valori dei secret da Secret Manager.

Etichette

Le etichette sono coppie chiave-valore che ti aiutano con l'organizzazione e l'identificazione. Ti aiutano a organizzare i cluster di Connect. Puoi collegare un'etichetta a ogni cluster Connect, quindi filtrare le risorse in base alle 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 connessione.

La creazione di un cluster di connessione richiede 20-30 minuti.

Console

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

    Vai a Connetti cluster

  2. Fai clic su Crea.

    Si apre la pagina Crea un cluster di connessione.

  3. Per il nome del cluster di connessione, inserisci una stringa.

    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.

  4. Per Cluster Kafka principale, seleziona un cluster Managed Service per Apache Kafka dal menu.

    Per maggiori informazioni sulle funzioni eseguite da questo cluster Managed Service per Apache Kafka, consulta Cluster Kafka primario.

  5. Per Posizione, seleziona una posizione supportata dal menu Regione o mantieni il valore predefinito.

    Per saperne di più su come selezionare la posizione giusta, consulta la sezione Cluster Kafka principale.

  6. Per la configurazione della capacità, inserisci i valori per vCPU e Memoria o mantieni i valori predefiniti.

    Per vCPUs, inserisci il numero di CPU virtuali per il cluster.

    Per Memoria, inserisci la quantità di memoria per CPU in GiB. Se la memoria per CPU è superiore a 8 GiB, viene visualizzato un messaggio di errore.

    Per maggiori informazioni su come dimensionare un cluster Managed Service per Apache Kafka, consulta Configurazione della capacità.

  7. Per Configurazione di rete, seleziona o mantieni la rete del cluster Managed Service per Apache Kafka primario dal menu Rete.

  8. Per Subnet worker, seleziona o mantieni la subnet dal menu.

    Il campo Percorso URI subnet viene compilato automaticamente. Per saperne di più, vedi Subnet worker.

  9. Per i domini DNS risolvibili, il dominio DNS del cluster Kafka principale viene aggiunto automaticamente come dominio DNS risolvibile.

    Per aggiungere altri domini DNS, espandi la sezione, se necessario.

  10. Fai clic su Aggiungi un dominio DNS.

    Seleziona un cluster Kafka dal menu.

    Il dominio DNS viene compilato automaticamente. Puoi anche digitare il nome di dominio DNS per un cluster Kafka esterno.

    Fai clic su Fine.

  11. Per le risorse di Secret Manager, espandi la sezione, se necessario.

  12. Fai clic su Aggiungi risorsa secret.

  13. Seleziona un secret dal menu Secret e una versione dal menu Versione secret. Puoi anche creare un nuovo secret.

    Assicurati che l'agente di servizio che esegue i worker di Connect disponga del ruolo Funzione di accesso ai secret di Secret Manager sui secret che vuoi utilizzare. Per ulteriori informazioni su Secret Manager, consulta Risorse di Secret Manager.

  14. Fai clic su Fine.

  15. Fai clic su Aggiungi risorsa secret se devi aggiungere altri secret.

  16. Per Etichette, espandi la sezione se necessario.

    Per organizzare il progetto, aggiungi etichette arbitrarie alle risorse sotto forma di coppie chiave/valore.

    Fai clic su Aggiungi etichetta per includere diversi ambienti, servizi, proprietari, team e così via.

  17. Fai clic su Crea.

gcloud

  1. In the Google Cloud console, activate Cloud Shell.

    Activate Cloud Shell

    At the bottom of the Google Cloud console, a Cloud Shell session starts and displays a command-line prompt. Cloud Shell is a shell environment with the Google Cloud CLI already installed and with values already set for your current project. It can take a few seconds for the session to initialize.

  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 posizione 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 la località, consulta Cluster Kafka principale.

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

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

    • WORKER_SUBNET: La subnet del worker 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 Google Cloud progetto. Se non 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 di connessione. Vedi Cluster Kafka. Il formato del cluster Kafka è projects/PROJECT_ID/locations/LOCATION/clusters/CLUSTER_ID.

      Dopo aver creato il cluster Connect, non puoi eseguire l'aggiornamento a un cluster Kafka diverso.

    • SECRET: (facoltativo) secret da caricare nei worker. Devono essere fornite le versioni esatte dei secret di Secret Manager, gli alias non sono supportati. È possibile caricare fino a 32 secret in un cluster. 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 di connessione. Il cluster Connect può accedere alle risorse utilizzando i nomi di dominio anziché basarsi sugli indirizzi IP. Consulta Peering DNS.

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

    • CONFIG_FILE: (facoltativo) il percorso del file JSON o YAML contenente la configurazione sottoposta a override rispetto ai valori predefiniti del cluster o del connettore. Questo file supporta anche JSON o YAML incorporati.

    • --async: (facoltativo) restituisce immediatamente il risultato, 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 che il cluster sia completamente aggiornato 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 OPERATION_ID per monitorare i progressi. Ad esempio, il valore qui è operation-1753590328249-63ae19098cc06-64300a0a-06512d02.

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

    Go

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