MirrorMaker 2.0-Quell-Connector erstellen

Mit einem MirrorMaker 2.0-Quell-Connector können Sie Themen und Daten von einem Kafka-Cluster (der Quelle) in einen anderen Kafka-Cluster (das Ziel) replizieren.

Typische Anwendungsfälle für diesen Connector:

  • Datenmigration Verschieben Sie Ihre Kafka-Arbeitslast in einen neuen Managed Service for Apache Kafka-Cluster.

  • Notfallwiederherstellung. Erstellen Sie einen Sicherungscluster, um die Geschäftskontinuität im Falle von Ausfällen zu gewährleisten.

  • Datenaggregation Daten aus mehreren Kafka-Clustern in einem zentralen Managed Service for Apache Kafka-Cluster konsolidieren, um Analysen durchzuführen.

Dieser Connector führt eine grundlegende Datenreplikation zwischen Kafka-Clustern durch. Verwenden Sie die folgenden Connectors in Verbindung mit dem MirrorMaker 2.0-Quell-Connector, um eine robustere Notfallwiederherstellung zu ermöglichen:

  • MirrorMaker 2.0 (Prüfpunkt)-Connector. Repliziert Consumer-Offsets, sodass Consumer ohne Datenverlust ein Failover auf den Zielcluster durchführen können.

  • MirrorMaker 2.0 (Heartbeat)-Connector. Generiert regelmäßige Heartbeat-Nachrichten, mit denen Sie die Verbindung und die Replikationsverzögerung überwachen können.

Clusterrollen in MirrorMaker 2.0

Ein MirrorMaker 2.0-Connector liest aus einem Kafka-Quellcluster und schreibt in einen Kafka-Zielcluster. In Managed Service for Apache Kafka muss der primäre Cluster des Connect-Clusters der Zielcluster sein. Diese Konfiguration minimiert die Latenz für Schreibvorgänge. Der Quellcluster kann ein anderer Managed Service for Apache Kafka-Cluster oder ein externer oder selbstverwalteter Kafka-Cluster sein.

Wenn Sie Daten von einem Managed Service for Apache Kafka-Cluster in einen externen oder selbstverwalteten Cluster replizieren möchten, empfehlen wir, Kafka Connect am selben Standort oder in derselben Region wie den anderen Cluster auszuführen und MirrorMaker 2.0 zu verwenden, um die Daten aus Managed Service for Apache Kafka aufzunehmen.

Sie müssen alle Properties für den Connector richtig konfigurieren. Dazu gehören auch Producer-Authentifizierungsattribute, die an den sekundären Cluster gerichtet sind. Weitere Informationen zu potenziellen Problemen finden Sie unter MirrorMaker 2.0-Clientkonfiguration verbessern.

Erforderliche Rollen und Berechtigungen

Bitten Sie Ihren Administrator, Ihnen die IAM-Rolle Managed Kafka Connector Editor (roles/managedkafka.connectorEditor) für Ihr Projekt zuzuweisen, um die Berechtigungen zu erhalten, die Sie zum Erstellen eines Connectors 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 Connectors erforderlich sind. Maximieren Sie den Abschnitt Erforderliche Berechtigungen, um die notwendigen Berechtigungen anzuzeigen:

Erforderliche Berechtigungen

Die folgenden Berechtigungen sind zum Erstellen eines Connectors erforderlich:

  • Connector erstellen: managedkafka.connectors.create

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

Netzwerk konfigurieren

In diesem Abschnitt werden die Netzwerkanforderungen für MirrorMaker 2.0 zum Herstellen einer Verbindung zum sekundären Kafka-Cluster beschrieben. Hier sind die allgemeinen Anforderungen:

  • Konnektivität: Die Connect-Cluster-Worker müssen eine Verbindung zum sekundären Kafka-Cluster herstellen können.

  • DNS-Auflösung Sie müssen die DNS-Domain des sekundären Clusters den auflösbaren DNS-Domains des Connect-Clusters hinzufügen. Außerdem muss die DNS-Domain über das Subnetz des Connect-Clusters auflösbar sein, z. B. durch Erstellen einer Cloud DNS-Zone.

  • Firewallregeln: Firewallregeln müssen es den Connect-Cluster-Workern ermöglichen, sowohl die Kafka-Quell- als auch die Kafka-Zielcluster zu erreichen.

