How do you actually configure bidirectional MM2 between two clusters (connectors, flows, prefixes), and how do consumers fail over between regions?

## Topology of a bidirectional flow MM2 is built on Kafka Connect. You describe **clusters** and **replication flows**. A minimal `mm2.properties`: ``` clusters = A, B A.bootstrap.servers = a1:9092,a2:9092 B.bootstrap.servers = b1:9092,b2:9092 A->B.enabled = true A->B.topics = orders|payments B->A.enabled = true B->A.topics = orders|payments replication.factor = 3 sync.topic.configs.enabled = true sync.group.offsets.enabled = true emit.checkpoints.enabled = true emit.heartbeats.enabled = true ``` Both `A->B.enabled` and `B->A.enabled` true is what makes it **bidirectional / active-active**. Distinct aliases `A` and `B` are mandatory so the prefix disambiguates origin. ## The three connectors per flow Each enabled `X->Y` flow instantiates three Kafka Connect connectors: 1. **MirrorSourceConnector** — the workhorse. Consumes source topics, **re-produces** them to the target under `DefaultReplicationPolicy` (`orders` → `X.orders`), and replicates topic configurations and ACLs. This is where prefixing (and thus loop prevention) happens. 2. **MirrorCheckpointConnector** — periodically reads source consumer-group committed offsets and writes **checkpoints** to a `<source>.checkpoints.internal` topic on the target, mapping each source offset to the corresponding target offset of the mirrored topic. This is the basis for consumer failover. 3. **MirrorHeartbeatConnector** — produces records to a `heartbeats` topic at a fixed interval. Consumers/operators measure the heartbeat's end-to-end delay to monitor replication liveness and lag (`replication-latency-ms`). ## Offset translation and consumer failover The subtle problem: offset `5000` in A's `orders` is **not** the same physical offset as in B's `A.orders` — replication starts at a different base and records may be filtered, so positions diverge. To fail a consumer group over from A to B you need the **translated** offset. MM2 solves this two ways: - **Checkpoints** via MirrorCheckpointConnector, queried programmatically with `RemoteClusterUtils.translateOffsets(props, "groupId", "A", timeout)` which returns the map of target offsets to seek to. - **`sync.group.offsets.enabled=true`** automatically writes the translated offsets into the target's `__consumer_offsets` for the group, so a restarted consumer on B simply finds committed offsets and resumes. With correct translation, a failed-over consumer resumes near where it stopped — minimal reprocessing (some duplicates, since it's at-least-once) and no large gaps. ## What gets synced - **Records** (MirrorSourceConnector). - **Topic configs** (`sync.topic.configs.enabled`) — retention, partitions tracked. - **ACLs** (`sync.topic.acls.enabled`). - **Consumer-group offsets** (`sync.group.offsets.enabled` + checkpoints). ## Running MM2 You can run the dedicated `connect-mirror-maker.sh mm2.properties` driver (standalone MM2 cluster), or deploy the connectors onto an existing Kafka Connect cluster. For HA, run multiple MM2 nodes; Connect distributes connector tasks. ## Edge cases - **Same alias on both clusters** breaks prefixing and loop prevention — keep aliases unique. - **Offset translation lag**: checkpoints are periodic (`emit.checkpoints.interval.seconds`); a failover right after a burst may translate to a slightly stale offset → some reprocessing. - **New topics**: MM2 picks up topics matching the `topics` regex on its refresh interval; there's a small delay before a new topic is mirrored. - **Heartbeats topic** must exist/be allowed on both sides for monitoring to work.