Connect-Cluster erstellen

Ein Connect-Cluster bietet eine Umgebung für Connectors, mit denen Daten aus vorhandenen Kafka-Bereitstellungen in einen Google Cloud Managed Service for Apache Kafka-Cluster oder aus einem Managed Service for Apache Kafka-Cluster in einen anderen Google Cloud Dienst oder einen anderen Kafka-Cluster übertragen werden können. Der sekundäre Kafka-Cluster kann ein weiterer Google Cloud Managed Service for Apache Kafka-Cluster, ein selbst verwalteter oder ein lokaler Cluster sein.

Hinweis

Sie haben bereits einen Managed Service for Apache Kafka-Cluster erstellt. Sie benötigen den Namen des Managed Service for Apache Kafka-Clusters, an den der Connect-Cluster angehängt werden soll.

Jeder Connect-Cluster ist mit einem Managed Service for Apache Kafka-Cluster verknüpft. In diesem Cluster wird der Status der Connectors gespeichert, die im Connect-Cluster ausgeführt werden.

Erforderliche Rollen und Berechtigungen zum Erstellen eines Connect-Clusters

Bitten Sie Ihren Administrator, Ihnen die IAM-Rolle Managed Kafka Connect Cluster Editor (roles/managedkafka.connectClusterEditor) für Ihr Projekt zuzuweisen, um die Berechtigungen zu erhalten, die Sie zum Erstellen eines Connect-Clusters benötigen. Weitere Informationen zum Zuweisen von Rollen finden Sie unter Zugriff auf Projekte, Ordner und Organisationen verwalten.

Diese vordefinierte Rolle enthält die Berechtigungen, die zum Erstellen eines Connect-Clusters erforderlich sind. Maximieren Sie den Abschnitt Erforderliche Berechtigungen, um die notwendigen Berechtigungen anzuzeigen:

Erforderliche Berechtigungen

Die folgenden Berechtigungen sind zum Erstellen eines Connect-Clusters erforderlich:

  • Erteilen Sie die Berechtigung zum Erstellen eines Connect-Clusters am angegebenen Standort: managedkafka.connectClusters.create

Sie können diese Berechtigungen auch mit benutzerdefinierten Rollen oder anderen vordefinierten Rollen erhalten.

Erforderliche ACL-Principals

Standardmäßig können Connect-Cluster in Managed Service for Apache Kafka-Clustern auf Ressourcen zugreifen, wenn keine ACLs konfiguriert sind. Dazu wird allow.everyone.if.no.acl.found auf true gesetzt, was die Standardeinstellung ist.

Wenn für den Managed Service for Apache Kafka-Cluster jedoch ACLs konfiguriert sind, hat der Connect-Cluster nicht automatisch die Lese- und Schreibberechtigungen für die Ressourcen. Sie müssen sie manuell gewähren.

Das Connect-Cluster-Dienstkonto, das als Hauptkonto in ACLs verwendet wird, hat das folgende Format: User:service-{consumer project number}@gcp-sa-managedkafka.iam.gserviceaccount.com.

Wenn Sie ACLs für Ihren Kafka-Cluster konfiguriert haben, gewähren Sie dem Connect-Cluster mit den folgenden Befehlen Lese- und Schreibberechtigungen für Themen und Leseberechtigungen für Nutzergruppen:

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

Weitere Informationen zu diesen Befehlen finden Sie unter Apache Kafka-ACLs für eine detaillierte Zugriffssteuerung konfigurieren.

Connect-Cluster in einem anderen Projekt erstellen

Managed Service for Apache Kafka verwendet einen Dienst-Agent, um aufGoogle Cloud -Ressourcen zuzugreifen. Der Dienst-Agent ist dem Projekt zugeordnet, in dem Sie den Cluster erstellen.

Wenn Sie einen Connect-Cluster in einem anderen Projekt als den Managed Service for Apache Kafka-Cluster erstellen, verwenden der Connect-Cluster und der Kafka-Cluster die Dienst-Agents, die mit den jeweiligen Projekten verknüpft sind. In diesem Fall benötigt der Dienst-Agent für den Connect-Cluster die Berechtigung, auf Google Cloud Ressourcen im Projekt des Kafka-Clusters zuzugreifen.

