All Products
Search

This Product

    E-MapReduce:Use MirrorMaker 2 (on Connect) to synchronize data across clusters

    Document Center

    E-MapReduce:Use MirrorMaker 2 (on Connect) to synchronize data across clusters

    Last Updated:Jun 20, 2026

    This topic shows how to use MirrorMaker 2 (MM2) on Kafka Connect to synchronize data across clusters.

    Background

    Use cases

    Kafka MM2 is suitable for the following use cases:
    • Remote data synchronization: MM2 allows you to replicate Kafka data between clusters in different regions.
    • Disaster recovery: You can use MM2 to build a high-availability architecture with primary and backup clusters in different data centers. MM2 synchronizes data between the clusters in real time. If one cluster fails, you can switch your applications to the other cluster for geo-disaster recovery.
    • Data migration: You can use MM2 for data migration and to ensure business continuity during cloud migration, hybrid cloud adoption, or cluster upgrades.
    • Data aggregation: You can use MM2 to synchronize data from multiple Kafka sub-clusters to a central Kafka cluster for data aggregation.

    Features

    As a data replication tool, MM2 provides the following features:
    • Replicates topic data and configurations.
    • Replicates consumer groups and their offsets.
    • Replicates access control lists (ACLs).
    • Automatically detects new topics and partitions.
    • Provides MM2 metrics.
    • Provides a high-availability and horizontally scalable framework.

    Job execution methods

    MM2 jobs can be run in the following ways:
    • Run as a connector in a distributed Connect cluster (Recommended): This method involves running MM2 connector tasks in an existing Kafka Connect cluster. You can follow the instructions in this topic to manage MM2 tasks using the Kafka Connect service.
    • Run in a dedicated MirrorMaker cluster: This method does not require a Kafka Connect cluster. You directly manage all MM2 tasks through a driver program. For detailed instructions, see Use MirrorMaker 2 (Dedicated) to synchronize data across clusters.
    • Run as a worker in a standalone Kafka Connect cluster: This method runs a single MirrorSourceConnector task and is suitable for testing environments.
    Note Run MM2 connector tasks on a distributed Kafka Connect cluster to manage them using the REST API of the Kafka Connect cluster.

    For more information about MM2, see the Apache Kafka documentation.

    Prerequisites

    Create two Kafka clusters: a source cluster named emrsource and a target cluster named emrdest. The Kafka service is selected for both clusters. For information about how to create a DataFlow cluster, see Create a cluster.
    Note In this example, both the source cluster and the target cluster are E-MapReduce V3.42.0 DataFlow clusters that reside in the same Virtual Private Cloud (VPC).

    Limitations

    The Kafka version of the target cluster must be 2.12_2.4.1 or later.

    Procedure

    1. Step 1: Create a Kafka Connect cluster in the target cluster
    2. Step 2: Use the MirrorMaker 2 connector

    Step 1: Create a Kafka Connect cluster

    1. Create a task node group for E-MapReduce.
      On the Nodes page of the emrdest target cluster in the E-MapReduce console, create a task node group.
      1. Click Create Node Group.
      2. In the Create Node Group panel, configure the following parameters. Use the default values for other parameters or configure them as needed.
        Parameter Description
        Node Group Type Select TASK (Task Node Group).
        Node Group Name Enter a name. This example uses emr-task.
        Storage Configuration Select a data disk.
    2. Scale out the task node group.
      1. On the Nodes page, find the emr-task node group and click Scale Out in the Actions column.
      2. In the dialog box that appears, specify the number of instances to add and select Terms of Service.
        This example adds one instance. You can scale out the number of task instances as needed. For a high-availability Kafka Connect cluster, add at least two instances.
      3. Click OK.
    3. Check the status of the KafkaConnect service to verify that the Kafka Connect cluster is running.
      1. At the top of the page, click Services.
      2. In the Kafka service section, click Status.
      3. In the Components section, check the status of KafkaConnect to ensure that it is running.
        In the Topology list, verify that the health status of the KafkaConnect node task-1-1 (Role: Task) is Healthy and the component status is Running.
    4. Use SSH to connect to the emrdest target cluster. For more information, see Connect to a cluster.
    5. Run the following command to check the status of the Kafka Connect REST service. 
      curl -X GET http://task-1-1:8083| jq .
      The output is similar to the following:
        % Total    % Received % Xferd  Average Speed   Time    Time     Time  Current
                                 Dload  Upload   Total   Spent    Left  Speed
