Zero‑Downtime High‑Throughput Kafka MirrorMaker 2 Migration with Dual‑Write and Rollback

Background and Objectives

The article uses an e‑commerce order platform as a case study to illustrate how to migrate a Kafka cluster across data‑centers without service interruption. The migration must keep production traffic available, avoid message loss, duplication, or out‑of‑order delivery, sustain high throughput (peak 60 k msgs/s, 12 M daily orders, 96 partitions), and provide a rollback‑able, observable migration loop.

Applicable Scenarios

Old Kafka cluster to a new cluster

Cross‑region active‑active replication

Kafka version upgrades

Replication of topics, consumer‑group offsets, ACLs, and configurations

What MirrorMaker 2 Actually Does

MM2 is built on Kafka Connect and consists of a Connect cluster, MirrorSourceConnector, MirrorCheckpointConnector, MirrorHeartbeatConnector, and internal topics that store configuration, status, offsets, checkpoints, and heartbeats. The data flow is: MirrorSourceConnector consumes from the source cluster.

The records are sent in batches to the target cluster.

Target topics are created with the same partitions and ACLs. MirrorCheckpointConnector periodically reads source consumer‑group offsets.

It writes a mapping between source and target offsets to a checkpoint topic.

Consumers can use the checkpoint to migrate offsets smoothly.

Offset mapping is asynchronous, so the source offset and target offset are never identical; they are only mapped.

Limitations of MM2

Asynchronous replication introduces latency.

It cannot guarantee zero duplication or strict global ordering.

Application‑level idempotence, de‑duplication, and compensation are still required.

Production‑Grade Migration Architecture

A three‑layer architecture is recommended:

Data replication layer : MM2 handles topic and offset copying.

Application layer : Dual‑write, gray‑scale switches, and idempotent consumption ensure smooth cut‑over.

Operations layer : Monitoring, load testing, reconciliation, and rollback plans keep risk under control.

The recommended deployment uses a distributed Kafka Connect cluster (≥3 workers) instead of a single script, providing high availability, parallel scaling, unified operations, and rolling upgrades.

Sample MM2 Configuration (production‑ready)

# mm2.properties
clusters = source,target
source.bootstrap.servers = old-kafka-1:9092,old-kafka-2:9092,old-kafka-3:9092
target.bootstrap.servers = new-kafka-1:9092,new-kafka-2:9092,new-kafka-3:9092
# enable replication
source->target.enabled = true
source->target.topics = order-.*,payment-.*,inventory-.*,coupon-.*
source->target.groups = order-service-group,payment-service-group,inventory-service-group
replication.policy.class = org.apache.kafka.connect.mirror.IdentityReplicationPolicy
sync.topic.configs.enabled = true
sync.topic.acls.enabled = true
emit.heartbeats.enabled = true
emit.checkpoints.enabled = true
sync.group.offsets.enabled = true
refresh.topics.interval.seconds = 30
refresh.groups.interval.seconds = 30
emit.checkpoints.interval.seconds = 20
sync.group.offsets.interval.seconds = 20
tasks.max = 24
# producer settings for high throughput
producer.acks = all
producer.enable.idempotence = true
producer.max.in.flight.requests.per.connection = 5
producer.compression.type = lz4
producer.batch.size = 262144
producer.linger.ms = 20
producer.buffer.memory = 536870912
producer.retries = 2147483647
# exclude internal topics
topics.exclude = .*-internal,.*\.replica,__consumer_offsets
groups.exclude = console-consumer-.*,connect-.*,__.*

Key Configuration Parameters Explained

IdentityReplicationPolicy

keeps target topic names identical to source, avoiding extra prefix handling. tasks.max controls connector parallelism; set based on total partitions, worker CPU cores, network bandwidth, and target write capacity (e.g., 8 for light workloads, 16‑32 for medium, up to 48+ for large). producer.enable.idempotence=true reduces duplicate writes caused by retries, but application‑level idempotence is still required. sync.group.offsets.enabled=true is essential for smooth consumer cut‑over, though manual checkpoint validation is still needed.

Dual‑Write Implementation (Spring Boot)

A production‑grade dual‑write publisher sends to the primary cluster first, then optionally to the shadow cluster. It records metrics, handles failures via a retry queue, and allows topic‑level toggling.