Weisen Sie dem Dienst-Agent des Connect-Clusters die Rolle Managed Kafka Service Agent für das Projekt des Kafka-Clusters zu, um die erforderlichen Berechtigungen zu erteilen. Wenn Sie beispielsweise einen Kafka-Cluster im Projekt kafka-project und einen Connect-Cluster im Projekt connect-project erstellen, gewähren Sie dem Dienst-Agent, der mit connect-project verknüpft ist, die Rolle „Managed Kafka Service Agent“ für kafka-project.

Die E-Mail-Adresse des Dienst-Agents hat das folgende Format: service-PROJECT_NUMBER@gcp-sa-managedkafka.iam.gserviceaccount.com, wobei PROJECT_NUMBER die Projektnummer ist. Weitere Informationen zum Zuweisen der Rolle finden Sie unter Rollen erstellen und ihnen Dienst-Agents zuweisen.

Attribute eines Connect-Clusters

In diesem Abschnitt werden die Eigenschaften eines Connect-Clusters beschrieben.

Name des Connect-Clusters

Der Name des Connect-Clusters, den Sie erstellen. Tipps zum Benennen von Connect-Clustern finden Sie in den Richtlinien zum Benennen einer Ressource von Managed Service for Apache Kafka. Der Name eines Clusters kann nicht geändert werden.

Primärer Kafka-Cluster

Der Managed Service for Apache Kafka-Cluster, der Ihrem Connect-Cluster zugeordnet ist. In diesem zugehörigen Cluster (primärer Cluster) wird der Status der Connectors gespeichert, die im Connect-Cluster ausgeführt werden. Im Allgemeinen dient der primäre Managed Service for Apache Kafka-Cluster auch als Ziel für alle Quell-Connectors und als Eingabe für alle Senken-Connectors, die im Connect-Cluster ausgeführt werden.

Ein einzelner Managed Service for Apache Kafka-Cluster kann mehrere Connect-Cluster haben. Wenn Sie einen Managed Service for Apache Kafka-Cluster in einem anderen Projekt auswählen, müssen Sie die entsprechenden Berechtigungen konfigurieren.

Nachdem Sie den Connect-Cluster erstellt haben, können Sie ihn nicht mehr auf einen anderen Kafka-Cluster aktualisieren.

Vorteile von Colocations in Regionen in Bezug auf Latenz und Netzwerkkosten

Wenn Sie Ihre Managed Service for Apache Kafka- und Connect-Cluster in derselben Region platzieren, werden Latenz und Netzwerkkosten reduziert. Angenommen, Ihr Managed Service for Apache Kafka-Cluster befindet sich in region-a und Sie verwenden einen Senken-Connector, um Daten aus diesem Managed Service for Apache Kafka-Cluster (Quelle) in eine BigQuery-Tabelle (Senke) zu schreiben, die sich ebenfalls in region-a befindet. Wenn Sie Ihren Connect-Cluster in region-a bereitstellen, wird durch diese Bereitstellung die Latenz für den BigQuery-Schreibvorgang minimiert und es fallen keine Kosten für die regionsübergreifende Netzwerkübertragung zwischen dem Managed Service for Apache Kafka-Cluster und dem Connect-Cluster an.

Überlegungen zu Latenz und Kosten bei mehreren Systemen

Kafka Connect verwendet Connectors, um Daten zwischen Systemen zu übertragen. Eine Seite des Connectors interagiert immer mit einem Managed Service for Apache Kafka-Cluster. In einem einzelnen Kafka Connect-Cluster können mehrere Connectors ausgeführt werden, die jeweils entweder als Quelle (Daten aus einem System abrufen) oder als Senke (Daten in ein System übertragen) fungieren.

Ein Connect-Cluster in derselben Region wie der Managed Service for Apache Kafka-Cluster profitiert zwar von einer geringeren Kommunikationslatenz zwischen den beiden Clustern, jeder Connector interagiert aber auch mit einem anderen System, z. B. einer BigQuery-Tabelle oder einem anderen Kafka-Cluster. Auch wenn sich der Connect-Cluster und der Managed Service for Apache Kafka-Cluster am selben Standort befinden, kann sich das andere System in einer anderen Region befinden. Das führt zu einer höheren Latenz und höheren Kosten. Die Gesamtlatenz der Pipeline hängt von den Standorten aller drei Systeme ab: des Managed Service for Apache Kafka-Clusters, des Connect-Clusters und des Quell- oder Zielsystems.

