Disaster Recovery using MirrorMaker 2

The MirrorSourceConnector mirrors the topics and their records. It periodically checks topics on the source cluster and applies any partition and configuration changes it detects to the target cluster. However topic deletions are not mirrored, so if a topic is deleted on the source cluster, its copy on the target cluster is left untouched. It provides a safety net in case the deletion was a mistake. Otherwise, an operator should go and delete that topic in the target cluster.

An important concept to understand is that records on each cluster are highly likely to have different offsets. This happens for several reasons. First, partitions always start at offset 0. So if a partition on the primary cluster has already had offsets deleted due to retention policies, there is always a difference in offsets. Another reason is that MirrorMaker does not mirror transaction markers. Transaction markers are emitted when a transaction is committed or aborted and they occupy offsets. In order to enable offset translation between both clusters, the connector periodically writes a record into the offset-syncs topic that contains both the source and target offsets of a record it just mirrored.

This connector also mirrors topic ACLs, but note that it does not mirror the ALLOW WRITE topic ACLs. This avoids clients (other than MirrorMaker) writing to the target cluster, which might cause inconsistency. It also does not mirror ACLs set on other resources such as groups, transactional IDs, users, or cluster. In case ACLs are used, you need another process to maintain them. For these reasons, if you require authorizations in your clusters, you can’t only rely on MirrorMaker to mirror them. Operational practices like GitOps are a great way to set and maintain ACLs.

The MirrorCheckpointConnector mirrors committed consumer group offsets. It uses the mappings from the offset-syncs topic to translate the committed offsets from the source cluster, and writes the results to the checkpoints topic. By enabling the sync offsets feature, the connector is also able to directly commit the translated offset to the target cluster. Committing the offsets makes it easier to move consuming applications across the clusters, otherwise applications must use the RemoteClusterUtils tool to find their offsets from the checkpoints topic when connecting to the new primary cluster.

The MirrorHeartbeatConnector is not required, and it is recommended that you do not run it for disaster recovery scenarios. This is because the MirrorHeartbeatConnector sends records to the source cluster whereas the other two connectors send records to the target cluster. Since we are deploying MirrorMaker near the target cluster for better performance with the data replication, the MirrorHeartbeatConnector would need to be deployed separately near the source cluster. This added complexity and cost do not justify the benefit, as sufficient metrics to monitor MirrorMaker can be obtained from the MirrorSourceConnector.

When setting up Kafka Connect for MirrorMaker 2, you need to configure both the Kafka Connect worker and the connectors. Kafka Connect offers dozens of configuration settings. You must carefully consider the following settings:

These are configurations to set for both MirrorSourceConnector and MirrorCheckpointConnector:

The following are key configurations for MirrorSourceConnector:

These are the main configurations to be set when you start the connector. However, there are more configurations that you can set for fine tuning. You can list them using the REST API, for example:

curl http://<CONNECT>/connector-plugins/MirrorSourceConnector/config

The following are key configurations for MirrorCheckpointConnector:

These are the main configurations to be set when you start the connector. However, there are more configurations that you can set for fine tuning. You can list them using the REST API, for example:

curl http://<CONNECT>/connector-plugins/MirrorCheckpointConnector/config

To avoid data loss or duplicates in the target cluster while mirroring, enable exactly-once semantics for Kafka Connect and MirrorSourceConnector. Exactly-once support is achieved by using transactions to ensure that the two actions: sending the data to Kafka and committing offsets to the offset storage topic, both succeed or both fail atomically.

This is an example configuration file (e.g. connect-distributed.properties) for a Kafka Connect worker:

bootstrap.servers=<TARGET_BOOTSTRAP_SERVERS>

group.id=connect-mirrormaker-A-to-B

plugin.discovery=service_load

key.converter=org.apache.kafka.connect.converters.ByteArrayConverter
value.converter=org.apache.kafka.connect.converters.ByteArrayConverter

