10.2.2 – MirrorMaker and MM2 Architecture

Module: M10 – Advanced Topics & Architecture
Section: 10.2 – Multi-Datacenter Replication & Geo Architecture
Goal: Master MirrorMaker 2 architecture and configuration for cross-cluster replication

Introduction

MirrorMaker 2 (MM2) is the standard tool for replicating data between Kafka clusters. Built on Kafka Connect, MM2 provides topic replication, offset translation, and consumer group synchronization.

This module covers architecture only. Connector basics are in M8: Kafka Connect.

MirrorMaker vs MirrorMaker 2

MirrorMaker (Legacy)

Architecture: Simple consumer-producer pair

Source Cluster Consumer → Mirror Maker Process → Target Cluster Producer

Limitations:

• No offset translation
• No consumer group sync
• Manual configuration
• Single-threaded

Status: Deprecated in Kafka 3.0+

MirrorMaker 2 (Current)

Architecture: Kafka Connect-based framework

Source Cluster ← MM2 Connectors → Target Cluster
  - MirrorSourceConnector (replicate topics)
  - MirrorCheckpointConnector (sync consumer offsets)
  - MirrorHeartbeatConnector (health monitoring)

Features: ✅ Automatic topic creation
✅ Offset translation
✅ Consumer group synchronization
✅ Bidirectional replication
✅ Topic renaming (prefix/suffix)
✅ ACL and config replication

MM2 Internal Architecture & Flow

Step-by-Step: How MM2 Works

Let's trace a complete replication flow from Cluster A (us-east) → Cluster B (us-west).

Step 1: MM2 Initialization

1. MM2 Connect cluster starts
2. Loads configuration from mm2.properties
3. Discovers source and target clusters
4. Initializes three main connectors per replication flow

Step 2: MirrorSourceConnector Startup

What Happens:

1. Topic Discovery: Periodically scans source cluster for topics matching patterns
   • Default: every 10 minutes (refresh.topics.interval.seconds=600)
   • Discovers new topics automatically
2. Topic Creation: Creates replicated topics on target cluster
   • Naming: <source-alias>.<topic-name> (e.g., us-east.orders)
   • Copies partition count and configs (if sync.topic.configs.enabled=true)
3. Consumer Group: Creates internal consumer group on source cluster
   • Group ID: mirrormaker2-cluster-<target-alias>
4. Producer: Creates internal producer writing to target cluster

Internal Flow:

MirrorSourceConnector
  ↓ (acts as)
Consumer (ClusterA) → reads messages → Producer (ClusterB)
  ↓
Writes to: ClusterA.topic1, ClusterA.topic2 (on ClusterB)

Step 3: Message Replication (MirrorSourceConnector)

Per Message:

1. Poll from Source: Consumer polls messages from source topic
2. Transform: Adds replication metadata:
   • Original topic name
   • Original partition
   • Original timestamp (preserved)
3. Produce to Target: Producer sends to target cluster
   • Topic: <source-alias>.<original-topic>
   • Partition: Same partition number (preserves ordering)
4. Offset Commit: Commits consumer offset on source cluster

Batching: Batches 100-1000 messages per poll for efficiency.

Step 4: MirrorCheckpointConnector Startup

What Happens:

1. Monitors Consumer Groups: Tracks all consumer groups on source cluster
2. Reads Offsets: Periodically reads committed offsets from __consumer_offsets (source)
3. Translates Offsets: Maps source offsets to target offsets using replication history
4. Writes Checkpoints: Writes checkpoint records to target cluster
   • Topic: mm2-offsets.<source-alias>.internal

Checkpoint Record Structure:

{
  "groupId": "payment-processor",
  "topic": "orders",
  "partition": 0,
  "offset": 1000,
  "metadata": "",
  "commitTimestamp": 1643723400000,
  "expireTimestamp": 0,
  "leaderEpoch": 0
}

Step 5: MirrorHeartbeatConnector Startup

What Happens:

1. Emits Heartbeats: Sends periodic heartbeat messages to source cluster
2. Heartbeat Topic: heartbeats (on both clusters)
3. Heartbeat Message: Contains timestamp and source cluster ID
4. Purpose: Consumers can measure lag by comparing heartbeat timestamps

