Page History

...

• UNKNOWN: The partition has no cached state (broker just became leader, state not loaded yet). Not an explicit API-driven state, just the absence of state.
• PREPARING: The coordinator for this partition detects via onMetadataUpdate that it leads a mirror partition. It fetches last mirrored mirror epochs from the source cluster and truncates logs to align the local log with the source.
• MIRRORING: All ISR set members have completed truncation. A mirror fetcher thread is started to continuously replicate records from the source cluster.
• PAUSING: Triggered by the pause operation. The system removes fetchers for the affected partitions. Valid from: MIRRORING only.
• PAUSED: Fetchers have been removed. The partition stays read-only with no active fetchers and no metadata sync (configs, consumer groups, ACLs). On resume, transitions directly to MIRRORING.
• STOPPING: Performs the following actions (see MirrorFetcherThread paragraph for more details).
  1. Remove mirror fetcher threads: Stops cross-cluster replication for the affected partitions.
  2. In parallel:
     1. Bump leader epoch (async): Sends a BumpLeaderEpochs request to the controller with the latest local log epoch as minLeaderEpoch, ensuring the destination leader epoch exceeds the source cluster's last known epoch.

     2. Truncate to LSO, then update LME: Truncates each partition's log to LSO, discarding any uncommitted tail, then persists the latest leader epoch from each partition's log into the __mirror_state coordinator topic, recording LME for future failback.

  3. Write PID reset barrier: Once both parallel branches complete, appends a MIRROR_PID_RESET control record to each partition, fencing stale producer IDs from the source cluster.
• STOPPED: The topic becomes writable on the destination cluster. The mirror fetcher is removed and the read-only flag is cleared.
• FAILED: An error occurred. Valid from: any state. The operator that wants to restart a failed mirror partition can remove the topic from the mirror and add it back again. The STOPPING handler already removes fetchers and truncates to the LSO, both of which are safe operations on a failed partition (fetchers are likely already gone, and truncation is a best-effort cleanup). More sophisticated recovery strategies can be added later with a follow-up KIP.

Example scenarios:

• Starting a mirror (UNKNOWN -> PREPARING -> MIRRORING): The StartMirrorTopics command sets mirror.name config via the controller. The metadata update propagates to brokers. The broker leading the partition finds out the partition state via the coordinator, and this might trigger readMirrorState RPC to query from the remote coordinator responsible for handling that mirror partition, and transitions to PREPARING if it's in a valid transition state. After truncation completes, it moves to MIRRORING and starts the mirror fetcher to fetch data from the source cluster.
• Failing over to destination (MIRRORING -> STOPPING -> STOPPED): The StopMirrorTopics command appends the ".removed” suffix to the mirror.name config. The partition leader  detects the stop request, transitions to STOPPING, truncates to LSO, persists the last mirrored epoch, then moves to STOPPED. The topic is now writable after the STOPPED state.
• Restarting a stopped mirror (STOPPED -> PREPARING -> MIRRORING): The mirror.name config is set again. The controller is notified, sees the partition in STOPPED state and transitions to PREPARING, re-truncating and resuming replication.

MirrorMetadataManager

The MirrorMetadataManager (MMM) implements periodic metadata synchronization between source and destination clusters. It maintains persistent network connections to all source clusters. During periodic metadata refresh, the broker validates that the source cluster ID has not changed. If a mismatch is detected, metadata sync for that mirror is halted and an error is logged. This prevents silent data corruption in case of misconfiguration or unintended source cluster replacement.

...

There are two main concepts to keep in mind when dealing with log convergence across Kafka clusters:

• Last Mirrored Mirror Epoch (LME): The greatest leader epoch of a given partition that a destination cluster recognizes from the source cluster and stores in the __mirror_state internal topic. It represents the synchronization point between source and destination. After the log is truncated to LME, the destination cluster does not contain any record with a leader epoch beyond the LME. Only the source cluster owns leader epochs exceeding the LME. This ensures the source leader epoch remains the source of truth, even when epoch histories diverge across clusters during asynchronous mirroring.
• Leader Epoch Bump (LEB): The leader epoch in the destination cluster remains unchanged during replication. That means it is possible that the fetched batch carries a leader epoch of X that does not match with the local leader epoch. To ensure the leader epoch remains monotonically increasing, it is incremented when a partition becomes writable after failover. New records produced on the cluster will then carry an epoch higher than anything in the existing log.

...

1. User sends StartMirrorTopics 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 Mirror 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. The fetched batch retains its original leader epoch from the source.
9. 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.
10. The MirrorMetadataManager also periodically synchronizes topic configs, consumer group offsets, and ACLs from the source cluster.

...

{
  "apiKey": 1,
  "type": "coordinator-key",
  "name": "LastMirrorEpochsKey",
  "validVersions": "0",
  "flexibleVersions": "none",
  "fields": [
    { "name": "MirrorName", "type": "string", "versions": "0",
      "about": "The cluster mirror name."}
  ]
}

{
  "apiKey": 1,
  "type": "coordinator-value",
  "name": "LastMirrorEpochsValue",
  "validVersions": "0",
  "flexibleVersions": "0+",
  "fields": [
    { "name": "Topics", "type": "[]Topic", "versions": "0+",
      "about": "The mirror topics for which we want to store the last mirroredmirror epochs.",  "fields": [
      { "name": "Name", "type": "string", "versions": "0",
        "about": "The topic name." },
      { "name": "Partitions", "type": "[]Partition", "versions": "0+",
        "about": "Each partition to record the last mirroredmirror epochs.", "fields": [
        { "name": "PartitionIndex", "type": "int32", "versions": "0+",
          "about": "The partition index." },
        { "name": "", "type": "int32", "versions": "0+",
          "about": "The last mirror leader epoch for this partition." }
      ]}
    ]}
  ]
}

...