offset.storage.topic=__connect-mirrormaker-A-to-B-offsets
offset.storage.replication.factor=-1
config.storage.topic=__connect-mirrormaker-A-to-B-configs
config.storage.replication.factor=-1
status.storage.topic=__connect-mirrormaker-A-to-B-status
status.storage.replication.factor=-1

exactly.once.source.support=enabled

This is an example configuration for a MirrorSourceConnector mirroring data from cluster A to cluster B:

{
    "name": "A-to-B-MirrorSourceConnector",
    "config": {
        "connector.class": "org.apache.kafka.connect.mirror.MirrorSourceConnector",
        "tasks.max": "30",

        "source.cluster.alias": "A",
        "target.cluster.alias": "B",
        "source.cluster.bootstrap.servers": "<SOURCE_BOOTSTRAP_SERVERS>",
        "target.cluster.bootstrap.servers": "<TARGET_BOOTSTRAP_SERVERS>",

        "offset-syncs.topic.location": "target",
        "replication.policy.class": "org.apache.kafka.connect.mirror.IdentityReplicationPolicy",

        "heartbeats.replication.enabled": "false",
        "replication.factor": "-1",
        "offset-syncs.topic.replication.factor": "-1",
        "sync.topic.acls.enabled": "false",
        "consumer.isolation.level": "read_committed",
        "topics": "<TOPICS>"
    }
}

This is an example configuration for a MirrorCheckpointConnector mirroring consumer groups from cluster A to cluster B:

{
    "name": "A-to-B-MirrorCheckpointConnector",
    "config": {
        "connector.class": "org.apache.kafka.connect.mirror.MirrorCheckpointConnector",
        "tasks.max": "30",

        "source.cluster.alias": "A",
        "target.cluster.alias": "B",
        "source.cluster.bootstrap.servers": "<SOURCE_BOOTSTRAP_SERVERS>",
        "target.cluster.bootstrap.servers": "<TARGET_BOOTSTRAP_SERVERS>",

        "replication.policy.class": "org.apache.kafka.connect.mirror.IdentityReplicationPolicy",
        "offset-syncs.topic.location": "target",

        "checkpoints.topic.replication.factor": "-1",
        "sync.group.offsets.enabled": "true",
        "groups": "<GROUPS>"
    }
}

Most production environments use authorizations to control access to Kafka resources. In that case, to run MirrorMaker, you must grant permissions to the users employed by the Kafka Connect workers and the MirrorMaker connectors for both source and target clusters.

$ bin/kafka-acls.sh \
  --bootstrap-server <TARGET_BOOTSTRAP_SERVERS> \
  --add --allow-principal <PRINCIPAL> \
  --operation Read \
  --operation Write \
  --operation Create \
  --operation Describe \
  --topic mm2-offset-syncs.<SOURCE_ALIAS>.internal

Properties props = new Properties();
props.put(AdminClientConfig.BOOTSTRAP_SERVERS_CONFIG, "<TARGET_BOOTSTRAP_SERVERS>");

try (Admin admin = Admin.create(props)) {
    String user = "<PRINCIPAL>";
    String topic = "mm2-offset-syncs.<SOURCE_ALIAS>.internal";

    ResourcePattern resource = new ResourcePattern(
        ResourceType.TOPIC,
        topic,
        PatternType.LITERAL
    );

    List<AclBinding> acls = List.of(
        new AclBinding(resource, new AccessControlEntry(user, "*", AclOperation.READ, AclPermissionType.ALLOW)),
        new AclBinding(resource, new AccessControlEntry(user, "*", AclOperation.WRITE, AclPermissionType.ALLOW)),
        new AclBinding(resource, new AccessControlEntry(user, "*", AclOperation.CREATE, AclPermissionType.ALLOW)),
        new AclBinding(resource, new AccessControlEntry(user, "*", AclOperation.DESCRIBE, AclPermissionType.ALLOW))
    );

    admin.createAcls(acls).all().get();
}

