Page History

...

1. The user sends CreateMirror requests to any broker with the mirror name and mirror related properties (bootstrap servers, security settings, etc.).
2. The broker forwards the request to the active controller.
3. The controller saves the properties into the metadata log as ConfigRecord entries with type MIRROR.
4. If this is the first mirror being created, the controller also auto creates the __mirror_state internal topic.
5. All brokers receive the metadata update and the MirrorMetadataManager registers the new mirror configuration.

...

Start Mirror Topics

...

1. User sends AddTopicsToMirror 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 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.

...

Stop Mirror Topics

...

1. User sends RemoveTopicsFromMirror StopMirrorTopics 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=my-mirror.removed, generating a ConfigRecord.
3. When the MirrorMetadataManager gets notified, it detects the .removed suffix on mirror.name. It queries the current mirror partition state from the coordinator, and transitions the mirror partition 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 for transactional consistency.
   4. The LME is recorded as sKey/LastMirrorEpochsValue records into the __mirror_state topic for potential future failback.
   5. A MIRROR_PID_RESET control record 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.

Pause Mirror Topics

1. User sends PauseMirrorTopics request with topics and mirror name.
2. The controller validates each topic belongs to the specified mirror and is currently in MIRRORING state. It appends the .paused suffix to the mirror name config, e.g. mirror.name=my-mirror.paused, generating a ConfigRecord.
3. When the MirrorMetadataManager in the partition leader node gets notified, it detects the .paused suffix. It transitions the state to PAUSING.
4. During PAUSING, the MirrorFetcherManager removes the fetcher threads for the affected partitions. No more data is replicated.
5. The state transitions from PAUSING to PAUSED. The partition remains read only. Metadata synchronization (configs, groups, ACLs) is also halted for the paused topics.
6. The partition state change is persisted to the __mirror_state topic.

Resume Mirror Topics

1. User sends ResumeMirrorTopics request with topics and mirror name.
2. The controller validates the topic is currently paused (has .paused suffix). It removes the .paused suffix, restoring the original mirror name, e.g. mirror.name=cluster1, generating a ConfigRecord.
3. When the MirrorMetadataManager in the partition leader node gets notified, it detects that mirror.name no longer has the .paused suffix.
4. The state transitions directly from PAUSED to MIRRORING. No log truncation is needed because the partition is already at the correct offset from before the pause.
5. New MirrorFetcherThread instances are created and resume replication from the current log end offset.
6. Metadata synchronization (configs, groups, ACLs) also resumes.

...

{
  "apiKey": TBD,
  "type": "response",
  "name": "CreateMirrorResponse",
  // Version 0 is the initial version.
  "validVersions": "0",
  "flexibleVersions": "0+",
  "fields": [
    { "name": "ThrottleTimeMs", "type": "int32", "versions": "0+",
      "about": "The duration in milliseconds for which the request was throttled due to a quota violation, or zero if the request did not violate any quota." },
    { "name": "ErrorCode", "type": "int16", "versions": "0+",
      "about": "The error code, or 0 if there was no error." },
    { "name": "ErrorMessage", "type": "string", "versions": "0+", "nullableVersions": "0+",
      "about": "The error message, or null if there was no error." }
  ]
}

StartMirrorTopics

Start mirroring for the specified topics. The broker validates that all target topic partitions are in either UNKNOWN or STOPPED state; otherwise, the request is rejected. Once validated, the request is forwarded to the controller, which sets the mirror.name topic config to the specified mirror name.

...

{
  "apiKey": TBD,
  "type": "response",
  "name": "StartMirrorTopicsResponse",
  // Version 0 is the initial version.
  "validVersions": "0",
  "flexibleVersions": "0+",
  "fields": [
    { "name": "ThrottleTimeMs", "type": "int32", "versions": "0+",
      "about": "The duration in milliseconds for which the request was throttled due to a quota violation, or zero if the request did not violate any quota." },
    { "name": "ErrorCode", "type": "int16", "versions": "0+",
      "about": "The error code, or 0 if there was no error." },
    { "name": "ErrorMessage", "type": "string", "versions": "0+", "nullableVersions": "0+", "default": "null",
      "about": "The top-level error message, or null if there was no error." },
    { "name": "MirrorName", "type": "string", "versions": "0+", "entityType": "mirrorName",
      "about": "The cluster mirror name." },
    { "name": "Topics", "type": "[]TopicResult", "versions": "0",
      "about": "The results for the topics.", "fields": [
      { "name": "Name", "type": "string", "versions": "0", "entityType": "topicName",
        "about": "The topic name." },
      { "name": "ErrorCode", "type": "int16", "versions": "0",
        "about": "The error code, or 0 if there was no error." }
    ]}
  ]
}

StopMirrorTopics

Stop mirroring for the specified topics. The broker validates that all target topic partitions are in either PREPARING or MIRRORING state. Once validated, the request is forwarded to the controller, which appends the ".removed" suffix to the mirror.name topic config to mark the topics as no longer mirrored.

...