Heartbeat Message:

{
  "sourceCluster": "us-east",
  "targetCluster": "us-west",
  "timestamp": 1643723400000
}

Step 6: MirrorMetadataManager (Internal Component)

Purpose: In-memory manager for sync state and metadata.

What It Tracks:

• Topic replication status
• Offset translation mappings
• Heartbeat history
• Consumer group checkpoints

Not user-facing, but critical for MM2 internals.

MM2 Component Architecture Diagram

[Source Cluster A (us-east)]
  |
  | (1) MirrorSourceConnector
  |     - Consumer: reads from topics
  |     - Producer: writes to target
  |
  | (2) MirrorCheckpointConnector
  |     - Reads: __consumer_offsets
  |     - Writes: mm2-offsets.us-east.internal (on target)
  |
  | (3) MirrorHeartbeatConnector
  |     - Writes: heartbeats (on both clusters)
  |
  ↓
[Kafka Connect + MM2 Workers]
  ↓
[Target Cluster B (us-west)]
  - Topics: us-east.topic1, us-east.topic2
  - Internal Topics:
      - mm2-offsets.us-east.internal
      - __mm2_heartbeats

MM2 Internal Topics Reference

MM2 Components (Detailed)

1. MirrorSourceConnector

Purpose: Replicates topic data from source to target.

What It Does:

• Creates topics on target cluster (auto or manual)
• Replicates messages (batched for efficiency)
• Preserves message timestamps
• Adds cluster alias prefix to topic names

Example:

Source Cluster: Topic "orders"
Target Cluster: Topic "us-east.orders" (with prefix "us-east")

Config:

connector.class=org.apache.kafka.connect.mirror.MirrorSourceConnector
source.cluster.alias=us-east
target.cluster.alias=us-west
topics=orders,payments
replication.factor=3

Advanced Config:

# Topic discovery
refresh.topics.interval.seconds=600  # How often to discover new topics

# Topic filtering
topics.regex=orders.*,payments.*     # Regex pattern for topics to replicate
topics.blacklist=test.*,temp.*       # Topics to exclude

# Replication settings
replication.factor=3                 # Replication factor on target
sync.topic.configs.enabled=true      # Copy topic configs from source

# Offset management
offset-syncs.topic.replication.factor=3
offset-syncs.topic.location=target   # Where to store offset sync data

2. MirrorCheckpointConnector

Purpose: Synchronizes consumer group offsets.

What It Does:

• Translates offsets from source to target
• Writes checkpoint records to mm2-offsets.<source-alias>.internal
• Enables consumer failover (resume from last checkpoint)

Why It Matters: Without offset sync, consumers fail over to beginning or end (data loss or duplication).

Example:

Source Cluster:
  Consumer Group "payment-processor" at offset 1000 on "orders"

Target Cluster (after checkpoint):
  Consumer Group "payment-processor" can resume at translated offset 1000 on "us-east.orders"

Config:

connector.class=org.apache.kafka.connect.mirror.MirrorCheckpointConnector
source.cluster.alias=us-east
target.cluster.alias=us-west
groups=payment-processor,inventory-service
emit.checkpoints.interval.seconds=60

Advanced Config:

# Consumer group filtering
groups.regex=payment.*,order.*       # Only checkpoint these groups
groups.blacklist=test.*,temp.*       # Exclude these groups

# Checkpoint settings
emit.checkpoints.interval.seconds=60  # How often to checkpoint
checkpoints.topic.replication.factor=3

# Offset sync
sync.group.offsets.enabled=true       # Enable offset synchronization
sync.group.offsets.interval.seconds=60

3. MirrorHeartbeatConnector

Purpose: Monitors replication lag and cluster health.

What It Does:

• Sends heartbeat messages to __mm2_heartbeats topic
• Consumers can measure replication lag
• Detects cluster connectivity issues

Config:

connector.class=org.apache.kafka.connect.mirror.MirrorHeartbeatConnector
source.cluster.alias=us-east
target.cluster.alias=us-west
emit.heartbeats.interval.seconds=5

Advanced Config:

# Heartbeat settings
emit.heartbeats.enabled=true          # Enable/disable heartbeats
emit.heartbeats.interval.seconds=5    # Heartbeat frequency

# Heartbeat topic
heartbeats.topic.replication.factor=3

MM2 Configuration Deep Dive

Complete Configuration Reference

Standalone Mode (Comprehensive Example)

Config File (mm2.properties):

# ============================================================
# Cluster Definitions
# ============================================================
clusters = us-east, us-west

# Source cluster (us-east)
us-east.bootstrap.servers = broker1-east:9092,broker2-east:9092,broker3-east:9092
us-east.security.protocol = SASL_SSL
us-east.sasl.mechanism = PLAIN
us-east.sasl.jaas.config = org.apache.kafka.common.security.plain.PlainLoginModule required username="admin" password="secret";

# Target cluster (us-west)
us-west.bootstrap.servers = broker1-west:9092,broker2-west:9092,broker3-west:9092
us-west.security.protocol = SASL_SSL
us-west.sasl.mechanism = PLAIN
us-west.sasl.jaas.config = org.apache.kafka.common.security.plain.PlainLoginModule required username="admin" password="secret";

# ============================================================
# Replication Flow Configuration
# ============================================================

# Enable replication: us-east → us-west
us-east->us-west.enabled = true

# Enable replication: us-west → us-east (bidirectional)
us-west->us-east.enabled = true

# ============================================================
# Topic Selection & Filtering
# ============================================================

# Replicate specific topics (regex patterns)
us-east->us-west.topics = orders.*, payments.*, inventory.*

# Exclude topics (blacklist)
us-east->us-west.topics.blacklist = test.*, temp.*, _.*

# Auto-create topics on target
us-east->us-west.auto.create.topics.enable = true

# ============================================================
# Replication Settings
# ============================================================

# Replication factor for replicated topics
us-east->us-west.replication.factor = 3

# Copy topic configs from source to target
us-east->us-west.sync.topic.configs.enabled = true

# Topic discovery frequency (seconds)
us-east->us-west.refresh.topics.interval.seconds = 600

# ============================================================
# Offset & Checkpoint Configuration
# ============================================================

# Enable consumer offset checkpointing
us-east->us-west.emit.checkpoints.enabled = true
us-east->us-west.emit.checkpoints.interval.seconds = 60

# Offset sync topic replication
us-east->us-west.offset-syncs.topic.replication.factor = 3

# Consumer groups to checkpoint
us-east->us-west.groups = payment-processor, inventory-service

# Consumer groups to exclude
us-east->us-west.groups.blacklist = test.*, temp.*

# Enable group offset sync
us-east->us-west.sync.group.offsets.enabled = true
us-east->us-west.sync.group.offsets.interval.seconds = 60

# ============================================================
# Heartbeat Configuration
# ============================================================

# Enable heartbeats
us-east->us-west.emit.heartbeats.enabled = true
us-east->us-west.emit.heartbeats.interval.seconds = 5

# Heartbeat topic replication
us-east->us-west.heartbeats.topic.replication.factor = 3

# ============================================================
# Producer & Consumer Tuning
# ============================================================

# Producer settings
us-east->us-west.producer.batch.size = 65536           # 64 KB
us-east->us-west.producer.linger.ms = 10
us-east->us-west.producer.compression.type = snappy
us-east->us-west.producer.acks = all
us-east->us-west.producer.max.in.flight.requests.per.connection = 5

# Consumer settings
us-east->us-west.consumer.fetch.min.bytes = 1048576    # 1 MB
us-east->us-west.consumer.max.poll.records = 1000
us-east->us-west.consumer.session.timeout.ms = 30000

# ============================================================
# Connect Worker Configuration
# ============================================================

# Tasks per connector (parallelism)
us-east->us-west.tasks.max = 8

# Offset storage
offset.storage.topic = mm2-offsets
offset.storage.replication.factor = 3

# Config storage
config.storage.topic = mm2-config
config.storage.replication.factor = 3

# Status storage
status.storage.topic = mm2-status
status.storage.replication.factor = 3

# ============================================================
# Topic Naming & Replication Policy
# ============================================================