Viele der Details hängen davon ab, wie und wo der sekundäre Kafka-Cluster gehostet wird, ob es sich um einen Managed Service for Apache Kafka-Cluster, einen selbstverwalteten Cluster in Google Cloudoder einen externen Cluster handelt.

  • Managed Service for Apache Kafka-Cluster

    • Wenn sich der sekundäre Cluster in einem anderen VPC-Netzwerk befindet, fügen Sie den verbundenen Subnetzen des sekundären Clusters ein Subnetz aus dem VPC-Netzwerk des Connect-Clusters hinzu. Weitere Informationen finden Sie unter Netzwerkkonfiguration.

  • Selbstverwalteter Kafka-Cluster in Google Cloud

  • Externer Kafka-Cluster

    • Folgende Optionen stehen für die Verbindung mit einem externen Cluster zur Verfügung:

      • Cloud VPN: Eine kostengünstige Lösung, die für geringere Bandbreiten geeignet ist. Er erstellt einen verschlüsselten Tunnel über das öffentliche Internet.

      • Cloud Interconnect: Bietet eine dedizierte Verbindung mit hoher Bandbreite zwischen Ihrem lokalen Netzwerk und Google Cloud. Sie können zwischen Dedicated Interconnect für eine direkte physische Verbindung oder Partner Interconnect für eine Verbindung über einen Dienstanbieter wählen.

    • Konfigurieren Sie Cloud NAT, damit die Connect-Worker auf das Internet zugreifen können.

    • Erstellen Sie im VPC-Netzwerk des Connect-Clusters eine Cloud DNS-Weiterleitungszone, um DNS-Abfragen für die Bootstrap-Adresse und die Broker-Endpunkte Ihres externen Clusters aufzulösen.

Authentifizierung konfigurieren

Wenn der Kafka-Quell- oder -Zielcluster ein Managed Service for Apache Kafka-Cluster ist, wird der Connector automatisch für die Verwendung der OAuthBearer-Authentifizierung konfiguriert. Sie müssen keine zusätzlichen Konfigurationen festlegen.

Für einen selbstverwalteten oder externen Kafka-Cluster müssen Sie die Authentifizierung und die TLS-Verschlüsselung konfigurieren. Die Konfigurationen hängen vom Authentifizierungsmechanismus ab, den der Kafka-Cluster unterstützt. Weitere Informationen finden Sie in der Apache Kafka-Dokumentation unter MirrorMaker Common Configs.

Wenn der Cluster die OAuthBearer-Authentifizierung unterstützt, verwenden Sie die folgenden Einstellungen:

source.cluster.security.protocol=SASL_SSL
source.cluster.sasl.mechanism=OAUTHBEARER
source.cluster.sasl.login.callback.handler.class=com.google.cloud.hosted.kafka.auth.GcpLoginCallbackHandler
source.cluster.sasl.jaas.config=org.apache.kafka.common.security.oauthbearer.OAuthBearerLoginModule required;

Speichern Sie die erforderlichen Anmeldedaten in Secret Manager, wie im nächsten Abschnitt beschrieben.

Secret Manager für Anmeldedaten verwenden

Wenn Sie eine Verbindung zu selbstverwalteten oder externen Kafka-Clustern herstellen, verwenden Sie Secret Manager, um vertrauliche Konfigurationswerte zu speichern, die für die Authentifizierung erforderlich sind. Diese Informationen können Passwörter, Schlüsselspeicherdateien und Trust Store-Dateien enthalten, je nachdem, wie die Authentifizierung im Cluster konfiguriert ist.

Weitere Informationen zum Verknüpfen von Secret Manager-Secrets mit einem Connect-Cluster finden Sie unter Secret Manager-Ressourcen.

Verweisen Sie in der Connector-Konfiguration so auf Secrets:

  • Verwenden Sie für Dateipfade das Format /var/secrets/PROJECT_NAME-SECRET_NAME-SECRET_VERSION. Beispiel: ssl.truststore.location=/var/secrets/project1-truststore-1.

  • Verwenden Sie für Passwörter das 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.

Funktionsweise eines MirrorMaker-Quellconnectors

Ein MirrorMaker-Quell-Connector ruft Daten aus einem oder mehreren Kafka-Themen in einem Quellcluster ab und repliziert diese Daten zusammen mit ACLs in Themen in einem Zielcluster.