package com.example.kafka.migration;

import java.time.Duration;
import java.util.Map;
import java.util.Set;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.CompletionStage;

import org.apache.kafka.clients.producer.RecordMetadata;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.kafka.support.SendResult;
import org.springframework.stereotype.Component;
import org.springframework.util.concurrent.ListenableFutureCallback;

@Component
public class DualWriteKafkaPublisher {
    private final KafkaTemplate<String, String> primaryKafkaTemplate;
    private final KafkaTemplate<String, String> shadowKafkaTemplate;
    private final MigrationTopicPolicy migrationTopicPolicy;
    private final FailedMirrorSendRepository failedMirrorSendRepository;
    private final MigrationMetrics migrationMetrics;

    public DualWriteKafkaPublisher(KafkaTemplate<String, String> primaryKafkaTemplate,
                                 KafkaTemplate<String, String> shadowKafkaTemplate,
                                 MigrationTopicPolicy migrationTopicPolicy,
                                 FailedMirrorSendRepository failedMirrorSendRepository,
                                 MigrationMetrics migrationMetrics) {
        this.primaryKafkaTemplate = primaryKafkaTemplate;
        this.shadowKafkaTemplate = shadowKafkaTemplate;
        this.migrationTopicPolicy = migrationTopicPolicy;
        this.failedMirrorSendRepository = failedMirrorSendRepository;
        this.migrationMetrics = migrationMetrics;
    }

    public CompletionStage<RecordMetadata> publish(String topic, String key, String payload) {
        CompletableFuture<RecordMetadata> result = new CompletableFuture<>();
        primaryKafkaTemplate.send(topic, key, payload).addCallback(new ListenableFutureCallback<SendResult<String, String>>() {
            @Override
            public void onSuccess(SendResult<String, String> primaryResult) {
                RecordMetadata metadata = primaryResult.getRecordMetadata();
                migrationMetrics.recordPrimarySuccess(topic);
                if (!migrationTopicPolicy.shouldDualWrite(topic)) {
                    result.complete(metadata);
                    return;
                }
                shadowKafkaTemplate.send(topic, key, payload).addCallback(new ListenableFutureCallback<SendResult<String, String>>() {
                    @Override
                    public void onSuccess(SendResult<String, String> shadowResult) {
                        migrationMetrics.recordShadowSuccess(topic);
                    }
                    @Override
                    public void onFailure(Throwable ex) {
                        migrationMetrics.recordShadowFailure(topic);
                        failedMirrorSendRepository.save(new FailedMirrorMessage(topic, key, payload, ex.getMessage()));
                    }
                });
                result.complete(metadata);
            }
            @Override
            public void onFailure(Throwable ex) {
                migrationMetrics.recordPrimaryFailure(topic);
                result.completeExceptionally(ex);
            }
        });
        return result.orTimeout(Duration.ofSeconds(10).toMillis(), java.util.concurrent.TimeUnit.MILLISECONDS);
    }

    @Component
    public static class MigrationTopicPolicy {
        private volatile Set<String> dualWriteTopics = Set.of("order-created", "payment-succeeded");
        public boolean shouldDualWrite(String topic) {
            return dualWriteTopics.contains(topic);
        }
        public void replaceTopics(Set<String> topics) {
            this.dualWriteTopics = Set.copyOf(topics);
        }
    }
}

Consumer Configuration (Spring Kafka)

package com.example.kafka.consumer;

import java.util.HashMap;
import java.util.Map;

import org.apache.kafka.clients.consumer.ConsumerConfig;
import org.apache.kafka.common.serialization.StringDeserializer;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.kafka.annotation.EnableKafka;
import org.springframework.kafka.config.ConcurrentKafkaListenerContainerFactory;
import org.springframework.kafka.core.DefaultKafkaConsumerFactory;