Wenn sich Ihr Managed Service for Apache Kafka-Cluster beispielsweise in region-a, Ihr Connect-Cluster in region-b und Sie einen Cloud Storage-Connector für einen Bucket in region-c verwenden, werden Ihnen zwei Netzwerk-Hops in Rechnung gestellt (region-a nach region-b und dann region-b nach region-c oder umgekehrt, je nach Connector-Richtung).

Berücksichtigen Sie bei der Planung der Platzierung Ihres Connect-Clusters alle beteiligten Regionen, um sowohl die Latenz als auch die Kosten zu optimieren.

Kapazitätskonfiguration

Im Rahmen der Kapazitätskonfiguration müssen Sie die Anzahl der vCPUs und die Größe des Arbeitsspeichers für jede vCPU für Ihren Connect-Cluster festlegen. Sie können die Kapazität eines Connect-Clusters nach dem Erstellen aktualisieren. Die Eigenschaften für die Kapazitätskonfiguration sind:

  • vCPUs: Die Anzahl der vCPUs, die einem Connect-Cluster zugewiesen sind. Der Mindestwert beträgt 3 vCPUs.

  • Arbeitsspeicher: Die Menge an Arbeitsspeicher, die jeder vCPU zugewiesen ist. Sie müssen zwischen 1 GiB und 8 GiB pro vCPU bereitstellen. Die Menge an Arbeitsspeicher kann nach der Clustererstellung innerhalb dieser Grenzen erhöht oder verringert werden.

    Wenn Sie beispielsweise einen Cluster mit 6 vCPUs erstellen, beträgt der minimale Arbeitsspeicher, den Sie dem Cluster zuweisen können, 6 GiB (1 GiB pro vCPU) und der maximale 48 GiB (8 GiB pro vCPU).

Die vCPU und der Arbeitsspeicher, die jedem Worker in einem Connect-Cluster zugewiesen werden, haben einen erheblichen Einfluss auf die Leistung, Kapazität und Kosten des Clusters. Hier finden Sie eine Aufschlüsselung der Auswirkungen von vCPU und Arbeitsspeicher auf einen Connect-Cluster.

vCPU Anzahl

  • Kafka Connect unterteilt die Arbeit eines Connectors in Aufgaben. Jede Aufgabe kann Daten parallel verarbeiten. Mehr vCPUs bedeuten, dass mehr Aufgaben gleichzeitig ausgeführt werden können, was zu einem höheren Durchsatz führt.

  • Durch mehr vCPUs steigen die Kosten für Ihren Connect-Cluster.

Arbeitsspeicher

  • Kafka Connect verwendet Arbeitsspeicher zum Puffern von Daten, während sie zwischen Connectors und Managed Service for Apache Kafka übertragen werden. Ein größerer Arbeitsspeicher ermöglicht größere Puffer. Ein großer Arbeitsspeicher kann den Durchsatz verbessern, insbesondere bei Datenstreams mit hohem Volumen. Für Connectors, die sehr große Nachrichten oder Datensätze verarbeiten, ist ausreichend Arbeitsspeicher erforderlich, damit keine OutOfMemoryError-Ausnahmen auftreten.

  • Mehr Arbeitsspeicher erhöht die Kosten für Ihren Connect-Cluster.

  • Wenn Sie eine komplexe Transformationslogik verwenden, benötigen Sie mehr Arbeitsspeicher.

Ihr Ziel ist es, die richtige Kapazitätskonfiguration für Ihren Connect-Cluster auszuwählen. Dazu müssen Sie den Durchsatz kennen, den Ihr Connect-Cluster verarbeiten kann.

Worker-Subnetz (primär)

Das Worker-Subnetz, auch primäres Subnetz genannt, verbindet Ihr VPC-Netzwerk mit dem Connect-Cluster. Über dieses Subnetz können die Cluster-Worker die Endpunkte von Quellen und Senken im Verbrauchernetzwerk erreichen, z. B. Managed Service for Apache Kafka-Cluster oder selbst gehostete Kafka-Cluster.