{
  "apiKey": TBD,
  "type": "response",
  "name": "StopMirrorTopicsResponse",
  // Version 0 is the initial version.
  "validVersions": "0",
  "flexibleVersions": "0+",
  "fields": [
    { "name": "ThrottleTimeMs", "type": "int32", "versions": "0+",
      "about": "The duration in milliseconds for which the request was throttled due to a quota violation, or zero if the request did not violate any quota." },
    { "name": "ErrorCode", "type": "int16", "versions": "0+",
      "about": "The error code, or 0 if there was no error." },
    { "name": "ErrorMessage", "type": "string", "versions": "0+", "nullableVersions": "0+", "default": "null",
      "about": "The top-level error message, or null if there was no error." },
    { "name": "MirrorName", "type": "string", "versions": "0+", "entityType": "mirrorName",
      "about": "The cluster mirror name." },
    { "name": "Topics", "type": "[]TopicResult", "versions": "0",
      "about": "The results for the topics.", "fields": [
      { "name": "Name", "type": "string", "versions": "0", "entityType": "topicName",
        "about": "The topic name." },
      { "name": "ErrorCode", "type": "int16", "versions": "0",
        "about": "The error code, or 0 if there was no error." }
    ]}
  ]
}

PauseMirrorTopics

Pauses data replication and metadata sync for the specified mirror topics, keeping them read-only on the destination cluster.

...

{
  "apiKey": TBD,
  "type": "response",
  "name": "PauseMirrorTopicsResponse",
  // Version 0 is the initial version.
  "validVersions": "0",
  "flexibleVersions": "0+",
  "fields": [
    { "name": "ThrottleTimeMs", "type": "int32", "versions": "0+",
      "about": "The duration in milliseconds for which the request was throttled due to a quota violation, or zero if the request did not violate any quota." },
    { "name": "ErrorCode", "type": "int16", "versions": "0+",
      "about": "The error code, or 0 if there was no error." },
    { "name": "ErrorMessage", "type": "string", "versions": "0+", "nullableVersions": "0+", "default": "null",
      "about": "The top-level error message, or null if there was no error." },
    { "name": "MirrorName", "type": "string", "versions": "0+", "entityType": "mirrorName",
      "about": "The cluster mirror name." },
    { "name": "Topics", "type": "[]TopicResult", "versions": "0",
      "about": "The results for the topics.", "fields": [
      { "name": "Name", "type": "string", "versions": "0", "entityType": "topicName",
        "about": "The topic name." },
      { "name": "ErrorCode", "type": "int16", "versions": "0",
        "about": "The error code, or 0 if there was no error." }
    ]}
  ]
}

ResumeMirrorTopics

Resumes data replication and metadata sync for previously paused mirror topics from where they left off.

...

{
  "apiKey": TBD,
  "type": "response",
  "name": "BumpLeaderEpochsResponse",
  // Version 0 is the initial version.
  "validVersions": "0",
  "flexibleVersions": "0+",
  "fields": [
    { "name": "ThrottleTimeMs", "type": "int32", "versions": "0+",
      "about": "The duration in milliseconds for which the request was throttled due to a quota violation, or zero if the request did not violate any quota." },
    { "name": "ErrorCode", "type": "int16", "versions": "0+",
      "about": "The error code, or 0 if there was no error." },
    { "name": "Topics", "type": "[]TopicPartitions", "versions": "0+",
      "about": "Each topic in the mirror.", "fields": [
      { "name": "TopicName", "type": "string", "versions": "0+",
        "about": "The topic name." },
      { "name": "Partitions", "type": "[]PartitionDetail", "versions": "0+",
        "about": "Each partition state.", "fields": [
        { "name": "PartitionIndex", "type": "int32", "versions": "0+",
          "about": "The partition index." },
        { "name": "ErrorCode", "type": "int16", "versions": "0",
          "about": "The error code, or 0 if there was no error." }
      ]}
    ]}
  ]
}

Cluster Metadata Records

LastMirrorEpochs

PartitionChangeRecord

New version adds MinLeaderEpoch to support leader epoch bumps triggered by the mirror state machine during failoverThe greatest leader epoch of a given partition that a destination cluster recognizes from the source cluster.

{
  "apiKey": 5,                                                                                                                                                                                                                                                                           
  "type": "metadata",                                                                                                                                                                                                                                                                    
  "name": "PartitionChangeRecord",
  "validVersions": "0-3",                                                                                                                                                                                                                                                                
  "flexibleVersions": "0+",
  "fields": [
    // existing fields unchanged
    {"name": "MinLeaderEpoch", "type": "int32", "versions": "3+", "default": -1,
      "about": "The minimum leader epoch requested."}
  ]
}

Mirror Metadata Records

LastMirrorEpochs

The greatest leader epoch of a given partition that a destination cluster recognizes from the source cluster.

{
  "apiKey": 1,
  "type": "coordinator-key",
  "name": "LastMirrorEpochsKey1,
  "type": "coordinator-key",
  "name": "sKey",
  "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 mirrored 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 mirrored 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." }
      ]}
    ]}
  ]
}

...