Hier finden Sie eine detaillierte Aufschlüsselung, wie der MirrorMaker-Quell-Connector Daten repliziert:

  • Der Connector liest Nachrichten aus angegebenen Kafka-Themen im Quellcluster. Geben Sie die zu replizierenden Themen mit dem Konfigurationsattribut topics an, das kommagetrennte Themennamen oder einen einzelnen regulären Ausdruck im Java-Stil akzeptiert. Beispiel: topic-a,topic-b oder my-prefix-.*.

  • Der Connector kann auch die Replikation bestimmter Themen überspringen, die Sie mit dem Attribut topics.exclude angeben. Ausschlüsse überschreiben Einschlüsse.

  • Der Connector schreibt die konsumierten Nachrichten in den Zielcluster.

  • Der Connector benötigt die Verbindungsdetails für den Quell- und Zielcluster, z. B. source.cluster.bootstrap.servers und target.cluster.bootstrap.servers.

  • Der Connector benötigt außerdem Aliase für die Quell- und Zielcluster, wie durch source.cluster.alias und target.cluster.alias angegeben. Standardmäßig werden replizierte Themen automatisch mit dem Quellalias umbenannt. Beispiel: Ein Thema namens orders aus einer Quelle mit dem Alias primary wird im Ziel zu primary.orders. Wenn Sie die ursprünglichen Themennamen im Zielcluster ohne Präfix beibehalten möchten, setzen Sie replication.policy.class auf org.apache.kafka.connect.mirror.IdentityReplicationPolicy.

  • ACLs, die mit den replizierten Themen verknüpft sind, werden ebenfalls vom Quell- zum Zielcluster synchronisiert. Dies kann mit dem Attribut sync.topic.acls.enabled deaktiviert werden.

  • Wenn für die Cluster erforderlich, müssen in der Konfiguration Authentifizierungsdetails für die Verbindung mit dem Quell- und dem Zielcluster angegeben werden. Sie müssen Attribute wie security.protocol, sasl.mechanism und sasl.jaas.config konfigurieren, wobei source.cluster. für die Quelle und target.cluster. für das Ziel vorangestellt werden muss.

  • Der Connector verwendet interne Themen. Möglicherweise müssen Sie entsprechende Attribute konfigurieren, z. B. offset-syncs.topic.replication.factor.

  • Der Connector verwendet die Kafka-Datensatzkonverter key.converter, value.converter und header.converter. Bei der direkten Replikation wird standardmäßig org.apache.kafka.connect.converters.ByteArrayConverter verwendet, wobei keine Konvertierung (Pass-Through) erfolgt.

  • Mit dem Attribut tasks.max wird der Grad der Parallelität für den Connector gesteuert. Durch Erhöhen von tasks.max kann der Durchsatz potenziell verbessert werden. Die effektive Parallelität ist jedoch häufig durch die Anzahl der Partitionen in den replizierten Kafka-Quellthemen begrenzt.

Eigenschaften eines MirrorMaker 2.0-Connectors

Geben Sie beim Erstellen oder Aktualisieren eines MirrorMaker 2.0-Connectors die folgenden Attribute an:

Connector-Name

Der Name oder die ID des Connectors. Richtlinien zum Benennen der Ressource finden Sie unter Richtlinien zum Benennen einer Managed Service for Apache Kafka-Ressource. Der Name ist unveränderlich.

Connector-Typ

Der Connectortyp muss einer der folgenden sein:

Quellcluster

Der Kafka-Quellcluster, aus dem der MirrorMaker 2.0-Connector liest.

  • Managed Service for Apache Kafka-Cluster: Wählen Sie einen Cluster aus dem Drop-down-Menü aus.

  • Selbstverwalteter oder externer Kafka-Cluster: Geben Sie die Bootstrap-Adresse im Format hostname:port_number ein. Beispiel: kafka-test:9092.

Themennamen oder reguläre Ausdrücke

Die zu replizierenden Themen. Geben Sie einzelne Namen (topic1, topic2) an oder verwenden Sie einen regulären Ausdruck (topic.*). Dieses Attribut ist für den MirrorMaker 2.0-Quell-Connector erforderlich. Der Standardwert ist .*.

Namen von Nutzergruppen oder reguläre Ausdrücke