Hier sind einige Anforderungen für die Konfiguration des Worker-Subnetzes:

  • Das Worker-Subnetzwerk ist erforderlich.

  • Das Subnetz muss sich in derselben Region wie der Connect-Cluster befinden.

  • Das Subnetz muss sich in derselben übergeordneten VPC wie eines der verbundenen Subnetze des primären Kafka-Clusters befinden.

  • Der CIDR-Bereich des Subnetzes muss mindestens die Größe /22 (1.024 Adressen) haben.

Den Cluster-Workern werden IP-Adressen im Worker-Subnetz zugewiesen. Dazu wird eine Private Service Connect-Schnittstelle verwendet. Worker können jedes Netzwerkziel erreichen, das über das VPC-Netzwerk des Subnetzes zugänglich ist. Dabei gelten die folgenden Anforderungen:

  • Der Endpunkt darf sich nicht im CIDR-Bereich 172.16.0.0/14 befinden. Dieser Bereich ist für die interne Verwendung von Managed Service for Apache Kafka Connect reserviert.
  • Firewallregeln müssen den Traffic zulassen. Weitere Informationen finden Sie unter Sicherheit für Netzwerkanhänge konfigurieren.
  • Für Internet-Traffic müssen Sie Cloud NAT konfigurieren. Cloud NAT ist beispielsweise erforderlich, damit ein MirrorMaker-Connector Daten aus einem Kafka-Cluster replizieren kann, auf den über das Internet zugegriffen werden kann.
  • Wenn Sie auf Private Service Connect-Endpunkte zugreifen möchten, die sich in einem anderen VPC als die VPC des Worker-Subnetzes befinden, müssen Sie eine unterstützte Nutzerkonfiguration verwenden (z. B. NCC). Weitere Informationen finden Sie unter Zugriff auf veröffentlichte Dienste über Endpunkte.

Auflösbare DNS-Domains

Auflösbare DNS-Domains, auch als DNS-Domainnamen bezeichnet, ermöglichen es, DNS-Adressen im VPC-Netzwerk für die Tenant-VPC verfügbar zu machen. So kann der Connect-Cluster DNS-Namen in IP-Adressen auflösen und die Kommunikation mit anderen Diensten erleichtern, einschließlich anderer Kafka-Cluster für MirrorMaker-Connectors.

Für auflösbare DNS-Domains können Sie einen Managed Service for Apache Kafka-Cluster auswählen. Sie müssen den DNS-Domainnamen für den primären Managed Service for Apache Kafka-Cluster nicht konfigurieren, da seine Bootstrap-Adresse automatisch in die Liste der auflösbaren DNS-Domains aufgenommen wird.

Sie können aber auch manuell eine DNS-Domain angeben. Das ist erforderlich, wenn Sie einen externen Kafka-Cluster auswählen. Die DNS-Domain des primären Managed Service for Apache Kafka-Clusters wird automatisch hinzugefügt. Für andere Kafka-Cluster müssen weiterhin DNS-Domains konfiguriert werden.

Secret Manager-Ressourcen

Für einige Connectors sind bei der Konfiguration vertrauliche Daten wie Passwörter erforderlich. Um diese Art von Daten sicher zu verwalten, können Sie sie in Secret Manager speichern und dem Connect-Cluster Zugriff auf das Secret gewähren.

So verwenden Sie Secret Manager-Secrets mit Kafka Connect:

  1. Weisen Sie dem Managed Kafka-Dienstkonto die Rolle Zugriffsperson für Secret Manager-Secret (roles/secretmanager.secretAccessor) zu. Mit dieser Rolle kann Ihr Connect-Cluster auf Secrets zugreifen.

  2. Erstellen Sie ein Secret, das die vertraulichen Daten enthält. Weitere Informationen finden Sie unter Secret erstellen.

  3. Geben Sie beim Erstellen oder Aktualisieren Ihres Connect-Clusters die Secrets an, auf die der Cluster Zugriff hat. Sie können bis zu 32 Secrets pro Connect-Cluster angeben.

Secrets werden als Dateien in den Cluster-Workern bereitgestellt. Connectors haben Lesezugriff auf diese Dateien. Wenn Sie einen Connector erstellen, können die Konfigurationseigenschaften des Connectors auf die Secrets verweisen.

  • Verwenden Sie das folgende Format, um auf den Pfad zu einer Secret-Datei zu verweisen:

    /var/secrets/PROJECT_NAME-SECRET_NAME-SECRET_VERSION

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

  • Wenn Sie den Wert eines Secrets als Konfigurationswert (z. B. ein Passwort) verwenden möchten, verwenden Sie das folgende Format:

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

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