100    91  100    91    0     0  13407      0 --:--:-- --:--:-- --:--:-- 15166
{
  "version": "2.4.1",
  "commit": "42ce056344c5625a",
  "kafka_cluster_id": "6Z7IdHW4SVO1Pbql4c****"
}

    Step 2: Use the MirrorMaker 2 connector

    1. Prepare the MM2 connector configuration files.
      Prepare the following files:
      • Prepare the MirrorSourceConnector configuration file.
        In this example, the MirrorSourceConnector configuration file is named mm2-source-connector.json. Use the following example and replace the parameter values to match your environment. For more information about configuration options, see the relevant sections in KIP-382.
        {
  "name": "mm2-source-connector",
  "connector.class": "org.apache.kafka.connect.mirror.MirrorSourceConnector",
  "clusters": "emrsource,emrdest",
  "source.cluster.alias": "emrsource",
  "target.cluster.alias": "emrdest",
  "target.cluster.bootstrap.servers": "core-1-1:9092;core-1-2:9092;core-1-3:9092",
  "source.cluster.bootstrap.servers": "10.0.**.**:9092",
  "topics": "^foo.*",
  "tasks.max": "4",
  "key.converter": " org.apache.kafka.connect.converters.ByteArrayConverter",
  "value.converter": "org.apache.kafka.connect.converters.ByteArrayConverter",
  "replication.factor": "3",
  "offset-syncs.topic.replication.factor": "3",
  "sync.topic.acls.interval.seconds": "20",
  "sync.topic.configs.interval.seconds": "20",
  "refresh.topics.interval.seconds": "20",
  "refresh.groups.interval.seconds": "20",
  "consumer.group.id": "mm2-mirror-source-consumer-group",
  "producer.enable.idempotence":"true",
  "source.cluster.security.protocol": "PLAINTEXT",
  "target.cluster.security.protocol": "PLAINTEXT"
}
        Note Parameters in this example:
        • source.cluster.bootstrap.servers: Replace the value of this parameter with the endpoint of the Kafka service in your emrsource source cluster. Ensure communication between the source Kafka cluster and the Kafka Connect cluster.
        • topics: This parameter replicates topics from the source cluster whose names start with foo.
      • Prepare the MirrorCheckpointConnector configuration file.
        In this example, the MirrorCheckpointConnector configuration file is named mm2-checkpoint-connector.json. Use the following example and replace the parameter values to match your environment. For more information about configuration options, see the relevant sections in KIP-382.
        {
    "name": "mm2-checkpoint-connector",
    "connector.class": "org.apache.kafka.connect.mirror.MirrorCheckpointConnector",
    "clusters": "emrsource,emrdest",
    "source.cluster.alias": "emrsource",
    "target.cluster.alias": "emrdest",
    "target.cluster.bootstrap.servers": "core-1-1:9092;core-1-2:9092;core-1-3:9092",
    "source.cluster.bootstrap.servers": "10.0.**.**:9092",
    "tasks.max": "1",
    "key.converter": " org.apache.kafka.connect.converters.ByteArrayConverter",
    "value.converter": "org.apache.kafka.connect.converters.ByteArrayConverter",
    "replication.factor": "3",
    "checkpoints.topic.replication.factor": "3",
    "emit.checkpoints.interval.seconds": "20",
    "source.cluster.security.protocol": "PLAINTEXT",
    "target.cluster.security.protocol": "PLAINTEXT"
  }
      • Prepare the MirrorHeartbeatConnector configuration file.
        In this example, the MirrorHeartbeatConnector configuration file is named mm2-heartbeat-connector.json. Use the following example and replace the parameter values to match your environment. For more information about configuration options, see the relevant sections in KIP-382.
        {
    "name": "mm2-heartbeat-connector",
    "connector.class": "org.apache.kafka.connect.mirror.MirrorHeartbeatConnector",
    "clusters": "emrsource,emrdest",
    "source.cluster.alias": "emrsource",
    "target.cluster.alias": "emrdest",
    "target.cluster.bootstrap.servers": "core-1-1:9092;core-1-2:9092;core-1-3:9092",
    "source.cluster.bootstrap.servers": "10.0.**.**:9092",
    "tasks.max": "1",
    "key.converter": " org.apache.kafka.connect.converters.ByteArrayConverter",
    "value.converter": "org.apache.kafka.connect.converters.ByteArrayConverter",
    "replication.factor": "3",
    "heartbeats.topic.replication.factor": "3",
    "emit.heartbeats.interval.seconds": "20",
    "source.cluster.security.protocol": "PLAINTEXT",
    "target.cluster.security.protocol": "PLAINTEXT"
  }
    2. Use the MirrorSourceConnector.
      1. Use the Kafka Connect REST service to create a MirrorSourceConnector task using the mm2-source-connector.json file. 
        curl -X PUT -H "Content-Type: application/json" --data @mm2-source-connector.json http://task-1-1:8083/connectors/mm2-source-connector/config
      2. Run the following command to check the status of the mm2-source-connector. 
        curl -s task-1-1:8083/connectors/mm2-source-connector/status | jq .
    3. Use the MirrorCheckpointConnector.
      1. Use the Kafka Connect REST service to create a MirrorCheckpointConnector task using the mm2-checkpoint-connector.json file. 
        curl -X PUT -H "Content-Type: application/json" --data @mm2-checkpoint-connector.json http://task-1-1:8083/connectors/mm2-checkpoint-connector/config
      2. Run the following command to check the status of the mm2-checkpoint-connector. 
        curl -s task-1-1:8083/connectors/mm2-checkpoint-connector/status | jq .
    4. Use the MirrorHeartbeatConnector.
      1. Use the Kafka Connect REST service to create a MirrorHeartbeatConnector task using the mm2-heartbeat-connector.json file. 
        curl -X PUT -H "Content-Type: application/json" --data @mm2-heartbeat-connector.json http://task-1-1:8083/connectors/mm2-heartbeat-connector/config
      2. Run the following command to check the status of the mm2-heartbeat-connector. 
        curl -s task-1-1:8083/connectors/mm2-heartbeat-connector/status | jq .
    5. On the target cluster, run the following command to view the topics related to MM2. 
      kafka-topics.sh --list --bootstrap-server core-1-1:9092
      The following topics are created in the target cluster:
      • Topics that start with emrsource.foo: Created by the MirrorSourceConnector.

        These correspond to the topics in your source cluster that start with foo.

      • emrsource.checkpoints.internal: Created by the MirrorCheckpointConnector to store information such as offsets.
      • heartbeats: Created by the MirrorHeartbeatConnector.