Multi-Cluster and Geo-Replication

MirrorMaker 2 has three connectors: MirrorSourceConnector copies topic data from the source cluster to the target; MirrorCheckpointConnector copies consumer-group offset progress; MirrorHeartbeatConnector emits periodic heartbeats to confirm the replication path is alive.

MM2 uses topics/topics.exclude and groups/groups.exclude regex lists (per flow). The 'topics'/'groups' allowlist says what to replicate; the 'exclude' denylist removes matches. By default internal and MM2 bookkeeping topics/groups are excluded.

When sync.topic.configs.enabled=true (the default), MM2 periodically copies the source topic's configuration (like cleanup.policy, retention, etc.) onto the mirrored target topic so they stay consistent. A property filter controls which config keys are synced.

MirrorSourceConnector divides the source topic-partitions it must replicate across tasks. Parallelism is bounded by tasks.max and by the number of source partitions — you never get more useful tasks than partitions, since each partition is handled by one task.

You define two flows (A->B and B->A), each running its own set of MM2 connectors. Loops are avoided because mirrored topics are prefixed with the source cluster alias, and MM2's default filters exclude already-remote topics, so B never re-mirrors A's data back to A.

Offset translation maps a consumer's position on the source cluster to the equivalent position on the target cluster. You can't reuse offsets directly because the same record gets a different offset on the replicated topic, so the raw number points to the wrong place.

MirrorCheckpointConnector reads source consumer-group offsets plus offset syncs and writes checkpoint records to the <source>.checkpoints.internal topic on the target. Each checkpoint stores a group, topic-partition, the upstream (source) offset, and the translated downstream (target) offset.

Check that emit.checkpoints is enabled, the group isn't excluded by groups/groups.exclude, the <source>.checkpoints.internal topic exists with data, the offset-syncs topic has syncs covering the group's partitions, and that you're querying the right target alias and renamed topics.

Call RemoteClusterUtils.translateOffsets to get translated target offsets for the group, commit them to the group on the target, then start consumers there pointed at the replicated topics. Prefer sync.group.offsets.enabled when you want MM2 to keep the target offsets continuously synced so failover needs no extra tooling.

MM2 emits offset syncs sparsely (not for every record) into the offset-syncs topic, and checkpoints are emitted on an interval. Between sync points and between emissions, the translated offset is approximate, causing translation lag and a bounded replay window on failover.

MirrorHeartbeatConnector periodically writes timestamped records to a 'heartbeats' topic on the source cluster. MM2 replicates them to the target, proving the replication path is alive and letting you measure how fast it flows.

MM2's MirrorSourceConnector exposes per-topic-partition JMX metrics: replication-latency-ms (time from source append to target append), record-age-ms (age of a record when consumed from source), and byte/record rates. You scrape these to track lag.

emit.heartbeats.interval.seconds sets how often MirrorHeartbeatConnector writes a heartbeat record (default 1s). emit.heartbeats.enabled (default true) turns the feature on/off. Shorter interval = finer latency resolution but more overhead.

Consume the replicated <source>.heartbeats topic on the target, read each record's emit timestamp, and compute now - timestamp. That delta is your end-to-end replication latency, which approximates worst-case RPO — how much data you'd lose on failover.

Stale heartbeats mean the flow is broken or stalled, not merely slow. Check the heartbeat connector/task state, the source heartbeats topic, and the MirrorSourceConnector replicating it. Rising replication-latency-ms instead means the flow works but is lagging.

By default MirrorMaker 2 prefixes the replicated topic with the source cluster's alias and a dot. A topic named 'orders' from cluster 'us-west' becomes 'us-west.orders' on the destination. This makes the topic's origin visible and prevents name clashes.

IdentityReplicationPolicy replicates topics keeping their original name (no source-cluster prefix), so 'orders' stays 'orders' on the destination. The trade-off: you lose automatic cycle detection and risk name collisions, so it's only safe for unidirectional (active/passive) replication.

MM2 prevents loops by encoding a topic's origin in its name via the source prefix. Before replicating, it checks whether a topic already originated from the destination cluster (or matches the configured source-cluster prefix) and skips it, so records don't bounce back forever.

replication.policy.separator is the character DefaultReplicationPolicy puts between the source cluster alias and the topic name. It defaults to a dot ('.'), so 'orders' becomes 'us-west.orders'. Changing it changes remote topic names; it must match everywhere or MM2 can't parse origins.

MM2 skips internal and system topics: Kafka's own __consumer_offsets and transaction-state topics, and MM2's own heartbeats, checkpoints, and offset-sync topics. It identifies them via the ReplicationPolicy's isInternalTopic check and the topic filter, so they aren't replicated as user data.

Cluster Linking is a broker-built-in feature that copies topics from a source Kafka cluster to a destination cluster over a direct link, with no separate Connect cluster. The destination gets read-only mirror topics that keep the same message offsets.