Ersetzen Sie Folgendes:

  • PROJECT_NAME: Der Name des Google Cloud -Projekts.
  • SECRET_NAME: Der Name des Secrets.
  • SECRET_VERSION: Die Secret-Version.

Labels

Labels sind Schlüssel/Wert-Paare, die Ihnen bei der Organisation und Identifizierung helfen. Sie helfen Ihnen beim Organisieren von Connect-Clustern. Sie können jedem Connect-Cluster ein Label zuweisen und dann die Ressourcen nach Labels filtern. Beispiele für Labels sind environment:prod und application:web-app.

Connect-Cluster erstellen

Bevor Sie einen Cluster erstellen, lesen Sie die Dokumentation zu Connect-Clustereigenschaften.

Das Erstellen eines Connect-Clusters dauert 20 bis 30 Minuten.

Console

  1. Rufen Sie in der Google Cloud Console die Seite Connect Clusters auf.

    Zu „Cluster verbinden“

  2. Klicken Sie auf Erstellen.

  3. Geben Sie im Feld Connect-Clustername einen Namen für den Connect-Cluster ein. Weitere Informationen finden Sie unter Richtlinien zum Benennen einer Managed Service for Apache Kafka-Ressource.

  4. Wählen Sie in der Liste Primärer Kafka-Cluster einen Managed Service for Apache Kafka-Cluster aus. Weitere Informationen finden Sie unter Primärer Kafka-Cluster.

  5. Wählen Sie in der Liste Region einen Standort für den Connect-Cluster aus. Weitere Informationen zum Auswählen eines Standorts finden Sie unter Primärer Kafka-Cluster.

  6. Geben Sie im Abschnitt Kapazitätskonfiguration Werte für die folgenden Felder ein oder behalten Sie die Standardwerte bei.

    • Geben Sie im Feld vCPUs die Anzahl der virtuellen CPUs für den Cluster ein.

    • Geben Sie im Feld Arbeitsspeicher die Menge an Arbeitsspeicher pro CPU in GiB ein. Der Wert darf 8 GiB pro CPU nicht überschreiten.

    Weitere Informationen finden Sie unter Kapazitätskonfiguration.

  7. Wählen Sie im Bereich Netzwerkkonfiguration in der Liste Netzwerk ein VPC-Netzwerk aus oder behalten Sie den Standardwert bei. Diese Liste wird gefüllt, wenn Sie den primären Kafka-Cluster auswählen.

  8. Wählen Sie im Bereich Worker-Subnetz ein Subnetz aus der Liste Subnetz aus oder behalten Sie den Standardwert bei. Weitere Informationen finden Sie unter Worker-Subnetz. Das Feld Subnetz-URI-Pfad wird automatisch ausgefüllt, wenn Sie das Subnetz auswählen.

  9. Optional: Fügen Sie eine auflösbare DNS-Domain hinzu. Die DNS-Domain des primären Kafka-Clusters wird automatisch als auflösbare DNS-Domain hinzugefügt. So geben Sie zusätzliche DNS-Domains an:

    1. Maximieren Sie den Bereich Auflösbare DNS-Domains.

    2. Klicken Sie auf DNS-Domain hinzufügen.

    3. Wenn Sie die DNS-Domain eines vorhandenen Managed Service for Apache Kafka-Clusters hinzufügen möchten, wählen Sie den Cluster aus der Liste Kafka-Cluster aus. Geben Sie andernfalls die DNS-Domain in das Feld DNS-Domain ein.

    4. Klicken Sie auf Fertig.

  10. Optional: So fügen Sie Secret Manager-Ressourcen hinzu:

    1. Maximieren Sie den Bereich Secret Manager-Ressourcen.

    2. Klicken Sie auf Secret-Ressource hinzufügen.

    3. Wählen Sie in der Liste Secret ein Secret aus.

    4. Wählen Sie in der Liste Secret-Version eine Version des Secrets aus.

    5. Klicken Sie auf Fertig.

  11. Optional: Fügen Sie Labels hinzu, um Ihr Projekt zu organisieren. So fügen Sie ein Label hinzu:

    1. Erweitern Sie den Abschnitt Labels.

    2. Klicken Sie auf Label hinzufügen.

    3. Geben Sie im Feld Schlüssel den Schlüssel des Labels ein.

    4. Geben Sie im Feld Wert den Wert des Labels ein.

  12. Klicken Sie auf Erstellen.

