Page History

...

1. User sends AddTopicsToMirror request with topics and mirror name.

2. The broker forwards to the active controller.

3. The controller validates that each topic exists and is not already in a mirror. It then sets the topic config mirror.name=<mirrorName> for each topic, generating a ConfigRecord per topic into the metadata log.
4. Response is sent back to clients with per topic results.
5. When the MirrorMetadataManager in the partition leader node gets notified about the topic config update, it detects that mirror.name is not empty and has no .removed or .paused suffix. It then queries the current mirror partition state from the coordinator. The coordinator could be located on a different broker node, so a ReadMirrorStates inter broker RPC may be needed.
6. Based on the current mirror partition state, the state machine transitions the partition. In most cases, from UNKNOWN to PREPARING.
7. During PREPARING, the mirror fetcher performs Last Mirrored Epoch (LME) truncation. The LME is the greatest leader epoch that the source cluster recognizes from the destination. If the source has no LME knowledge (first time mirroring), it returns -1 and the destination truncates everything and replicates from scratch. Otherwise, the destination truncates at the start offset of the first epoch beyond the LME. It then waits until all ISR members (or all replicas if mirror.support.unclean.leader.election=true) complete the truncation.
8. Once all ISR members have completed truncation, the state transitions from PREPARING to MIRRORING. A MirrorFetcherThread is created and starts sending consumer Fetch requests (not follower requests) to the source cluster to replicate data. The Fetch protocol handles any offset level divergence by truncating to the exact offset where the source epoch ends.

9. The fetched batch retains its original leader epoch from the source. When the partition later becomes writable after failover, the leader epoch is bumped to ensure monotonically increasing epochs for new records.

10. The partition state is persisted to the __mirror_state topic on each state change via local append or WriteMirrorStates (when coordinator is remote) as MirrorPartitionStateKey/MirrorPartitionStateValue records, distributed by hash(mirrorName, topicId, partition) % numPartitions.
11. The MirrorMetadataManager also periodically synchronizes topic configs, consumer group offsets, and ACLs from the source cluster.

Remove Topics from Mirror

...

1. User sends RemoveTopicsFromMirror request with topics and mirror name.
2. The controller validates each topic belongs to the specified mirror and is in MIRRORING state. It then updates the topic config by appending the .removed suffix, e.g. mirror.name=cluster1.removed, generating a ConfigRecord.
3. When the MirrorMetadataManager in the partition leader node gets notified, it detects the .removed suffix on mirror.name. It queries the current mirror partition state from the coordinator, and transitions to STOPPING.
4. During STOPPING:
   1. The MirrorFetcherManager removes all fetcher threads for the affected partitions, stopping replication.
   2. Bump the leader epoch for the partitions to ensure monotonically increasing epochs for new records.
   3. The log is truncated to LSO (Last Stable Offset) for transactional consistency.
   4. The LME is recorded as LastMirroredEpochsKey/LastMirroredEpochsValue records into the __mirror_state topic for potential future failback.
   5. A MIRROR_PID_RESET control batch is written to the partition log, which expires all ProducerStateManager entries so that new producers get fresh PIDs with no collision risk.
5. The state transitions from STOPPING to STOPPED. The read only flag is cleared and the topic becomes writable. New producers can start producing with fresh PIDs starting at sequence 0 and a higher leader epoch.

...