A mirror topic starts in an ACTIVE (mirroring, read-only) state. To make it writable you stop mirroring and convert it: 'promote' is a graceful, synced cutover (no data loss); 'failover' is an immediate cutover used when the source is unreachable, accepting possible loss of un-replicated records.

The link can be configured to periodically copy committed consumer-group offsets and ACLs from the source to the destination. Because offsets are byte-for-byte identical across clusters, the copied commit positions are directly valid, so consumers can resume at the same place after failover.

On the destination, you create a named link pointing at the source's bootstrap servers and security settings, then create mirror topics on that link. Key configs include the source bootstrap/security (SASL/SSL), whether to auto-create mirror topics, and the offset/ACL sync flags.

Run primary in region A, with a link in region B mirroring A's topics (read-only) plus offset and ACL sync. On a disaster, promote/failover the mirror topics in B and repoint producers/consumers there. Fail-back later means setting up a reverse link from B to A, catching it up, then cutting back.

In active-passive DR, one Kafka cluster (primary) serves all traffic while a second cluster (standby) only receives a one-way replicated copy. Clients use the standby only after failover. Active-active runs producers/consumers on both clusters at once.

RPO (Recovery Point Objective) is the max acceptable data loss, driven by replication lag — async replication means you can lose whatever hasn't reached the standby. RTO (Recovery Time Objective) is the max acceptable downtime, driven by how fast you detect, decide, and cut clients over.

A record's offset on the standby differs from its offset on the primary because replication starts at different points and may compact/skip. MirrorMaker 2's MirrorCheckpointConnector records the primary→standby offset mapping and writes translated consumer-group checkpoints so failed-over consumers resume near where they left off.

Planned failover (maintenance, drills) lets you stop producers, drain replication so the standby fully catches up, then cut over with RPO near zero. Unplanned failover (primary crashes) gives no chance to drain, so you lose whatever wasn't yet replicated — RPO equals the lag at failure.

The consumer reconnects to the standby's bootstrap servers, looks up its group's translated offsets (via RemoteClusterUtils/MirrorClient or pre-synced __consumer_offsets), seeks each partition to those offsets, then resumes. Pitfalls: stale checkpoints causing reprocessing, topic-name prefixes, and running the same group active on both clusters.

MM2 prefixes replicated topics with the source cluster's alias (e.g. topic 'orders' from cluster A becomes 'A.orders' on cluster B). Because the prefix marks where data came from, MM2 won't replicate a topic back to the cluster it originated from, so records don't loop forever.

Two regions can write conflicting updates to the same entity concurrently, and MM2 gives no cross-region ordering or conflict resolution — it just copies records. You handle conflicts at the application layer: attach idempotency keys so re-delivered or duplicate records are deduped, and use last-writer-wins, CRDTs, or entity-region affinity to resolve concurrent updates.

Each region has local topics (written there) plus remote mirrored topics from other regions. A region-local consumer reads just local topics; a globally-aware consumer reads both local and remote (prefixed) topics — via a regex subscription or by consuming a separate aggregate cluster that holds all regions' data in one place.

Define both clusters and enable replication in both directions in the MM2 config (A->B and B->A). MM2 runs three connectors per flow: MirrorSourceConnector (copies records, applies the prefix), MirrorCheckpointConnector (translates consumer offsets), and MirrorHeartbeatConnector (liveness). For failover, consumers use the translated checkpoint offsets to resume on the other cluster.

Choose active-active MM2 when both regions must serve low-latency local writes and tolerate eventual, asynchronously-replicated cross-region data. Avoid it when you need strong global consistency or strict ordering — a stretched cluster gives synchronous consistency (at WAN-latency cost), and active-passive gives simpler failover without write conflicts. The key limit to communicate: it's eventually consistent, at-least-once, with no global ordering or conflict resolution.

broker.rack labels each broker with its location (e.g. its availability zone). Kafka uses these labels to spread the copies of each partition across different zones, so losing one zone does not lose all copies of the data.

Use replication factor 3 (one replica per AZ), min.insync.replicas=2, and producers with acks=all. Then a write is only acknowledged after two zones have it, so losing one AZ still leaves a committed copy and writes keep working.

With acks=all the leader waits for in-sync replicas in other zones before acknowledging, so each write pays a cross-AZ round trip. That raises per-record latency. You hide it with batching (linger.ms, batch.size) and concurrency (in-flight requests), and by keeping zones close.

Follower fetching lets a consumer read from a nearby replica in its own AZ instead of always from the leader. This cuts cross-AZ network traffic and cost. You enable it with a broker replica.selector.class and a consumer client.rack matching its zone.

A 2.5-DC design runs Kafka across two full datacenters plus a tiny third site holding just a tie-breaker (a witness/observer). The third site holds no data but lets the cluster keep a majority quorum if one of the two main DCs fails, avoiding split-brain.

