Google utilizza la tecnologia AI per tradurre i contenuti nella tua lingua preferita. Le traduzioni generate dall'AI potrebbero contenere errori.

Crea un connettore di origine MirrorMaker 2.0

Un connettore di origine MirrorMaker 2.0 consente di replicare argomenti e dati da un cluster Kafka (l'origine) a un altro cluster Kafka (la destinazione).

I casi d'uso tipici di questo connettore includono:

Migrazione dei dati. Sposta il tuo workload Kafka in un nuovo cluster Managed Service per Apache Kafka.
Disaster recovery. Crea un cluster di backup per garantire la continuità aziendale in caso di errori.
Aggregazione dei dati. Consolidare i dati di più cluster Kafka in un cluster Managed Service per Apache Kafka centrale per eseguire analisi.

Questo connettore esegue la replica di base dei dati tra i cluster Kafka. Per garantire ripristino di emergenza più efficace, utilizza i seguenti connettori insieme al connettore di origine MirrorMaker 2.0:

Connettore MirrorMaker 2.0 Checkpoint. Replica gli offset dei consumer, in modo che questi possano eseguire il failover sul cluster di destinazione senza perdere dati.
Connettore MirrorMaker 2.0 Heartbeat. Genera messaggi di heartbeat periodici, che ti consentono di monitorare la connettività e il ritardo di replica.

Informazioni sui ruoli dei cluster in MirrorMaker 2.0

Un connettore MirrorMaker 2.0 legge da un cluster Kafka di origine e scrive in un cluster Kafka di destinazione. In Managed Service per Apache Kafka, il cluster principale del cluster Connect deve essere il cluster di destinazione. Questa configurazione riduce al minimo la latenza per le operazioni di scrittura. Il cluster di origine può essere un altro cluster Managed Service per Apache Kafka o un cluster Kafka esterno o autogestito.

Se vuoi replicare i dati da un cluster Managed Service per Apache Kafka a un cluster esterno o autogestito, ti consigliamo di eseguire Kafka Connect nella stessa località o regione dell'altro cluster e di utilizzare MirrorMaker 2.0 per importare i dati da Managed Service per Apache Kafka.

Devi configurare correttamente tutte le proprietà del connettore. Queste includono anche le proprietà di autenticazione del producer indirizzate al cluster secondario. Per informazioni dettagliate sui potenziali problemi, consulta Migliora la configurazione del client MirrorMaker 2.0.

Ruoli e autorizzazioni richiesti

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

Autorizzazioni obbligatorie

Per creare un connettore sono necessarie le seguenti autorizzazioni:

Crea un connettore: managedkafka.connectors.create

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

Configura il networking

Questa sezione descrive i requisiti di networking per consentire a MirrorMaker 2.0 di connettersi al cluster Kafka secondario. Ecco i requisiti generali:

Connettività. I worker del cluster di connessione devono essere in grado di connettersi al cluster Kafka secondario.
Risoluzione DNS. Devi aggiungere il dominio DNS del cluster secondario ai domini DNS risolvibili del cluster di connessione. Inoltre, il dominio DNS deve essere risolvibile dalla subnet del cluster di connessione, ad esempio creando una zona Cloud DNS.
Regole firewall. Le regole firewall devono consentire ai worker del cluster di connessione di raggiungere sia il cluster Kafka di origine sia quello di destinazione.

Molti dei dettagli dipendono da come e dove è ospitato il cluster Kafka secondario, se si tratta di un cluster Managed Service per Apache Kafka, di un cluster autogestito in Google Cloudo di un cluster esterno.

Cluster Managed Service per Apache Kafka
- Se il cluster secondario si trova in una rete VPC diversa, aggiungi una subnet dalla rete VPC del cluster di connessione alle subnet connesse del cluster secondario. Per saperne di più, consulta Configurazione di rete.
Cluster Kafka autogestito in Google Cloud
- Se il cluster secondario si trova in una rete VPC diversa, le opzioni di connettività includono il peering di rete VPC, Network Connectivity Center (NCC) e Cloud VPN.
- Se il cluster secondario si trova in una rete VPC diversa, crea una zona di peering Cloud DNS che esegue il peering con la rete VPC del cluster secondario e sia visibile nella subnet del cluster di connessione.
Cluster Kafka esterno
- Le opzioni per connettersi a un cluster esterno includono:
  - Cloud VPN: una soluzione conveniente adatta per una larghezza di banda inferiore. Crea un tunnel criptato sulla rete internet pubblica.
  - Cloud Interconnect: fornisce una connessione dedicata a larghezza di banda elevata tra la tua rete on-premise e Google Cloud. Puoi scegliere tra Dedicated Interconnect per una connessione fisica diretta o Partner Interconnect per connetterti tramite un fornitore di servizi.
- Configura Cloud NAT per consentire ai worker di Connect di accedere a internet.
- Crea una zona di forwarding Cloud DNS nella rete VPC del cluster di connessione per risolvere le query DNS per l'indirizzo di bootstrap e gli endpoint del broker del cluster esterno.

Configura l'autenticazione

Se il cluster Kafka di origine o di destinazione è un cluster Managed Service per Apache Kafka, il connettore viene configurato automaticamente per utilizzare l'autenticazione OAuthBearer. Non è necessario impostare altre configurazioni.

Per un cluster Kafka autogestito o esterno, devi configurare l'autenticazione e la crittografia TLS. Le configurazioni dipendono dal meccanismo di autenticazione supportato dal cluster Kafka. Per saperne di più, consulta Configurazioni comuni di MirrorMaker nella documentazione di Apache Kafka.

Se il cluster supporta l'autenticazione OAuthBearer, utilizza le seguenti impostazioni:

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;

Archivia le credenziali richieste in Secret Manager, come descritto nella sezione successiva.

Utilizza Secret Manager per le credenziali

Quando ti connetti a cluster Kafka autogestiti o esterni, utilizza Secret Manager per archiviare i valori di configurazione sensibili necessari per l'autenticazione. Queste informazioni potrebbero includere password, file di archivio chiavi e file di archivio di attendibilità, a seconda di come è configurata l'autenticazione nel cluster.

Per saperne di più su come associare i secret di Secret Manager a un cluster Connect, consulta Risorse di Secret Manager.

Nella configurazione del connettore, fai riferimento ai secret come segue:

Per i percorsi file, utilizza il formato /var/secrets/PROJECT_NAME-SECRET_NAME-SECRET_VERSION. Esempio: ssl.truststore.location=/var/secrets/project1-truststore-1.
Per le password, utilizza il formato ${directory:/var/secrets:PROJECT_NAME-SECRET_NAME-SECRET_VERSION}. Esempio: password=${directory:/var/secrets:project1-database_password-3}.

Sostituisci quanto segue:

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

Come funziona un connettore di origine MirrorMaker

Un connettore di origine MirrorMaker estrae i dati da uno o più argomenti Kafka in un cluster di origine e li replica, insieme alle ACL, negli argomenti di un cluster di destinazione.

Ecco una suddivisione dettagliata di come il connettore MirrorMaker Source replica i dati:

Il connettore utilizza i messaggi degli argomenti Kafka specificati all'interno del cluster di origine. Specifica gli argomenti da replicare utilizzando la proprietà di configurazione topics, che accetta nomi di argomenti separati da virgole o una singola espressione regolare in stile Java. Ad esempio, topic-a,topic-b o my-prefix-.*.
Il connettore può anche saltare la replica di argomenti specifici che indichi utilizzando la proprietà topics.exclude; le esclusioni sostituiscono le inclusioni.
Il connettore scrive i messaggi consumati nel cluster di destinazione.
Il connettore richiede i dettagli di connessione del cluster di origine e di destinazione, come source.cluster.bootstrap.servers e target.cluster.bootstrap.servers.
Il connettore richiede anche alias per i cluster di origine e di destinazione, come specificato da source.cluster.alias e target.cluster.alias. Per impostazione predefinita, gli argomenti replicati vengono rinominati automaticamente utilizzando l'alias di origine. Ad esempio, un argomento denominato orders di un'origine con alias primary diventa primary.orders nella destinazione. Per mantenere i nomi degli argomenti originali nel cluster di destinazione senza un prefisso, imposta replication.policy.class su org.apache.kafka.connect.mirror.IdentityReplicationPolicy.
Anche gli ACL associati agli argomenti replicati vengono sincronizzati dal cluster di origine a quello di destinazione. Questa opzione può essere disattivata utilizzando la proprietà sync.topic.acls.enabled.
I dettagli di autenticazione per la connessione ai cluster di origine e di destinazione devono essere forniti nella configurazione, se richiesto dai cluster. Devi configurare proprietà come security.protocol, sasl.mechanism e sasl.jaas.config, con il prefisso source.cluster. per l'origine e target.cluster. per la destinazione.
Il connettore si basa su argomenti interni. Potresti dover configurare proprietà correlate, ad esempio offset-syncs.topic.replication.factor.
Il connettore utilizza i convertitori di record Kafka key.converter, value.converter e header.converter. Per la replica diretta, spesso il valore predefinito è org.apache.kafka.connect.converters.ByteArrayConverter, che non esegue alcuna conversione (pass-through).
La proprietà tasks.max controlla il livello di parallelismo per il connettore. L'aumento di tasks.max può potenzialmente migliorare il throughput, ma il parallelismo effettivo è spesso limitato dal numero di partizioni negli argomenti Kafka di origine che vengono replicati.

Proprietà di un connettore MirrorMaker 2.0

Quando crei o aggiorni un connettore MirrorMaker 2.0, specifica queste proprietà:

Nome connettore

Il nome o l'ID del connettore. Per linee guida su come assegnare un nome alla risorsa, consulta le linee guida per assegnare un nome a una risorsa Managed Service per Apache Kafka. Il nome è immutabile.

Tipo di connettore

Il tipo di connettore deve essere uno dei seguenti:

Cluster di origine

Il cluster Kafka di origine da cui legge il connettore MirrorMaker 2.0.

Cluster Managed Service per Apache Kafka: seleziona un cluster dal menu a discesa.
Cluster Kafka autogestito o esterno: inserisci l'indirizzo di bootstrap nel formato hostname:port_number. Ad esempio kafka-test:9092.

Nomi di argomenti o espressioni regolari

Gli argomenti da replicare. Specifica i singoli nomi (topic1, topic2) o utilizza un'espressione regolare (topic.*). Questa proprietà è obbligatoria per il connettore di origine MirrorMaker 2.0. Il valore predefinito è .*

Nomi dei gruppi di consumer o espressioni regolari

I gruppi di consumer da replicare. Specifica i singoli nomi (group1, group2) o utilizza un'espressione regolare (group.*). Questa proprietà è obbligatoria per il connettore di checkpoint MirrorMaker 2.0. Il valore predefinito è .*

Configurazione

Questa sezione consente di specificare proprietà di configurazione aggiuntive specifiche per il connettore MirrorMaker 2.0.

Poiché i dati negli argomenti Kafka possono essere in vari formati, come Avro, JSON o byte non elaborati, una parte fondamentale della configurazione prevede la specifica dei convertitori. I convertitori traducono i dati dal formato utilizzato negli argomenti Kafka al formato interno standardizzato di Kafka Connect.

Per informazioni più generali sul ruolo dei convertitori in Kafka Connect, sui tipi di convertitori supportati e sulle opzioni di configurazione comuni, consulta Convertitori.

Alcune configurazioni comuni per tutti i connettori MirrorMaker 2.0 includono:

source.cluster.alias: alias del cluster di origine.
target.cluster.alias: alias del cluster di destinazione.
replication.policy.class: classe di policy per la denominazione degli argomenti replicati nel cluster di destinazione. Il valore predefinito è org.apache.kafka.connect.mirror.DefaultReplicationPolicy, che aggiunge ai nomi degli argomenti replicati il prefisso dell'alias del cluster di origine. Se l'alias del cluster di origine è primary, un argomento denominato orders viene replicato in primary.orders nel cluster di destinazione. Per mantenere i nomi degli argomenti originali senza un prefisso, imposta questa proprietà su org.apache.kafka.connect.mirror.IdentityReplicationPolicy.

Configurazioni utilizzate per escludere risorse specifiche durante la replica dei dati:

topics.exclude: argomenti esclusi. Supporta nomi di argomenti e regex separati da virgole. Le esclusioni hanno la precedenza sulle inclusioni. Utilizzato per il connettore di origine MirrorMaker 2.0. Il valore predefinito è mm2.*.internal,.*.replica,__.*
groups.exclude: escludi i gruppi. Supporta ID gruppo ed espressioni regolari separati da virgole. Le esclusioni hanno la precedenza sulle inclusioni. Utilizzato per il connettore di checkpoint MirrorMaker 2.0. Il valore predefinito è console-consumer-.*,connect-.*,__.*

Le proprietà di configurazione disponibili dipendono dal connettore specifico. Controlla la versione del connettore MirrorMaker 2.0 supportata per vedere quali esempi di configurazione sono supportati. Consulta i seguenti documenti:

Conversione dei record Kafka

Kafka Connect utilizza org.apache.kafka.connect.converters.ByteArrayConverter come convertitore predefinito per chiave e valore, che fornisce un'opzione di pass-through che non esegue alcuna conversione.

Puoi configurare header.converter, key.converter e value.converter per utilizzare altri convertitori.

Numero di attività

Il valore tasks.max configura il numero massimo di attività utilizzate da Kafka Connect per eseguire i connettori MirrorMaker. Controlla il livello di parallelismo di un connettore. L'aumento del numero di attività può aumentare la velocità effettiva, ma è limitato da fattori come il numero di partizioni degli argomenti Kafka.

Crea un connettore di origine MirrorMaker 2.0

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

Console

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

Vai a Cluster connessi
Fai clic sul cluster di connessione in cui vuoi creare il connettore.

Viene visualizzata la pagina Dettagli cluster di connessione.
Fai clic su Crea connettore.

Viene visualizzata la pagina Crea connettore Kafka.
In Nome connettore, inserisci una stringa.

Per saperne di più su come assegnare un nome a un connettore, consulta le linee guida per assegnare un nome a una risorsa Managed Service per Apache Kafka.
In Plug-in connettore, seleziona "MirrorMaker 2.0 Source".
In Cluster di origine, scegli una delle seguenti opzioni:
- Cluster Managed Service per Apache Kafka: seleziona dal menu.
- Cluster Kafka autogestito o esterno: inserisci l'indirizzo di bootstrap nel formato hostname:port_number.
Inserisci i nomi o regex degli argomenti separati da virgole.
Esamina e modifica le configurazioni, incluse le impostazioni di sicurezza richieste.

Per saperne di più sulla configurazione e sull'autenticazione, consulta Configurazione.
Seleziona la Policy di riavvio attività. Per saperne di più, consulta Policy di riavvio delle attività.
Fai clic su Crea.

gcloud

Esegui il comando gcloud managed-kafka connectors create:

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

Sostituisci quanto segue:

CONNECTOR_ID: l'ID o il nome del connettore.
CONNECT_CLUSTER_ID: l'ID del cluster Connect in cui viene creato il connettore.
LOCATION: la località del cluster Connect.
CONFIG_FILE: il percorso di un file di configurazione YAML o JSON locale.

Ecco un esempio di file di configurazione per il connettore di origine MirrorMaker 2.0:

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"

Sostituisci quanto segue:

TOPIC_LIST: un elenco separato da virgole di argomenti Kafka da replicare. L'elenco può includere espressioni regolari.
SOURCE_CLUSTER_BOOTSTRAP_ADDRESS: l'indirizzo di bootstrap del cluster Kafka di origine.
TARGET_CLUSTER_BOOTSTRAP_ADDRESS: l'indirizzo di bootstrap del cluster Kafka di destinazione.

Se il cluster di origine è un cluster Kafka autogestito, potresti dover includere proprietà aggiuntive per configurare l'autenticazione. Per saperne di più, consulta Configura l'autenticazione.

Terraform

Puoi utilizzare una risorsa Terraform per creare un connettore.

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

Per scoprire come applicare o rimuovere una configurazione Terraform, consulta Comandi Terraform di base.

Vai

Prima di provare questo esempio, segui le istruzioni di configurazione di Go in Installa le librerie client. Per saperne di più, consulta la documentazione di riferimento dell'API Go di Managed Service per Apache Kafka.

Per eseguire l'autenticazione in Managed Service per Apache Kafka, configura le Credenziali predefinite dell'applicazione(ADC). Per saperne di più, consulta Configura ADC per un ambiente di sviluppo locale.

import (
	"context"
	"fmt"
	"io"

	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

Prima di provare questo esempio, segui le istruzioni di configurazione di Java riportate in Installa le librerie client. Per saperne di più, consulta la documentazione di riferimento dell'API Java di Managed Service per Apache Kafka.

Per eseguire l'autenticazione in Managed Service per Apache Kafka, configura le Credenziali predefinite dell'applicazione. Per saperne di più, consulta Configura ADC per un ambiente di sviluppo locale.


import com.google.api.gax.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

Prima di provare questo esempio, segui le istruzioni di configurazione di Python in Installa le librerie client. Per saperne di più, consulta la documentazione di riferimento dell'API Python di Managed Service per Apache Kafka.

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

Passaggi successivi

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

Crea un connettore di origine MirrorMaker 2.0 Mantieni tutto organizzato con le raccolte Salva e classifica i contenuti in base alle tue preferenze.

Informazioni sui ruoli dei cluster in MirrorMaker 2.0

Ruoli e autorizzazioni richiesti

Autorizzazioni obbligatorie

Configura il networking

Configura l'autenticazione

Utilizza Secret Manager per le credenziali

Come funziona un connettore di origine MirrorMaker

Proprietà di un connettore MirrorMaker 2.0

Nome connettore

Tipo di connettore

Cluster di origine

Nomi di argomenti o espressioni regolari

Nomi dei gruppi di consumer o espressioni regolari

Configurazione

Conversione dei record Kafka

Numero di attività

Crea un connettore di origine MirrorMaker 2.0

Console

gcloud

Terraform

Vai

Java

Python

Passaggi successivi

Crea un connettore di origine MirrorMaker 2.0