The MirrorSourceConnector mirrors data sequentially, so for each partition newer data is only mirrored after older data. This means that if Kafka Connect or the disaster recovery cluster don’t have enough capacity, there is a risk the disaster recovery cluster will never catch up.

The following formula estimates the time it takes for the disaster recovery cluster to catch up to the primary cluster:

t = B / (Rmm2 - Rin)

Where:

Rmm2 must be larger than Rin; otherwise the disaster recovery cluster will never catch up with the primary cluster.

For example, if the primary cluster has 100000 MB of existing data (B), the MirrorMaker cluster can handle 150MB/s (Rmm2), and incoming throughput on the primary cluster is 100MB/s (Rin), the estimated catch-up time for disaster recovery is approximately 33 minutes:

100000 / (150 - 100)   = 2000 seconds (~33 minutes)

You need to decide the starting position of the MirrorSourceConnector, as this determines the value of B. There are three options, earliest, latest or custom.

6.1.1. Earliest position

Copiar enlace

By default, the MirrorSourceConnector starts from "earliest", which means it starts mirroring from the beginning of all selected partitions. This sets B in the equation to its maximum value, effectively mirroring all existing data to the disaster recovery cluster, but with some drawbacks.

First, if there is a lot of data (such as topics using tiered storage) MirrorMaker begins with a huge backlog of data. To keep up, it must have a much higher throughput than the incoming traffic to the primary cluster.

Lastly, topics have retention policies to determine when to delete data. If these trigger and delete a segment while the MirrorSourceConnector is mirroring it, it may create gaps in the data in the disaster recovery cluster.

Figure 6.1. Offset 2 on the source partition has been deleted so it cannot be mirrored

When MirrorMaker tries to read offset 2, it gets an error from the source cluster and resets its position to offset 3, which is the current earliest offset.

Figure 6.2. Since offset 2 was not mirrored, there is a data gap in the target partition

So it effectively skipped offset 2. For these reasons, unless the source cluster only contains very little data, don’t mirror from the earliest position.

bin/kafka-get-offsets.sh \
  --bootstrap-server <PRIMARY_BOOTSTRAP_SERVERS> \
  --time <TIME> \
  --topic-partitions <TOPICS>

curl -X PUT http://<CONNECT>/connectors/A-to-B-MirrorSourceConnector/stop

curl -X DELETE http://<CONNECT>/connectors/A-to-B-MirrorSourceConnector/offsets

Having started Kafka Connect, you can now use the Kafka Connect REST API to start the MirrorMaker connectors.

First, start the MirrorSourceConnector. If you want to use a custom position, initialize it in the stopped state by setting "initial_state" to "STOPPED" in its configuration.

For example:

curl -X POST http://<CONNECT>/connectors \
     -H "Content-type: application/json" \
     -d '{
           "name": "A-to-B-MirrorSourceConnector",
           "config": {
             "connector.class": "org.apache.kafka.connect.mirror.MirrorSourceConnector",
             "tasks.max": "10",
             …
           },
           "initial_state": "STOPPED"
         }'

Then you can set its starting offsets via another call to the Kafka Connect REST API:

curl -X PATCH http://<CONNECT>/connectors/A-to-B-MirrorSourceConnector/offsets \
  -H "Content-type: application/json" \
  -d '{
       "offsets": [
         {
           "partition": {
             "cluster": "A",
             "topic": "my-topic",
             "partition": 0
           },
           "offset": {
             "offset": 100
           }
         }
       ]
    }'

Finally, you can start the connector using the following curl command:

curl -X PUT http://<CONNECT>/connectors/A-to-B-MirrorSourceConnector/resume

When using the earliest or latest starting position, don’t set the "initial_state" configuration. You can start the connector directly:

curl -X POST http://<CONNECT>/connectors \
     -H "Content-type: application/json" \
     -d '{
           "name": "A-to-B-MirrorSourceConnector",
           "config": {
             "connector.class": "org.apache.kafka.connect.mirror.MirrorSourceConnector",
             "tasks.max": "10",
             …
           }
         }'