Normally Kafka consumers read only from the partition leader. Fetch-from-follower (KIP-392) lets a consumer read from a nearby replica (follower) in its own availability zone instead, cutting cross-zone network traffic and the cloud egress cost that comes with it.

On every broker set broker.rack to its AZ and set replica.selector.class to RackAwareReplicaSelector. On every consumer set client.rack to its own AZ. The broker then matches client.rack to broker.rack and points the consumer at a same-AZ follower.

A follower can only serve records up to its own high-watermark — the committed offset it has replicated and acknowledged. If the follower lags the leader, the consumer reads slightly older data, and end-to-end latency can grow, though it never reads uncommitted records.

Tag brokers with broker.rack=AZ, enable RackAwareReplicaSelector, ensure every partition has a replica in each consumer AZ (rack-aware placement), and set each consumer's client.rack to its AZ. Savings collapse if there's no in-rack replica, racks are misconfigured, or consumers fall back to the leader.

replica.selector.class plugs in a class implementing org.apache.kafka.common.replica.ReplicaSelector. Kafka ships LeaderSelector (default) and RackAwareReplicaSelector. You can write your own select() logic — e.g., choose by latency or a custom topology — returning the replica the consumer should read from.

MirrorMaker 2 renames remote topics with the source-cluster alias as a prefix (e.g. us-east.orders). A topic that already carries another cluster's prefix is recognized as remote and is not replicated again, breaking the loop.

Replication lag is how far behind the target cluster's copy is from the source. If you lose the source, any unreplicated records are lost, so lag directly sets your worst-case RPO (data-loss window). You bound it by monitoring lag/throughput and provisioning enough replication capacity.

Drift is when the target topic ends up with a different partition count or different topic configs than the source. It's dangerous because key-based ordering and partitioning break after failover. MM2 periodically syncs partition counts and topic configs to keep them matched.

Without the alias prefix, replicas share the original name, so producers can write to the 'same' topic on both clusters and replication has no way to tell origin from replica. You must use one-way flows or strict topics.exclude filters to avoid loops and collisions.

Inside one cluster you throttle replica traffic with leader.replication.throttled.rate / follower.replication.throttled.rate (set via kafka-configs / kafka-reassign-partitions). Across clusters, MM2 runs as Connect clients, so you throttle it with producer/consumer quotas and tasks.max. Throttle too low and lag/RPO grows; too high and you saturate the WAN.

acks=all waits for all in-sync replicas to confirm a write. If some replicas live in another region, every write waits for a cross-region round trip, adding latency. You can use acks=1 or fewer in-sync replicas for speed, but you risk losing data if a region fails.

A stretch cluster is one Kafka cluster whose brokers span regions, giving one global view but paying cross-region latency on every write. A replicated topology is separate clusters per region linked by a copy tool (MirrorMaker 2). Stretch favors consistency; replicated favors latency and isolation.

Place replicas in at least three failure domains (often two regions plus a tiebreaker), require enough in-sync replicas in different regions, and use acks=all. If you size it wrong, losing a region can either stop all writes (no quorum/ISR) or, worse, lose acked data.

Some data legally must stay inside a specific country/region. That forbids stretching partitions or replicating those topics to other regions. You pin data to its home region, keep replicas and consumers in-region, and only move non-restricted or anonymized data across borders.

Cloud providers charge for data leaving a region or AZ. Kafka replication and cross-region consumers multiply traffic, so egress can dominate the bill. Reduce it with follower fetching (read locally), compression, fewer cross-region replicas, and rack-aware placement to avoid needless cross-AZ hops.

MirrorMaker 2 is just a Kafka client. Securing the link means: (1) encrypt traffic in transit with TLS, and (2) authenticate MM2 to each cluster (mTLS or SASL) so only an authorized identity can read the source and write the target.

On the source: Read on the mirrored topics and the consumer group, plus Describe. On the target: Write and Create on the mirrored/internal topics, plus DescribeConfigs/AlterConfigs for config sync. MM2 also needs access to its Connect internal topics on whichever cluster hosts them.

Use the cluster-alias prefixes in MM2's config: set source.security.protocol=SASL_SSL with source.sasl.* for SCRAM, and target.security.protocol=SSL with target.ssl.keystore.* for the client cert. Each cluster's connection is configured independently under its alias.

With mTLS the broker takes the client certificate's full Distinguished Name (DN) as the principal by default, e.g. 'User:CN=mm2,OU=...,O=...'. ssl.principal.mapping.rules lets you rewrite that DN with regex into a shorter principal like 'User:mm2' that your ACLs reference.

Kafka clients first hit a bootstrap broker, which returns the addresses in advertised.listeners; the client then reconnects directly to each broker. Over PrivateLink/peering those advertised addresses must be resolvable and routable from the remote VPC, or the bootstrap succeeds but all follow-up connections fail.