gcloud

  1. Aktivieren Sie Cloud Shell in der Google Cloud Console.

    Cloud Shell aktivieren

    Unten in der Google Cloud Console wird eine Cloud Shell-Sitzung gestartet und eine Eingabeaufforderung angezeigt. Cloud Shell ist eine Shell-Umgebung, in der das Google Cloud CLI bereits installiert ist und Werte für Ihr aktuelles Projekt bereits festgelegt sind. Das Initialisieren der Sitzung kann einige Sekunden dauern.

  2. Führen Sie den Befehl gcloud managed-kafka connect-clusters create aus:

    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]

    Ersetzen Sie Folgendes:

    • CONNECT_CLUSTER_ID: Die ID oder der Name des Connect-Clusters. Tipps zum Benennen von Connect-Clustern finden Sie in den Richtlinien zum Benennen einer Ressource von Managed Service for Apache Kafka. Der Name eines Connect-Clusters ist unveränderlich.

    • LOCATION: Der Ort, an dem Sie den Connect-Cluster erstellen. Dies muss eine unterstützte Google CloudRegion sein. Sie können den Standort eines Connect-Clusters nach der Erstellung nicht mehr ändern. Eine Liste der verfügbaren Standorte finden Sie unter Managed Service for Apache Kafka-Standorte. Weitere Informationen zu Standortempfehlungen finden Sie unter Primärer Kafka-Cluster.

    • CPU: Die Anzahl der vCPUs für den Connect-Cluster. Der Mindestwert beträgt 3 vCPUs. Weitere Informationen finden Sie unter vCPU-Anzahl.

    • MEMORY: Die Größe des Arbeitsspeichers für den Connect-Cluster. Verwenden Sie die Einheiten „MB“, „MiB“, „GB“, „GiB“, „TB“ oder „TiB“. Beispiel: „3GiB“ Sie müssen zwischen 1 GiB und 8 GiB pro vCPU bereitstellen. Weitere Informationen finden Sie unter Arbeitsspeicher.

    • WORKER_SUBNET: Das Worker-Subnetz für den Connect-Cluster.

      Das Format des Subnetzes ist projects/PROJECT_ID/regions/REGION/subnetworks/SUBNET_ID.

      Das Worker-Subnetz muss sich in derselben Region wie der Connect-Cluster befinden.

    • PROJECT_ID: (Optional) Die ID desGoogle Cloud -Projekts. Wenn nichts angegeben ist, wird das aktuelle Projekt verwendet.

    • KAFKA_CLUSTER: Die ID oder der voll qualifizierte Name des primären Managed Service for Apache Kafka-Clusters, das dem Connect-Cluster zugeordnet ist. Weitere Informationen finden Sie unter Kafka-Cluster. Das Format des Kafka-Clusters ist projects/PROJECT_ID/locations/LOCATION/clusters/CLUSTER_ID.

      Nachdem Sie den Connect-Cluster erstellt haben, können Sie ihn nicht mehr auf einen anderen Kafka-Cluster aktualisieren.

    • SECRET: (Optional) Secrets, die in Worker geladen werden sollen. Es müssen genaue Secret-Versionen aus Secret Manager angegeben werden. Aliase werden nicht unterstützt. In einen Cluster können bis zu 32 Secrets geladen werden. Format: projects/PROJECT_ID/secrets/SECRET_NAME/versions/VERSION_ID

    • DNS_DOMAIN_NAME: (Optional) DNS-Domainnamen aus dem Subnetz, die für den Connect-Cluster sichtbar gemacht werden sollen. Der Connect-Cluster kann über Domainnamen auf Ressourcen zugreifen, anstatt auf IP-Adressen angewiesen zu sein. Weitere Informationen finden Sie unter DNS-Peering.

    • LABELS: (Optional) Labels, die dem Cluster zugeordnet werden sollen. Weitere Informationen zum Format von Labels finden Sie unter Labels. Liste der hinzuzufügenden KEY=VALUE-Labelpaare. Schlüssel müssen mit einem Kleinbuchstaben beginnen und dürfen nur Bindestriche (-), Unterstriche (_), Kleinbuchstaben und Zahlen enthalten. Werte dürfen nur Bindestriche (-), Unterstriche (_), Kleinbuchstaben und Zahlen enthalten.

    • CONFIG_FILE: (Optional) Der Pfad zur JSON- oder YAML-Datei, die die Konfiguration enthält, die von den Cluster- oder Connector-Standardeinstellungen überschrieben wird. Diese Datei unterstützt auch Inline-JSON oder -YAML.

    • --async: (Optional) Gibt die Steuerung sofort an das Terminal zurück, ohne auf den Abschluss des Vorgangs zu warten. Mit dem Flag --async können Sie mit anderen Aufgaben fortfahren, während der Cluster im Hintergrund erstellt wird. Wenn Sie das Flag nicht verwenden, wartet das System, bis der Vorgang abgeschlossen ist, bevor es eine Antwort zurückgibt. Sie müssen warten, bis der Cluster vollständig aktualisiert wurde, bevor Sie mit anderen Aufgaben fortfahren können.

    Sie erhalten eine Antwort ähnlich der folgenden:

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

    Speichere die OPERATION_ID, um den Fortschritt zu verfolgen. In diesem Beispiel ist der Wert operation-1753590328249-63ae19098cc06-64300a0a-06512d02.