MirrorCheckpointConnector does not maintain a position, so you can start it directly:

curl -X POST http://<CONNECT>/connectors \
     -H "Content-type: application/json" \
     -d '{
           "name": "A-to-B-MirrorCheckpointConnector",
           "config": {
             "connector.class": "org.apache.kafka.connect.mirror.MirrorCheckpointConnector",
             "tasks.max": "10",
             …
           }
         }'

OffsetSync
  {topicPartition=<TOPIC_PARTITION>,
  upstreamOffset=<SOURCE_OFFSET>,
  downstreamOffset=<TARGET_OFFSET>
}

bin/kafka-configs.sh
  --bootstrap-server <TARGET_BOOTSTRAP_SERVERS> \
  --alter \
  --add-config segment.bytes=536870912 \
  --add-config min.compaction.lag.ms=172800000 \
  --entity-type topics \
  --entity-name mm2-offset-syncs.<SOURCE_CLUSTER_ALIAS>.internal

Checkpoint{
  consumerGroupId=<GROUP_ID>,
  topicPartition=<TOPIC_PARTITION>,
  upstreamOffset=<SOURCE_OFFSET>,
  downstreamOffset=<TARGET_OFFSET>,
  metadata=<METADATA>
}

bin/kafka-configs.sh
  --bootstrap-server <TARGET_BOOTSTRAP_SERVERS> \
  --alter \
  --add-config segment.bytes=52428800 \
  --entity-type topics \
  --entity-name <SOURCE_CLUSTER_ALIAS>.checkpoints.internal

It’s critical to collect, graph, and monitor metrics from Kafka Connect, MirrorMaker and the two Kafka clusters involved to operate a disaster recovery environment.

Each component emits a lot of metrics. Instead of trying to collect everything, select the key set of metrics that you want to monitor. Set up dashboards displaying the most important metrics. Most monitoring tools also allow setting up alerts when metrics reach certain thresholds, which can be used to detect unexpected states and notify operators.

Kafka Connect and MirrorMaker both expose metrics via JMX - under the domain names used by Kafka Connect in MBean format. If a different tool than JMX is used to export and monitor metrics, be aware that the metric names would look different depending on the tool used.

While the metrics allow you to monitor your environment, the REST API enables you to operate MirrorMaker and retrieve important information when you detect an issue.

Once the environment is running the most common operations are:

The following REST endpoints are useful for operating the connectors.

Check the status of a connector and its tasks:

curl -X GET http://<CONNECT>/connectors/<CONNECTOR>/status

Restart a connector and its tasks that have "failed" status by providing "includeTasks" and "onlyFailed" flags:

curl -X POST http://<CONNECT>/connectors/<CONNECTOR>/restart?includeTasks=true&onlyFailed=true

If "includeTasks" is not set or set to false, only the connector instance is restarted. If "onlyFailed" is not set or set to false, the connector and all of its tasks are restarted regardless of their status.

Update configuration such as "tasks.max" to increase the parallelism:

curl -X PATCH \
  -H "Content-Type: application/json" \
  -d '{"tasks.max": "10"}' \
  http://<CONNECT>/connectors/<CONNECTOR>/config

This automatically stops all the current tasks for the connector, applies the configuration update and creates the tasks based on the configured number, assigns them to available workers, and restarts the connector and the tasks. This endpoint is also used to update the other connector configurations.

Check the assignment of all the currently running tasks:

curl -X GET http://<CONNECT>/connectors/<CONNECTOR>/tasks

MirrorSourceConnector tasks have the task.assigned.partitions configuration that lists the partitions the task is mirroring. Similarly MirrorCheckpointConnector tasks have the task.assigned.groups configuration that lists consumer groups the task is translating offsets for.

Pause the connector and its tasks:

curl -X PUT http://<CONNECT>/connectors/<CONNECTOR>/pause

This stops message processing until the connector is resumed. The tasks are still assigned to workers and any resources claimed by them stays allocated. Once resumed, it is quick to start processing data.