@EnableKafka
@Configuration
public class KafkaConsumerConfiguration {
    @Bean
    public ConcurrentKafkaListenerContainerFactory<String, String> migrationKafkaListenerFactory() {
        Map<String, Object> props = new HashMap<>();
        props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, "new-kafka-1:9092,new-kafka-2:9092,new-kafka-3:9092");
        props.put(ConsumerConfig.GROUP_ID_CONFIG, "order-service-group");
        props.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, false);
        props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);
        props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);
        props.put(ConsumerConfig.MAX_POLL_RECORDS_CONFIG, 500);
        props.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "latest");
        props.put(ConsumerConfig.ISOLATION_LEVEL_CONFIG, "read_committed");
        ConcurrentKafkaListenerContainerFactory<String, String> factory = new ConcurrentKafkaListenerContainerFactory<>();
        factory.setConsumerFactory(new DefaultKafkaConsumerFactory<>(props));
        factory.setConcurrency(12);
        factory.getContainerProperties().setAckMode(org.springframework.kafka.listener.ContainerProperties.AckMode.MANUAL);
        return factory;
    }
}

Runbook – Step‑by‑Step Migration Process

Phase 0 – Inventory & Load Test : List all topics, groups, ACLs, quotas, retention policies; identify critical vs. delay‑tolerant flows; benchmark target write/read capacity and cross‑region bandwidth.

Phase 1 – Deploy Target Cluster & MM2 : Provision new Kafka brokers, create required topics/ACLs, launch a 3‑node Connect cluster, enable topic/config/offset sync, verify internal topics and heartbeats.

Phase 2 – Historical Catch‑Up : Run MM2 without changing production traffic; monitor lag, task failures, rebalance counts; ensure large topics are fully caught up.

Phase 3 – Enable Dual‑Write : Start shadow writes for low‑risk topics; watch target TPS, error rates, and shadow‑write failure metrics; feed failures into compensation queue.

Phase 4 – Gray‑Scale Consumer Cut‑Over : Switch consumer groups batch‑by‑batch (low‑risk → high‑risk); after each batch verify latency, error rates, offset continuity, and business KPIs; roll back a batch if anomalies appear.

Phase 5 – Primary Producer Switch : Once most consumers run on target, switch main producers; keep dual‑write as a buffer for a few peak cycles.

Phase 6 – Decommission Source : Stop source writes, confirm no critical consumers remain, perform final topic/lag/offset validation, and either shut down MM2 or repurpose it for disaster‑recovery replication.

Monitoring & Alerting

Four categories of metrics are essential:

MM2/Connect : connector and task status, restart count, record poll/write rates, batch size, retry and dead‑letter counts.

Broker : BytesIn/Out, request handler idle %, under‑replicated partitions, ISR changes, request latency, disk usage, page‑cache hit ratio.

Consumer : per‑group lag, rebalance frequency, consumption error rate, processing latency, idempotent‑dedup hit count.

Business : order success rate, payment success rate, inventory lock failure, order‑state latency, compensation backlog.

Sample alert thresholds include MM2 task non‑RUNNING >1 min, checkpoint stale >2 min, target lag >30 s, shadow‑write failure rate >0.1 %, order latency >2 min, and broker disk >75 % (warning) / >85 % (critical).

Performance Tuning for High Concurrency

Increase producer.batch.size, linger.ms, consumer.max.poll.records, and max.partition.fetch.bytes to boost batch throughput.

Raise tasks.max and add more Connect workers; split hot topics into dedicated connectors.

Scale target broker resources (CPU, network, disk) and increase partition count after migration completes.

Fault Handling & Rollback

Common failures such as stale checkpoints, ACL mismatches, replication lag, or duplicate consumption are addressed with explicit troubleshooting steps and mitigation actions (e.g., enable emit.checkpoints.enabled, verify ACL sync, expand bandwidth, isolate hot topics). Rollback is a staged process: pause further consumer cut‑over, revert switched consumer groups batch‑by‑batch, switch producers back, keep dual‑write for safety, and collect logs for post‑mortem.

When Not to Use MM2

Strict exactly‑once transactional accounting across clusters.

Global strict ordering requirements.

Sub‑millisecond latency guarantees.

Bidirectional same‑topic read/write scenarios.

These cases typically require application‑level CDC with outbox patterns, log‑based replay systems, or dedicated disaster‑recovery platforms.

Conclusion

The migration is not merely “run a script”; it is a comprehensive, production‑ready framework that combines MM2 replication, dual‑write pre‑heating, gray‑scale cut‑over, rigorous monitoring, data validation, and a well‑defined rollback plan. Only by treating the migration as an engineered system can teams achieve true zero‑downtime, high‑throughput Kafka cluster upgrades.