Die zu replizierenden Nutzergruppen. Geben Sie einzelne Namen (group1, group2) an oder verwenden Sie einen regulären Ausdruck (group.*). Diese Property ist für den MirrorMaker 2.0-Prüfpunkt-Connector erforderlich. Der Standardwert ist .*.

Konfiguration

In diesem Abschnitt können Sie zusätzliche, connectorspezifische Konfigurationseigenschaften für den MirrorMaker 2.0-Connector angeben.

Da Daten in Kafka-Themen in verschiedenen Formaten wie Avro, JSON oder Rohbytes vorliegen können, ist die Angabe von Konvertern ein wichtiger Teil der Konfiguration. Konverter übersetzen Daten aus dem in Ihren Kafka-Themen verwendeten Format in das standardisierte, interne Format von Kafka Connect.

Weitere allgemeine Informationen zur Rolle von Konvertern in Kafka Connect, zu unterstützten Konvertertypen und zu gängigen Konfigurationsoptionen finden Sie unter Konverter.

Einige gängige Konfigurationen für alle MirrorMaker 2.0-Connectors:

  • source.cluster.alias: Alias für den Quellcluster.

  • target.cluster.alias: Alias für den Zielcluster.

  • replication.policy.class: Richtlinienklasse für die Benennung replizierter Themen im Zielcluster. Der Standardwert ist org.apache.kafka.connect.mirror.DefaultReplicationPolicy, wodurch replizierten Themennamen das Quellcluster-Alias vorangestellt wird. Wenn der Alias des Quellclusters primary lautet, wird ein Thema mit dem Namen orders im Zielcluster in primary.orders repliziert. Wenn Sie die ursprünglichen Themennamen ohne Präfix beibehalten möchten, setzen Sie dieses Attribut auf org.apache.kafka.connect.mirror.IdentityReplicationPolicy.

Konfigurationen, mit denen bestimmte Ressourcen bei der Datenreplikation ausgeschlossen werden:

  • topics.exclude: Ausgeschlossene Themen. Unterstützt kommagetrennte Themennamen und reguläre Ausdrücke. Ausschlüsse haben Vorrang vor Einschlüssen. Wird für den MirrorMaker 2.0-Quellconnector verwendet. Der Standardwert ist mm2.*.internal,.*.replica,__.*.

  • groups.exclude: Gruppen ausschließen. Unterstützt kommagetrennte Gruppen-IDs und reguläre Ausdrücke. Ausschlüsse haben Vorrang vor Einschlüssen. Wird für den MirrorMaker 2.0-Prüfpunkt-Connector verwendet. Der Standardwert ist console-consumer-.*,connect-.*,__.*.

Die verfügbaren Konfigurationsattribute hängen vom jeweiligen Connector ab. Prüfen Sie, welche Version des MirrorMaker 2.0-Connectors unterstützt wird, um zu sehen, welche Konfigurationsbeispiele unterstützt werden. Weitere Informationen finden Sie in folgenden Dokumenten:

Kafka-Datensatzkonvertierung

Kafka Connect verwendet org.apache.kafka.connect.converters.ByteArrayConverter als Standardkonverter für Schlüssel und Wert. Dadurch wird eine Pass-Through-Option bereitgestellt, die keine Konvertierung vornimmt.

Sie können header.converter, key.converter und value.converter so konfigurieren, dass andere Konverter verwendet werden.

Taskanzahl

Mit dem Wert tasks.max wird die maximale Anzahl von Aufgaben konfiguriert, die Kafka Connect zum Ausführen von MirrorMaker-Connectors verwendet. Er steuert den Parallelitätsgrad für einen Connector. Wenn Sie die Anzahl der Aufgaben erhöhen, kann sich der Durchsatz erhöhen. Dies ist jedoch durch Faktoren wie die Anzahl der Kafka-Themenpartitionen begrenzt.

MirrorMaker 2.0-Quell-Connector erstellen

Bevor Sie einen Connector erstellen, lesen Sie die Dokumentation zu Connector-Eigenschaften.