Stop the connector and shut down its tasks:

curl -X PUT http://<CONNECT>/connectors/<CONNECTOR>/stop

This removes assigned tasks from the worker and releases any resources used by the tasks. It saves resource usage compared to the pause action, but it can take much longer to begin processing data once resumed.

Resume a stopped or paused connector and its tasks:

curl -X PUT http://<CONNECT>/connectors/<CONNECTOR>/resume

Kafka Connect has a number of other REST API endpoints, you can find them all in the Apache Kafka Connect REST API documentation.

When doing a planned failover or migration you can expect little to no data loss and a minimal impact on the applications. However, it requires certain conditions to be met before it can be done.

The first condition is to validate the gap (lag) between both clusters. MirrorMaker mirrors data asynchronously, so there is always a small data gap between the clusters. As described in the Metrics section, you can estimate the time delay between the clusters using the record-age-ms metric. For a planned failover, you ideally want this duration to be as short as possible. It should also match the RTO specified in the disaster recovery plan.

The second condition is the ability to shut down and restart applications in a controlled fashion, and with their configuration updated with the bootstrap servers for the other cluster. Before restarting, wait for the following to pass:

The default value for these intervals is the same (600 seconds). If you want to adjust the values, it’s best to keep them aligned to not complicate this process.

Once these conditions are met, you can start the process. In this example, failover is from cluster A to cluster B. In this initial state, cluster B is very close to cluster A, it is just missing a small amount of records as the mirroring process is asynchronous.

Figure 8.1. Mirroring is asynchronous, so there is always some data on the source cluster that has not been mirrored yet

Unfortunately, a failover is not always planned and if a hard failure impacts the primary cluster you may have to perform an unplanned failover. There is a large spectrum of possible failures, so the impact on your data and workload can vary significantly.

In the worst case, you can lose all data that has not been mirrored. The amount of data can be estimated using metrics. The record-age-ms metric indicates the gap in seconds between the two clusters. You can also use the last incoming byte rate from the source cluster to estimate the amount of data potentially lost.

The accuracy of the translated consumer group offsets is lower than in a planned failover because translation happens at a regular interval and the time of the last translation may be unknown.

Depending on the state of cluster A, attempting to recover data might be possible. However, plan for a full loss of the cluster. Do not rely on extracting data after a disaster.

Figure 8.5. Cluster A is impacted by a failure

In this scenario, the failover is straightforward, as you simply need to stop the two MirrorMaker connectors and then restart all your applications, now configured to connect to cluster B, which has become the new primary cluster.

Before restarting consuming applications, you can use the RemoteClusterUtils utility to translate consumer group offsets using the latest available checkpoints as this can produce slightly better offsets than what MirrorCheckpointConnector last committed. For example, you can run the following logic that translates consumer group offsets and commits them to the target cluster:

Map<String, Object> props = Map.of(
    CommonClientConfigs.BOOTSTRAP_SERVERS_CONFIG, "<TARGET_BOOTSTRAP_SERVERS>",
    MirrorClientConfig.REPLICATION_POLICY_CLASS, IdentityReplicationPolicy.class.getName()
);

// Get the translated offsets
Map<TopicPartition, OffsetAndMetadata> translatedOffsets =
    RemoteClusterUtils.translateOffsets(
        props, 

        "A", 

        "my-consumer-group", 

        Duration.ofSeconds(600)); 

Map<String, Object> adminProps = Map.of(
    CommonClientConfigs.BOOTSTRAP_SERVERS_CONFIG, "<TARGET_BOOTSTRAP_SERVERS>"
);
try (Admin admin = Admin.create(adminProps)) {
    // Use Admin API to override the offsets for the consumer group
    admin.alterConsumerGroupOffsets("my-consumer-group", translatedOffsets)
        .all()
        .get(10, TimeUnit.SECONDS);
}

The arguments are as follows:

Figure 8.6. The state of the disaster recovery environment after losing cluster A and performing an unplanned failover