# Topic naming policy (default: DefaultReplicationPolicy)
replication.policy.class = org.apache.kafka.connect.mirror.DefaultReplicationPolicy

# Replication policy separator (default: ".")
replication.policy.separator = .

# ============================================================
# Performance & Monitoring
# ============================================================

# Max bytes per replication request
us-east->us-west.max.message.bytes = 1048576  # 1 MB

# Timeout for replication requests
us-east->us-west.request.timeout.ms = 30000

# Metrics reporters
metric.reporters = org.apache.kafka.common.metrics.JmxReporter

Run:

bin/connect-mirror-maker.sh mm2.properties

Key Configuration Parameters Explained

Configuration Examples by Use Case

1. High-Throughput Replication

us-east->us-west.tasks.max = 16
us-east->us-west.producer.batch.size = 131072  # 128 KB
us-east->us-west.producer.linger.ms = 50
us-east->us-west.producer.compression.type = zstd
us-east->us-west.consumer.fetch.min.bytes = 10485760  # 10 MB
us-east->us-west.consumer.max.poll.records = 5000

2. Low-Latency Replication

us-east->us-west.tasks.max = 4
us-east->us-west.producer.batch.size = 16384  # 16 KB
us-east->us-west.producer.linger.ms = 0       # Send immediately
us-east->us-west.consumer.max.poll.records = 100
us-east->us-west.emit.checkpoints.interval.seconds = 10

3. Resource-Constrained Replication

us-east->us-west.tasks.max = 2
us-east->us-west.producer.batch.size = 8192   # 8 KB
us-east->us-west.consumer.max.poll.records = 100
us-east->us-west.refresh.topics.interval.seconds = 1200  # 20 min
us-east->us-west.emit.checkpoints.interval.seconds = 300  # 5 min

Distributed Mode (Kafka Connect)

Submit connectors via REST API:

{
  "name": "us-east-to-us-west-source",
  "config": {
    "connector.class": "org.apache.kafka.connect.mirror.MirrorSourceConnector",
    "source.cluster.alias": "us-east",
    "target.cluster.alias": "us-west",
    "source.cluster.bootstrap.servers": "broker1-east:9092",
    "target.cluster.bootstrap.servers": "broker1-west:9092",
    "topics": "orders,payments",
    "replication.factor": 3
  }
}

Offset Translation

The Problem

Source Cluster:

Topic: orders, Partition: 0
  Offset 0: message A
  Offset 1: message B
  Offset 2: message C

Target Cluster (after replication):

Topic: us-east.orders, Partition: 0
  Offset 0: message A
  Offset 1: message B
  Offset 2: message C

Consumer Issue: Consumer on source was at offset 2. After failover to target, where should it resume?

MM2 Solution: Checkpoint Topics

MM2 writes checkpoint records mapping source offsets to target offsets:

Checkpoint Record:

{
  "sourceCluster": "us-east",
  "sourceT"opic": "orders",
  "sourcePartition": 0,
  "sourceOffset": 2,
  "targetCluster": "us-west",
  "targetTopic": "us-east.orders",
  "targetPartition": 0,
  "targetOffset": 2
}

Consumer Failover:

1. Consumer reads checkpoint topic
2. Translates source offset 2 → target offset 2
3. Resumes consumption from offset 2 on target cluster

Bidirectional Replication & Loop Prevention

The Loop Problem

Scenario: Active-active replication without loop prevention

Cluster A (us-east):
  Topic: "orders"
  
Cluster B (us-west):
  Topic: "orders"

Replication Flow (BROKEN):
1. Cluster A: "orders" → replicated to B as "us-east.orders"
2. Cluster B: "us-east.orders" → replicated back to A as "us-west.us-east.orders"
3. Cluster A: "us-west.us-east.orders" → replicated to B as "us-east.us-west.us-east.orders"
4. Infinite loop! 🔥

MM2 Loop Prevention Mechanism

How MM2 Prevents Loops:

1. Source Cluster Alias in Topic Name:

   • MM2 adds the source cluster alias as a prefix to replicated topics
   • Example: us-east.orders (replicated from us-east)