Terraform

Sie können eine Terraform-Ressource verwenden, um einen Connect-Cluster zu erstellen.

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

Informationen zum Anwenden oder Entfernen einer Terraform-Konfiguration finden Sie unter Grundlegende Terraform-Befehle.

Go

Bevor Sie dieses Beispiel ausprobieren, folgen Sie der Einrichtungsanleitung für Go unter Clientbibliotheken installieren. Weitere Informationen finden Sie in der Referenzdokumentation zur Managed Service for Apache Kafka Go API.

Richten Sie zur Authentifizierung bei Managed Service for Apache Kafka die Standardanmeldedaten für Anwendungen(Application Default Credentials, ADC) ein. Weitere Informationen finden Sie unter ADC für eine lokale Entwicklungsumgebung einrichten.

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

Bevor Sie dieses Beispiel ausprobieren, folgen Sie der Anleitung zur Einrichtung von Java unter Clientbibliotheken installieren. Weitere Informationen finden Sie in der Referenzdokumentation zur Java API für Managed Service for Apache Kafka.

Richten Sie die Standardanmeldedaten für Anwendungen ein, um sich bei Managed Service for Apache Kafka zu authentifizieren. Weitere Informationen finden Sie unter ADC für eine lokale Entwicklungsumgebung einrichten.


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

Bevor Sie dieses Beispiel ausprobieren, folgen Sie der Anleitung für die Einrichtung von Python unter Clientbibliotheken installieren. Weitere Informationen finden Sie in der Referenzdokumentation zur Python API für Managed Service for Apache Kafka.

Richten Sie die Standardanmeldedaten für Anwendungen ein, um sich bei Managed Service for Apache Kafka zu authentifizieren. Weitere Informationen finden Sie unter ADC für eine lokale Entwicklungsumgebung einrichten.

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

Vorgang zur Clustererstellung überwachen

Sie können den folgenden Befehl nur ausführen, wenn Sie die gcloud CLI zum Erstellen des Connect-Clusters verwendet haben.

  • Das Erstellen eines Connect-Clusters dauert in der Regel 20 bis 30 Minuten. Um den Fortschritt der Clustererstellung zu verfolgen, verwendet der Befehl gcloud managed-kafka connect-clusters create einen Vorgang mit langer Ausführungszeit (Long-Running Operation, LRO), den Sie mit dem folgenden Befehl überwachen können:

    gcloud managed-kafka operations describe OPERATION_ID \
    --location=LOCATION

    Ersetzen Sie Folgendes:

    • OPERATION_ID mit dem Wert der Vorgangs-ID aus dem vorherigen Abschnitt.
    • LOCATION durch den Wert des Standorts aus dem vorherigen Abschnitt.

Nächste Schritte

Apache Kafka® ist eine eingetragene Marke der Apache Software Foundation oder ihrer Tochtergesellschaften in den USA und/oder anderen Ländern.