On failover to the standby cluster, why are source offsets not valid, and how does MirrorMaker 2 offset translation let consumers resume correctly?

## Why offsets don't carry across clusters In Kafka, an **offset** is a monotonically increasing integer that identifies a record's position **within one partition of one cluster**. Each cluster assigns offsets independently as records are appended to its own log. When MirrorMaker 2 copies records from the primary to the standby, the standby is just another producer target: it appends the mirrored records to its own log and assigns its **own** offsets. These will not match the source offsets, because: - The standby's mirror topic may have started replicating at a point where the source already had millions of records — the standby's offset 0 corresponds to source offset 4,000,000, say. - Producing to the target can fail and retry, or batching/timing differs, so there's no fixed arithmetic constant linking the two. So a consumer that committed **source offset 1000** cannot simply `seek(1000)` on the target — that's a different, possibly meaningless, record. ## How MM2 captures the mapping MM2 runs three connectors: - **MirrorSourceConnector** — copies the actual records. Crucially, as it produces each batch to the target, it learns the correspondence between source offset and the resulting target offset and periodically writes **offset-sync** records to an internal topic named like `mm2-offset-syncs.<target>.internal`. Each sync says "source partition P offset X landed at target offset Y." - **MirrorCheckpointConnector** — periodically reads the source cluster's committed consumer-group offsets (from the source `__consumer_offsets`), looks up the nearest offset-sync, **translates** the source committed offset into the equivalent target offset, and emits **Checkpoint** records to a `<source>.checkpoints.internal` topic on the target. With `sync.group.offsets.enabled=true` it can also write the translated offsets straight into the target's `__consumer_offsets`. - **MirrorHeartbeatConnector** — emits heartbeats to measure connectivity/lag. ## Consuming the translation at failover Three common paths: 1. **Automated**: with `sync.group.offsets.enabled=true`, the target's `__consumer_offsets` already holds translated commits, so a consumer group simply starts on the target and resumes — provided the group isn't actively committing there (only inactive/failed-over groups should be synced, controlled by `emit.checkpoints` and group filters). 2. **Programmatic**: `RemoteClusterUtils.translateOffsets(...)` or the `MirrorClient.remoteConsumerOffsets(...)` API reads checkpoints and returns the target offsets to `seek()` to before consuming. 3. **Manual tooling**: read the checkpoints topic directly. ## Why translation is approximate Offset syncs are emitted **periodically**, not per-record (controlled by `offset.lag.max` — emit a new sync when the offset has advanced more than this). Translation rounds the committed source offset down to the nearest sync point. Result: a failed-over consumer may resume **slightly before** its true last position and **reprocess a handful of records** — which is fine for **at-least-once** processing but means you must not assume exactly-once across a DR cutover. ## Edge cases and gotchas - **IdentityReplicationPolicy**: keeps topic names identical across clusters (no `primary.` prefix). Checkpoint/translation still works, and it's preferred for DR so apps don't have to rename topics on failover. - **Lag in checkpoints**: if the primary dies, the last few group commits may not have been checkpointed, so the translated position may be staler than the consumer's actual last commit — contributing to reprocessing. - **Newer alternatives**: Confluent **Cluster Linking** preserves offsets byte-for-byte (mirror topics share offsets), eliminating translation entirely — a key reason teams pick it for DR. - Don't run the same group **actively** on both clusters while syncing offsets, or you'll clobber live commits.