2. Topic Filtering:

   • MM2 detects the source cluster alias in topic names
   • Excludes topics that already contain a cluster alias prefix
   • Prevents re-replication of already-replicated topics

Example:

Cluster A (us-east): Topic "orders"
  ↓ (replicated to B)
Cluster B (us-west): Topic "us-east.orders"
  ↓ (B tries to replicate back to A)
  ✅ BLOCKED: Topic name starts with "us-east." (source cluster prefix)
  → MM2 recognizes this is already a replicated topic
  → Does NOT replicate back to us-east

Loop Prevention Configuration

Method 1: Automatic (Default Behavior)

MM2 automatically excludes topics with cluster alias prefixes:

clusters = us-east, us-west

# Bidirectional replication
us-east->us-west.enabled = true
us-west->us-east.enabled = true

# MM2 automatically excludes:
# - Topics starting with "us-east." when replicating from us-west
# - Topics starting with "us-west." when replicating from us-east

Result: Local topics replicated, but replicated topics are NOT re-replicated.

Method 2: Explicit Regex Filtering

For more control, use negative lookahead regex:

clusters = us-east, us-west

# us-east → us-west: Replicate all EXCEPT topics starting with "us-west."
us-east->us-west.enabled = true
us-east->us-west.topics = (?!us-west\\.).*

# us-west → us-east: Replicate all EXCEPT topics starting with "us-east."
us-west->us-east.enabled = true
us-west->us-east.topics = (?!us-east\\.).*

Regex Explanation:

• (?!us-west\\.) = Negative lookahead: NOT starting with "us-west."
• .* = Match any remaining characters

Method 3: Custom Replication Policy

Implement a custom replication policy to control topic naming:

public class CustomReplicationPolicy extends DefaultReplicationPolicy {
    @Override
    public String formatRemoteTopic(String sourceClusterAlias, String topic) {
        // Custom logic for topic naming
        if (topic.startsWith(sourceClusterAlias + ".")) {
            return null;  // Don't replicate (already replicated)
        }
        return sourceClusterAlias + "." + topic;
    }
}

replication.policy.class = com.example.CustomReplicationPolicy

Active-Active Configuration (Complete Example)

# ============================================================
# Bidirectional Active-Active Setup
# ============================================================

clusters = us-east, us-west

# Cluster bootstrap servers
us-east.bootstrap.servers = broker1-east:9092,broker2-east:9092
us-west.bootstrap.servers = broker1-west:9092,broker2-west:9092

# ============================================================
# us-east → us-west Replication
# ============================================================

us-east->us-west.enabled = true
us-east->us-west.topics = (?!us-west\\.).*  # Exclude already-replicated topics
us-east->us-west.emit.checkpoints.enabled = true
us-east->us-west.emit.heartbeats.enabled = true
us-east->us-west.sync.group.offsets.enabled = true

# ============================================================
# us-west → us-east Replication
# ============================================================

us-west->us-east.enabled = true
us-west->us-east.topics = (?!us-east\\.).*  # Exclude already-replicated topics
us-west->us-east.emit.checkpoints.enabled = true
us-west->us-east.emit.heartbeats.enabled = true
us-west->us-east.sync.group.offsets.enabled = true

# ============================================================
# Replication Policy (Default)
# ============================================================

replication.policy.class = org.apache.kafka.connect.mirror.DefaultReplicationPolicy

Result:

Cluster A Topics:
  - orders (local)
  - payments (local)
  - us-west.orders (replicated from us-west)
  - us-west.payments (replicated from us-west)

Cluster B Topics:
  - orders (local)
  - payments (local)
  - us-east.orders (replicated from us-east)
  - us-east.payments (replicated from us-east)

Verifying Loop Prevention

Check Topic List on Both Clusters:

# On us-east
kafka-topics --list --bootstrap-server us-east:9092

# Expected:
# orders
# payments
# us-west.orders  ← replicated from us-west
# us-west.payments

# Should NOT see:
# us-west.us-east.orders  ← this would indicate a loop

Monitor Replication Metrics:

# Check MM2 connector logs for filtering
grep "Skipping topic" /var/log/kafka/connect.log