Following a planned or unplanned failover, you don’t have a disaster recovery environment anymore, so you need to set up a new disaster recovery cluster.

The most straightforward approach to rebuild a disaster recovery environment is to restart from scratch once a new disaster recovery Kafka cluster is ready. In this case, the new disaster recovery cluster is empty and has no data. You can follow the same steps used when setting up the initial disaster recovery environment, but inverting the role of the cluster so that cluster B is the primary cluster and cluster A is the disaster recovery cluster.

Figure 9.1. Starting a new disaster recovery environment mirroring data from cluster B to cluster A

After a planned failover, you might want to fail back. This can be the case if you just practiced a planned failover, or if you executed a planned failover anticipating a failure and the failure did not happen. In this scenario, it is inefficient to delete all data from the second cluster and restart from scratch as this cluster has most of your data already. It would also significantly increase your recovery time and potentially leave you without an in-sync disaster recovery environment for a prolonged period of time. Instead, you can set up MirrorMaker to only mirror the data delta between the two clusters. For example, two records produced to my-topic-0 in cluster B after the failover are the delta:

Figure 9.2. Mirroring only the delta, the last 2 records in my-topic-0, between cluster B and cluster A

Following a planned failover, mirroring the delta is only possible if there is some data overlap between the two clusters, meaning that the primary cluster (cluster B) still has the log end offsets captured when failing over. In order to validate whether this is the case, you can use the kafka-get-offsets.sh tool to retrieve the log start offset for each partition and confirm they are smaller or equal to the log end offsets you captured earlier plus 1:

bin/kafka-get-offsets.sh \
  --bootstrap-server <PRIMARY_BOOTSTRAP_SERVERS> \
  --time earliest \
  --topic-partitions <TOPICS>

The arguments are:

For all the partitions that still contain the log end offset you captured during the disaster recovery, you can mirror only the data delta. You use the captured log end offsets as the starting positions when restarting MirrorSourceConnector. For the partitions that don’t contain the captured log end offsets, you need to pick their starting position explicitly, as described in the "Starting Position" section.

The following examples show how to determine whether you can mirror the delta or not. After the failover, cluster A is now the disaster recovery cluster, mirroring from primary cluster B. In both cases, the log end offset captured for my-topic-0 when failing over is 2.

In this first example, the earliest offset on the primary cluster is 0. Since it is smaller than 2, you can mirror only the delta (offsets 3 and 4). You can set the starting position to the next offset of the captured offset which is 3 in this case.

Figure 9.3. Cluster B log start offset is smaller than the log end offset captured when failing over

In this second example, the earliest offset of the source topic is 6, which is greater than the captured offset (2). Therefore, you have no delta you can mirror, the records with offsets from 3 to 5 are not on the clusters. In this case, you would have to choose the starting position explicitly, and you must clear all data from cluster A and reset mirroring as described in the Fresh Disaster Recovery Cluster section.

Figure 9.4. Cluster B log start offset is larger than the log end offset captured when failing over

In some cases a cluster may be preferred as the primary cluster after it is restored. This can be because it’s in a closer or cheaper datacenter.

Failing back is the act of switching the primary cluster back to the preferred cluster. Once the clusters are back in-sync, failing back is done by first doing a planned failover and then resetting MirrorMaker configuration to just mirror the data delta.

Suppose cluster B is the primary cluster that all your applications use, and MirrorMaker is mirroring data from cluster B to cluster A.

Figure 9.5. The state of the disaster recovery environment before failing back to cluster A

After validating all the prerequisites for a planned failover are achieved, you can fail over to cluster A:

Figure 9.6. The state of the disaster recovery environment after failing back to cluster A

At this point, you now need to reset mirroring. You should be able to mirror only the delta between the clusters by setting the custom starting positions of the MirrorSourceConnector:

Figure 9.7. The state of the disaster recovery environment mirroring data from cluster A to cluster B

Once MirrorMaker gets both clusters back in-sync, you have returned to the initial state.