Page History

...

# when the source cluster is back, the operator can failback by creating a mirror with the same name
echo "bootstrap.servers=localhost:9094" > /tmp/my-mirror.properties
bin/kafka-mirrors.sh --bootstrap-server :9091 --create --mirror my-mirror --mirror-config /tmp/my-mirror.properties
bin/kafka-mirrors.sh --bootstrap-server :"9091 --add --topic .* --mirror my-mirror
# 9091 (destination) <----- 9094 (source)

Create Mirror

...

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 ConfigResource(Type.MIRROR, mirrorName).
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.

Add Topics to Mirror

...

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 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.

Pause 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=cluster1.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 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.

Delete Mirror

...

1. The user sends a DeleteMirror request with the mirror name.
2. The controller validates that the mirror is empty (no topics assigned) or all its partitions are in STOPPED state.
3. If valid, the controller tombstones the mirror configuration in the cluster metadata log, removing all ConfigRecord entries for the mirror.
4. All mirror state entries in __mirror_state for this mirror are cleaned up.
5. Any remaining coordinator state is shut down, source cluster connections are closed, and the mirror name becomes available for reuse.
6. After deletion, failback using this mirror configuration is no longer possible.

...

1. The user sends ListMirrorsRequest to any broker (no parameters required).

2. The broker handler gets all configured mirror partitions from, which reads from the in memory metadata cache.
3. For each authorized mirror, the broker returns: mirror name, source cluster ID, source bootstrap servers, and topic count.
4. No metadata records are written. This is a read only operation against the local metadata cache.

Describe Mirrors

...

1. The user sends DescribeMirrorsRequest with optional mirror names (empty means all mirrors).

2. The broker handler queries two sources:

   1. The ReplicaManager which provides source offset, destination offset, and lag for each partition.

   2. The MirrorCoordinator which provides the current partition state from the metadata manager cache.

3. The request is forwarded to each broker that only reports partitions for which it has lag information or is the partition leader. This avoids duplicate reporting across brokers.

4. For each partition, the response includes: mirror name, topic name, partition ID, source offset, destination offset, lag, and current state.
5. No metadata records are written. This is a read only operation.

...

Code Block
// new added field in PartitionData type { "name": "MirrorLeaderEpoch", "type": "int32", "versions": "19+", "default": "-1", "taggedVersions": "19+", "tag": 3, "ignorable": true, "about": "The latest known mirrored leader epoch." },

CreateMirror

Allows users to create a mirror and supply its configuration. When the broker receives the request, it validates that the mirror name is not already in use, contains only permitted characters, and does not end with ".removed" or ".paused" suffix. Once validated, the request is forwarded to the controller, which persists the configuration in the metadata log.

...

{
  "apiKey": 103,
  "type": "response",
  "name": "ResumeMirrorTopicsResponse",
  // 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." }
    ]}
  ]
}

DeleteMirror

Permanently deletes a cluster mirror, including its configuration. The mirror must be empty (no topics) or all its partitions must be in STOPPED state. After deletion, all metadata are tombstoned, making failback impossible. This is an irreversible operation.

...

{
  "apiKey": TBD,
  "type": "response",
  "name": "ListMirrorsResponse",
  // 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": "Mirrors", "type": "[]ListedMirror", "versions": "0+",
      "about": "Each mirror in the response.", "fields": [
      { "name": "MirrorName", "type": "string", "versions": "0+", "entityType": "mirrorName",
        "about": "The cluster mirror name." },
      { "name": "SourceBootstrap", "type": "string", "versions": "0+",
        "about": "The source cluster bootstrap servers." },
      { "name": "SourceClusterId", "type": "string", "versions": "0+", "default": "",
        "about": "The source cluster ID, or empty if not yet resolved." },
      { "name": "TopicCount", "type": "int32", "versions": "0+", "default": "0",
        "about": "The number of topics configured for this mirror. 0 indicates an empty mirror with no topics." }
    ]}
  ]
}

DescribeMirrors

Returns the current mirroring status, state, and configuration for the specified mirror topics on the destination cluster.

...