# Expected output:
# Skipping topic us-west.orders (matches blacklist)

Performance and Scaling

Throughput Tuning

Increase parallelism:

tasks.max = 8  # More tasks = more parallelism

Batch settings:

producer.batch.size = 65536  # 64 KB
producer.linger.ms = 10
producer.compression.type = snappy

Consumer settings:

consumer.fetch.min.bytes = 1048576  # 1 MB
consumer.max.poll.records = 1000

MM2 Monitoring & Troubleshooting

MM2 Internal Topics Reference

Inspecting MM2 Internal Topics

1. Check Checkpoint Data

Purpose: Verify offset checkpointing is working

kafka-console-consumer --bootstrap-server target:9092 \
  --topic mm2-offsets.us-east.internal \
  --from-beginning \
  --property print.key=true \
  --property key.separator=" | "

Sample Output:

[payment-processor,orders,0] | {"groupId":"payment-processor","topic":"orders","partition":0,"offset":1000,"metadata":"","commitTimestamp":1643723400000}
[inventory-service,products,1] | {"groupId":"inventory-service","topic":"products","partition":1,"offset":5420,"metadata":"","commitTimestamp":1643723410000}

2. Check Heartbeat Data

Purpose: Measure replication lag

kafka-console-consumer --bootstrap-server target:9092 \
  --topic __mm2_heartbeats \
  --from-beginning \
  --property print.key=true

Sample Output:

us-east | {"sourceCluster":"us-east","targetCluster":"us-west","timestamp":1643723400000}
us-west | {"sourceCluster":"us-west","targetCluster":"us-east","timestamp":1643723405000}

Calculate Lag:

# Compare heartbeat timestamp with current time
current_time=$(date +%s%3N)
heartbeat_time=1643723400000
lag_ms=$((current_time - heartbeat_time))
echo "Replication lag: ${lag_ms}ms"

3. Check Offset Sync Data

Purpose: Verify offset translation is working

kafka-console-consumer --bootstrap-server target:9092 \
  --topic __mm2_offsets_sync \
  --from-beginning \
  --property print.key=true

Monitoring with JMX Metrics

Enable JMX on MM2 Connect Workers:

export KAFKA_JMX_OPTS="-Dcom.sun.management.jmxremote.port=9999 \
  -Dcom.sun.management.jmxremote.authenticate=false \
  -Dcom.sun.management.jmxremote.ssl=false"

Key MM2 Metrics Table

Access Metrics via JConsole:

jconsole localhost:9999
# Navigate to: MBeans → kafka.connect → source-task-metrics

Prometheus Exporter Configuration (kafka-connect.yml):

rules:
- pattern: kafka.connect<type=source-task-metrics, connector=(.+), task=(.+)><>source-record-poll-total
  name: kafka_connect_source_record_poll_total
  labels:
    connector: "$1"
    task: "$2"

- pattern: kafka.connect<type=source-task-metrics, connector=(.+), task=(.+)><>source-record-write-total
  name: kafka_connect_source_record_write_total
  labels:
    connector: "$1"
    task: "$2"

- pattern: kafka.connect<type=connector-task-metrics, connector=(.+), task=(.+)><>batch-size-avg
  name: kafka_connect_batch_size_avg
  labels:
    connector: "$1"
    task: "$2"

Grafana Dashboard Queries (PromQL):

# Replication throughput (records/sec)
rate(kafka_connect_source_record_write_total[5m])

# Replication lag (records)
kafka_connect_source_record_poll_total - kafka_connect_source_record_write_total

# Average batch size
kafka_connect_batch_size_avg

# Alert: No records written in 5 minutes
rate(kafka_connect_source_record_write_total[5m]) == 0

Replication Lag Monitoring

Method 1: Consumer Lag Tool

kafka-consumer-groups --bootstrap-server target:9092 \
  --describe --group mirrormaker2-cluster-us-west

# Output:
# GROUP                           TOPIC           PARTITION  CURRENT-OFFSET  LOG-END-OFFSET  LAG
# mirrormaker2-cluster-us-west   orders          0          1000            1050            50
# mirrormaker2-cluster-us-west   payments        0          2000            2100            100