Console

  1. Rufen Sie in der Google Cloud Console die Seite Cluster verbinden auf.

    Zu „Connect-Cluster“

  2. Klicken Sie auf den Connect-Cluster, in dem Sie den Connector erstellen möchten.

    Die Seite Connect-Clusterdetails wird angezeigt.

  3. Klicken Sie auf Connector erstellen.

    Die Seite Kafka-Connector erstellen wird angezeigt.

  4. Geben Sie für den Connector-Namen einen String ein.

    Weitere Informationen zum Benennen eines Connectors finden Sie in den Richtlinien zum Benennen einer Managed Service for Apache Kafka-Ressource.

  5. Wählen Sie für Connector-Plug-in die Option „MirrorMaker 2.0-Quelle“ aus.

  6. Wählen Sie für Quellcluster eine der folgenden Optionen aus:

    • Managed Service for Apache Kafka-Cluster: Wählen Sie einen Cluster aus dem Menü aus.
    • Selbstverwalteter oder externer Kafka-Cluster: Geben Sie die Bootstrap-Adresse im Format hostname:port_number ein.

  7. Geben Sie die kommagetrennten Themennamen oder den Themen-Regex ein.

  8. Prüfen Sie die Konfigurationen, einschließlich der erforderlichen Sicherheitseinstellungen, und passen Sie sie an.

    Weitere Informationen zur Konfiguration und Authentifizierung finden Sie unter Konfiguration.

  9. Wählen Sie die Richtlinie für Task-Neustart aus. Weitere Informationen finden Sie unter Richtlinie für den Neustart von Aufgaben.

  10. Klicken Sie auf Erstellen.

gcloud

Führen Sie den Befehl gcloud managed-kafka connectors create aus:

gcloud managed-kafka connectors create CONNECTOR_ID \
    --connect-cluster=CONNECT_CLUSTER_ID \
    --location=LOCATION \
    --config-file=CONFIG_FILE

Ersetzen Sie Folgendes:

  • CONNECTOR_ID: Die ID oder der Name des Connectors.
  • CONNECT_CLUSTER_ID: Die ID des Connect-Clusters, in dem der Connector erstellt wird.
  • LOCATION: Der Standort des Connect-Clusters.
  • CONFIG_FILE: Der Pfad zu einer lokalen YAML- oder JSON-Konfigurationsdatei.

Hier sehen Sie ein Beispiel für eine Konfigurationsdatei für den MirrorMaker 2.0-Quell-Connector:

connector.class: "org.apache.kafka.connect.mirror.MirrorSourceConnector"
source.cluster.alias: "source"
target.cluster.alias: "target"
topics: "TOPIC_LIST"
source.cluster.bootstrap.servers: "SOURCE_CLUSTER_BOOTSTRAP_ADDRESS"
target.cluster.bootstrap.servers: "TARGET_CLUSTER_BOOTSTRAP_ADDRESS"

Ersetzen Sie Folgendes:

  • TOPIC_LIST: Eine durch Kommas getrennte Liste von Kafka-Themen, die repliziert werden sollen. Die Liste kann reguläre Ausdrücke enthalten.

  • SOURCE_CLUSTER_BOOTSTRAP_ADDRESS: Die Bootstrap-Adresse des Kafka-Quellclusters.

  • TARGET_CLUSTER_BOOTSTRAP_ADDRESS: Die Bootstrap-Adresse des Kafka-Zielclusters.

Wenn der Quellcluster ein selbstverwalteter Kafka-Cluster ist, müssen Sie möglicherweise zusätzliche Attribute zur Konfiguration der Authentifizierung angeben. Weitere Informationen finden Sie unter Authentifizierung konfigurieren.

Terraform

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

# A single MirrorMaker 2 Source Connector to replicate from one source to one target.
resource "google_managed_kafka_connector" "default" {
  project         = data.google_project.default.project_id
  connector_id    = "mm2-source-to-target-connector-id"
  connect_cluster = google_managed_kafka_connect_cluster.default.connect_cluster_id
  location        = "us-central1"

  configs = {
    "connector.class"      = "org.apache.kafka.connect.mirror.MirrorSourceConnector"
    "name"                 = "mm2-source-to-target-connector-id"
    "tasks.max"            = "3"
    "source.cluster.alias" = "source"
    "target.cluster.alias" = "target"
    "topics"               = ".*" # Replicate all topics from the source
    # The value for bootstrap.servers is a comma-separated list of hostname:port pairs
    # for one or more Kafka brokers in the source/target cluster.
    "source.cluster.bootstrap.servers" = "source_cluster_dns"
    "target.cluster.bootstrap.servers" = "target_cluster_dns"
    # You can define an exclusion policy for topics as follows:
    # To exclude internal MirrorMaker 2 topics, internal topics and replicated topics,.
    "topics.exclude" = "mm2.*\\.internal,.*\\.replica,__.*"
  }

  provider = google-beta
}

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 API-Referenzdokumentation für Managed Service for Apache Kafka Go.

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