Method 2: Heartbeat Lag

# Read latest heartbeat timestamp
kafka-console-consumer --bootstrap-server target:9092 \
  --topic __mm2_heartbeats \
  --partition 0 \
  --offset latest \
  --max-messages 1 \
  --property print.timestamp=true

# Compare with current time to calculate lag

Method 3: Prometheus Query

# Time since last heartbeat
time() - (kafka_heartbeat_timestamp_seconds / 1000)

# Alert if lag > 60 seconds
(time() - (kafka_heartbeat_timestamp_seconds / 1000)) > 60

Troubleshooting Common MM2 Issues

Issue 1: Topics Not Replicating

Symptoms: Topics exist on source but not on target

Debugging Steps:

1. Check topic filter patterns:

# Verify topic matches regex
kafka-topics --list --bootstrap-server source:9092 | grep "orders"

# Check MM2 config
grep "topics" mm2.properties

2. Check connector status:

curl http://connect-worker:8083/connectors/mirror-source/status

3. Check connector logs:

grep "Skipping topic" /var/log/kafka/connect.log

Issue 2: High Replication Lag

Symptoms: Lag > 10,000 messages or > 60 seconds

Possible Causes:

• Insufficient parallelism (tasks.max too low)
• Network bottleneck
• Target cluster overloaded
• Large message sizes

Solutions:

# Increase parallelism
us-east->us-west.tasks.max = 16

# Increase batch size
us-east->us-west.producer.batch.size = 131072
us-east->us-west.consumer.max.poll.records = 5000

# Tune network
us-east->us-west.consumer.fetch.min.bytes = 10485760

Issue 3: Offset Translation Not Working

Symptoms: Consumers fail over to target but lose position

Debugging Steps:

1. Verify checkpointing is enabled:

us-east->us-west.emit.checkpoints.enabled = true

2. Check checkpoint topic exists:

kafka-topics --list --bootstrap-server target:9092 | grep "mm2-offsets"

3. Inspect checkpoint data (see "Inspecting MM2 Internal Topics" above)

MM2 Health Check Checklist

Daily Monitoring:

• Check replication lag (should be < 1000 messages or < 60 seconds)
• Verify all connectors are RUNNING status
• Check heartbeat lag (should be < 10 seconds)
• Monitor task failures (should be 0)
• Verify checkpointing is active (check mm2-offsets.* topic)

Weekly Monitoring:

• Review topic discovery (new topics should appear within refresh.topics.interval.seconds)
• Check for stuck consumers (replication consumer lag)
• Verify ACL/config sync (if enabled)
• Review disk usage on target cluster

Alert Configuration (Prometheus):

groups:
- name: mm2_replication
  rules:
  - alert: HighReplicationLag
    expr: kafka_consumer_lag{group="mirrormaker2-cluster-*"} > 10000
    for: 10m
    annotations:
      summary: "MM2 replication lag > 10K messages for {{ $labels.group }}"
      
  - alert: NoHeartbeat
    expr: time() - (kafka_heartbeat_timestamp_seconds / 1000) > 60
    for: 5m
    annotations:
      summary: "No MM2 heartbeat in 60 seconds"
      
  - alert: ConnectorDown
    expr: kafka_connect_connector_status{status!="RUNNING"} == 1
    for: 5m
    annotations:
      summary: "MM2 connector {{ $labels.connector }} is {{ $labels.status }}"

Key Takeaways

1. MM2: Kafka Connect-based tool for cross-cluster replication (replaces legacy MirrorMaker).

2. Three Connectors: MirrorSourceConnector (data), MirrorCheckpointConnector (offsets), MirrorHeartbeatConnector (health).

3. Offset Translation: Checkpoints map source offsets to target offsets for consumer failover.

4. Bidirectional: Supports active-active with loop prevention via topic prefixes.

5. Scaling: Increase tasks.max for parallelism, tune producer/consumer batching.

6. Monitoring: Track replication lag, throughput, and checkpoint sync rate.

What's Next?

Next Module: 10.2.3 – Active-Passive vs Active-Active

References:

• Kafka Connect basics → M8: Kafka Connect
• Multi-cluster patterns → 10.2.1