import (
	"context"
	"fmt"
	"io"

	managedkafka "cloud.google.com/go/managedkafka/apiv1"
	"cloud.google.com/go/managedkafka/apiv1/managedkafkapb"
	"google.golang.org/api/option"
)

// createMirrorMaker2SourceConnector creates a MirrorMaker 2.0 Source connector.
func createMirrorMaker2SourceConnector(w io.Writer, projectID, region, connectClusterID, connectorID, sourceBootstrapServers, targetBootstrapServers, tasksMax, sourceClusterAlias, targetClusterAlias, topics, topicsExclude string, opts ...option.ClientOption) error {
	// TODO(developer): Update with your config values. Here is a sample configuration:
	// projectID := "my-project-id"
	// region := "us-central1"
	// connectClusterID := "my-connect-cluster"
	// connectorID := "mm2-source-to-target-connector-id"
	// sourceBootstrapServers := "source_cluster_dns"
	// targetBootstrapServers := "target_cluster_dns"
	// tasksMax := "3"
	// sourceClusterAlias := "source"
	// targetClusterAlias := "target"
	// topics := ".*"
	// topicsExclude := "mm2.*.internal,.*.replica,__.*"
	ctx := context.Background()
	client, err := managedkafka.NewManagedKafkaConnectClient(ctx, opts...)
	if err != nil {
		return fmt.Errorf("managedkafka.NewManagedKafkaConnectClient got err: %w", err)
	}
	defer client.Close()

	parent := fmt.Sprintf("projects/%s/locations/%s/connectClusters/%s", projectID, region, connectClusterID)

	config := map[string]string{
		"connector.class":      "org.apache.kafka.connect.mirror.MirrorSourceConnector",
		"name":                 connectorID,
		"tasks.max":            tasksMax,
		"source.cluster.alias": sourceClusterAlias,
		"target.cluster.alias": targetClusterAlias, // This is usually the primary cluster.
		// Replicate all topics from the source
		"topics": topics,
		// The value for bootstrap.servers is a hostname:port pair for the Kafka broker in
		// the source/target cluster.
		// For example: "kafka-broker:9092"
		"source.cluster.bootstrap.servers": sourceBootstrapServers,
		"target.cluster.bootstrap.servers": targetBootstrapServers,
		// You can define an exclusion policy for topics as follows:
		// To exclude internal MirrorMaker 2 topics, internal topics and replicated topics.
		// topicsExclude := "mm2.*.internal,.*.replica,__.*"
		"topics.exclude": topicsExclude,
	}

	connector := &managedkafkapb.Connector{
		Name:    fmt.Sprintf("%s/connectors/%s", parent, connectorID),
		Configs: config,
	}

	req := &managedkafkapb.CreateConnectorRequest{
		Parent:      parent,
		ConnectorId: connectorID,
		Connector:   connector,
	}

	resp, err := client.CreateConnector(ctx, req)
	if err != nil {
		return fmt.Errorf("client.CreateConnector got err: %w", err)
	}
	fmt.Fprintf(w, "Created MirrorMaker 2.0 Source connector: %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 Managed Service for Apache Kafka Java API.

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


import com.google.api.gax.rpc.ApiException;
import com.google.cloud.managedkafka.v1.ConnectClusterName;
import com.google.cloud.managedkafka.v1.Connector;
import com.google.cloud.managedkafka.v1.ConnectorName;
import com.google.cloud.managedkafka.v1.CreateConnectorRequest;
import com.google.cloud.managedkafka.v1.ManagedKafkaConnectClient;
import java.io.IOException;
import java.util.HashMap;
import java.util.Map;

public class CreateMirrorMaker2SourceConnector {

  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 maxTasks = "3";
    String connectClusterId = "my-connect-cluster";
    String connectorId = "my-mirrormaker2-connector";
    String sourceClusterBootstrapServers = "my-source-cluster:9092";
    String targetClusterBootstrapServers = "my-target-cluster:9092";
    String sourceClusterAlias = "source";
    String targetClusterAlias = "target"; // This is usually the primary cluster.
    String connectorClass = "org.apache.kafka.connect.mirror.MirrorSourceConnector";
    String topics = ".*";
    // You can define an exclusion policy for topics as follows:
    // To exclude internal MirrorMaker 2 topics, internal topics and replicated topics.
    String topicsExclude = "mm2.*.internal,.*.replica,__.*";
    createMirrorMaker2SourceConnector(
        projectId,
        region,
        maxTasks,
        connectClusterId,
        connectorId,
        sourceClusterBootstrapServers,
        targetClusterBootstrapServers,
        sourceClusterAlias,
        targetClusterAlias,
        connectorClass,
        topics,
        topicsExclude);
  }

  public static void createMirrorMaker2SourceConnector(
      String projectId,
      String region,
      String maxTasks,
      String connectClusterId,
      String connectorId,
      String sourceClusterBootstrapServers,
      String targetClusterBootstrapServers,
      String sourceClusterAlias,
      String targetClusterAlias,
      String connectorClass,
      String topics,
      String topicsExclude)
      throws Exception {

    // Build the connector configuration
    Map<String, String> configMap = new HashMap<>();
    configMap.put("tasks.max", maxTasks);
    configMap.put("connector.class", connectorClass);
    configMap.put("name", connectorId);
    configMap.put("source.cluster.alias", sourceClusterAlias);
    configMap.put("target.cluster.alias", targetClusterAlias);
    configMap.put("topics", topics);
    configMap.put("topics.exclude", topicsExclude);
    configMap.put("source.cluster.bootstrap.servers", sourceClusterBootstrapServers);
    configMap.put("target.cluster.bootstrap.servers", targetClusterBootstrapServers);

    Connector connector = Connector.newBuilder()
        .setName(
            ConnectorName.of(projectId, region, connectClusterId, connectorId).toString())
        .putAllConfigs(configMap)
        .build();

    try (ManagedKafkaConnectClient managedKafkaConnectClient = ManagedKafkaConnectClient.create()) {
      CreateConnectorRequest request = CreateConnectorRequest.newBuilder()
          .setParent(ConnectClusterName.of(projectId, region, connectClusterId).toString())
          .setConnectorId(connectorId)
          .setConnector(connector)
          .build();

      // This operation is being handled synchronously.
      Connector response = managedKafkaConnectClient.createConnector(request);
      System.out.printf("Created MirrorMaker2 Source connector: %s\n", response.getName());
    } catch (IOException | ApiException e) {
      System.err.printf("managedKafkaConnectClient.createConnector got err: %s\n", e.getMessage());
    }
  }
}

Python

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

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

from google.api_core.exceptions import GoogleAPICallError
from google.cloud.managedkafka_v1.services.managed_kafka_connect import (
    ManagedKafkaConnectClient,
)
from google.cloud.managedkafka_v1.types import Connector, CreateConnectorRequest

connect_client = ManagedKafkaConnectClient()
parent = connect_client.connect_cluster_path(project_id, region, connect_cluster_id)

configs = {
    "connector.class": "org.apache.kafka.connect.mirror.MirrorSourceConnector",
    "name": connector_id,
    "tasks.max": tasks_max,
    "source.cluster.alias": source_cluster_alias,
    "target.cluster.alias": target_cluster_alias,  # This is usually the primary cluster.
    # Replicate all topics from the source
    "topics": topics,
    # The value for bootstrap.servers is a hostname:port pair for the Kafka broker in
    # the source/target cluster.
    # For example: "kafka-broker:9092"
    "source.cluster.bootstrap.servers": source_bootstrap_servers,
    "target.cluster.bootstrap.servers": target_bootstrap_servers,
    # You can define an exclusion policy for topics as follows:
    # To exclude internal MirrorMaker 2 topics, internal topics and replicated topics.
    "topics.exclude": topics_exclude,
}

connector = Connector()
# The name of the connector.
connector.name = connector_id
connector.configs = configs

request = CreateConnectorRequest(
    parent=parent,
    connector_id=connector_id,
    connector=connector,
)

try:
    operation = connect_client.create_connector(request=request)
    print(f"Waiting for operation {operation.operation.name} to complete...")
    response = operation.result()
    print("Created Connector:", response)
except GoogleAPICallError as e:
    print(f"The operation failed with error: {e}")

Nächste Schritte

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