...

The mirror name is stored as a topic-level configuration (mirror.name) that propagates through Kafka's metadata log as configuration change records. When topics are added to a mirror via the addTopicsToMirror API, the controller generates configuration records that are replicated to all brokers through the standard metadata update mechanism.

...

Starting a mirror (UNKNOWN -> PREPARING -> MIRRORING): The addTopicsToMirror command sets mirror.name config via the controller. The metadata update propagates to brokers. The broker leading the partition finds out the partition sees it's state via the coordinator, finds no cached state (UNKNOWN), and transitions to PREPARINGand this might trigger readMirrorState RPC to query from the remote coordinator and transitions to PREPARING if it’s in a valid transition state (e.g. UNKNOWN). After truncation completes, it moves to MIRRORINGand starts the mirror fetcher to fetch data from the source cluster.

Failover (MIRRORING -> STOPPING -> STOPPED): The removeTopicsFromMirror command clears appends the ".removed” suffix in mirror.name config. The coordinator detects partition leader  detects the stop request, transitions to STOPPING, persists the last offset, 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. onMetadataUpdate sees the partition in STOPPED state and transitions to PREPARING, re-truncating and resuming replication.

...

1. Connection Management: The manager maintains a connection pool with one blocking sender per source cluster. These connections are created lazily when the first topic for a mirror is added. Each sender uses the security credentials and network settings from the mirror configuration, allowing different mirrors to use different authentication mechanisms.
2. Topic Metadata Synchronization: Every refresh cycle, the manager fetches topic metadata from source clusters using standard MetadataRequest calls. For each topic in the mirror configuration:
1. Topic Creation: If a topic exists in the source but not the destination, the manager sends a CreateTopics request to the controller with identical partition count and configurations.
2. Partition Expansion: If the source topic has more partitions than the destination, the manager sends a CreatePartitions request to scale up the destination topic to match.
3. Configuration Sync: Topic configurations are compared between source and destination. Any differences trigger an IncrementalAlterConfigs request to align destination configs with the source.
4. Topic Deletion: When a topic is deleted on the source cluster, the mirror partitions on the destination cluster moves to STOPPED state. This prevents accidental deletions to affect the destination cluster. In case it was intentional, the operator would need to manually remove the topic from the mirror.
3. Consumer Group Offset Synchronization: The manager synchronizes classic and share consumer group offsets to enable seamless failover (no offset translation):
1. Lists all consumer groups using ListGroups request.
2. Fetches committed offsets for each group using OffsetFetch request or DescribeShareGroupOffsets request.
3. Commits those offsets to the destination cluster’s group coordinator using the internal OffsetCommit or AlterShareGroupOffsets request.
4. ACL Synchronization: Access control lists are mirrored from source to destination to maintain consistent security policies:
1. Fetches all ACLs from the source using DescribeAcls request.
2. Compares with the destination cluster’s current ACLs from the metadata image.
3. Creates missing ACLs using CreateAcls request.
4. Deletes ACLs that exist in destination but not in source using DeleteAcls request.

...

• The mirror.topic.properties.exclude config controls which topic configuration properties are excluded from synchronization using regex patterns.
• follower.replication.throttled.replicas, leader.replication.throttled.replicas, message.timestamp.difference.max.ms, log.message.timestamp.before.max.ms, log.message.timestamp.after.max.ms, message.timestamp.type, unclean.leader.election.enable, min.insync.replicas, mirror.name (default)
• .*throttled.* (exclude all throttle-related properties)
• min.insync.replicas,unclean.leader.election.enable (exclude only these two properties)

...

The MirrorFetcherManager (MFM) extends AbstractFetcherManager to handle fetcher thread lifecycle for mirror partitions. It uses a three-dimensional key (fetcher ID, source broker endpoint, mirror name) to organize threads, ensuring that:

• Partitions from different mirrors use separate threads for authentication isolation.
• Partitions from the same mirror are distributed across multiple threads for load balancing.
• Leader changes in the source partition trigger thread reassignment or recreation to the new source broker.

The MirrorFetcherThread (MFT) is a specialized implementation of AbstractFetcherThread that handles cross-cluster data replication with consumer Fetch requests and different epoch semantics than standard intra-cluster replication, but keeping the same log consistency validations. In Cluster Mirroring, destination partition leaders operate in a The destination cluster's replica is not registered as a follower in the source cluster. Using a follower Fetch request would cause the source broker to attempt updating follower replica status for a replica it doesn't know about. A consumer Fetch request avoids this issue, as it carries no such side effects on the source broker's replica state. In other words, destination partition leaders operate in a dual-role. They act as followers when fetching committed data (from the source cluster leader up to the last stable offset (LSO) from the source cluster leader, while simultaneously serving as leaders for their local replicas in the destination cluster. To maintain data consistency, destination partitions are read-only and reject produce requests from clients with ReadOnlyTopicException.

A mirror topic is created with the same topic ID as in the source cluster. This serves two purposes: it satisfies fetch request validation on the source broker, and it enables identity verification during failback where the destination cluster can confirm it is working with the exact same topic by comparing topic IDs.

A mirror leader partition begins with an unknown source leader epoch. When it sends Fetch requests to the source cluster, the source leader may respond with a FencedLeaderEpochException. When such an error occurs, the mirror fetcher extracts the current source leader epoch from the error response and updates its internal fetch state to track the source cluster's actual leader epoch.  The last fetched epoch is always set to empty to disable log divergence checks due to unclean leader election (see non-goals section).

...

Failover is initiated by calling the RemoveTopicsFromMirror API, which appends a .removed suffix into the mirror.name internal config. This transitions the mirror topics from read-only to writable state after the stopping process completes gracefully.

...

Before transitioning a mirror partition from PREPARING to MIRRORING, the MirrorCoordinator must ensure that all in-sync replicas in the destination cluster have truncated their logs to the correct offset. If less than min ISR are available, we will skip and retry in the following fetch. This coordination step validates that every ISR member has completed truncation before the partition is allowed to begin actively fetching from the source cluster. Without it, the mirror leader could start appending new data from the source while local followers still hold divergent log segments, causing inconsistencies within the destination cluster. After truncation, reverse mirroring begins normally. Note that the log truncation on the reverse mirroring may cause the data loss for the records that didn’t get mirrored to the old destination cluster earlier.

...

Cluster Mirroring ensures transactional consistency when stopping by truncating to the Last Stable Offset (LSO). Note the LSO. Note that this doesn’t mean it supports exactly-once semantics (EOS) across clusters, which would require synchronous communication.

During the mirror stopping transition, the MirrorCoordinator performs a log truncation operation that resets each mirror partition to its LSO. This offset represents the point in the log where all transactions have been decided (committed or aborted), essentially the highest offset where data is known to be consistent from a transactional perspective. Any records beyond this point may belong to incomplete transactions and should not persist after mirroring stops. Note that the actual lag may be greater than what’s reported by the metrics.

This approach prevents a critical consistency issue: the destination cluster could retain partial transaction data that would never be completed since mirroring has stopped. This would leave the topic in an inconsistent state where read_committed consumers may be blocked due to incomplete transaction data. Additionally, the transaction coordinator would not be able to rollback these hanging transactions because there is no __transaction_state metadata in the destination cluster.

...

1. Destination Cluster Throttling: To avoid conflicts with intra-cluster replication controls, mirror-specific throttling configurations operate independently from standard replication throttling. The system provides two configuration levels: a broker-level rate limit that sets the overall bandwidth ceiling for mirror replication traffic, and a topic-level replica list that specifies which partition-broker combinations should be throttled using the standard partition-index and broker-id notation. Operators can dynamically adjust throttling rates at runtime without restarting brokers, first setting a cluster-wide default rate, then fine-tuning specific topic partitions as mirroring progresses. This allows gradual bandwidth allocation as mirror relationships are established.

...

2. Source Cluster Throttling: The source cluster side requires a different approach because mirror fetch requests operate as consumer traffic rather than replication traffic. This design is intentional since the mirroring must fetch only up to the LSO to maintain transactional consistency, which is a consumer-level guarantee not available through the replication protocol. Consequently, standard leader replication throttling mechanisms cannot apply to mirror traffic. Instead, the source cluster leverages Kafka's client quota system. Each mirror fetcher thread presents itself with a deterministic client identifier that encodes the broker ID, fetcher thread number, and mirror name. Operators can apply per-client byte rate quotas to these identifiers, effectively throttling the outbound mirror traffic from the source cluster. This approach integrates seamlessly with Kafka's existing quota enforcement infrastructure.

Tiered Storage

Tiered Storage is not initially supported, but a detailed design of the metadata synchronization protocol, API schema, and state management will be provided in a follow-up KIP. A mirror follower that receives an OffsetMovedToTieredStorageException from the source leader handles it by marking the partition as failed, and also the mirror partition state will move to FAILED state.

Share Group

Cluster Mirroring supports both traditional consumer groups and share consumer groups (Kafka Queue functionality) to ensure seamless failover for all consumer types. While the data mirroring mechanism remains identical, the offset synchronization strategy differs based on the group type.

...

At the time of writing, the Diskless Topics KIP (KIP-1500 and other sub-KIPs) are still under discussion, so there will be future KIPs to support this feature.

Active-Active Writes

Active-active topology is not initially supported in Cluster Mirroring, though it could potentially be achieved through topic prefixing and removing the reliance on topic ID for mirroring. This is a candidate for a future improvement KIP. 

...

// new added
{ "name": "MirrorInfo", "type": "MirrorInfo", "versions": "8+", "nullableVersions": "8+", "ignorable": true,
  "about": "Mirror information for creating a mirror topic from a source cluster.", "fields": [
    { "name": "TopicId", "type": "uuid", "versions": "8+",
    "about": "The topic ID from the source cluster." }
]}

The topic ID field ensures mirror topics retain the same topic ID as the source cluster topic. This allows fetch requests to pass validation on the source broker, and enables the system to verify that a topic being mirrored to a same-named topic in the destination cluster is indeed the same logical topic, not a name collision.

In normal topic creation, the MirrorInfo field will be null. When receiving the CreateTopic request, the controller will check the new field. If it is not set, the topic ID will be generated with random UUID as usual. Otherwise, the controller will do the following validation:

1. This topic ID is not used by other topics in the current cluster
2. The replicas for the partition assignment are all active and not in fenced or controlled shutdown. This is to make sure when a topic gets deleted and re-created with the same topic ID, the stale offline log dir won’t be treated as the active log dir after it becomes online (KAFKA-16234).

CreateMirror

CreateMirrorRequest

The CreateMirror API 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 the .removed suffix. Once validated, the request is forwarded to the controller, which persists the configuration in the metadata log.

CreateMirrorRequest

{ 
  "apiKey": TBD, 
  "type": "request", 
  "listeners": 

{ 
  "apiKey": TBD, 
  "type": "request", 
  "listeners": ["broker", "controller"], 
  "name": "CreateMirrorRequest", 
  "validVersions": "0", 
  "flexibleVersions": "0+", 
  "fields": [ 
    { "name": "MirrorName", "type": "string", "versions": "0+", "nullableVersions": "0+", 
      "about": "The cluster mirror name."}, 
    { "name": "Config", "type": "[]MirrorConfig", "versions": "0+", 
      "about": "The cluster mirror configurations.",  "fields": [ 
      { "name": "Name", "type": "string", "versions": "0+", "mapKey": true, 
        "about": "The configuration key name." }, 
      { "name": "Value", "type": "string", "versions": "0+", "nullableVersions": "0+", 
        "about": "The value to set for the configuration key."} 
    ]} 
  ] 
} 

...

{ 
  "apiKey": TBD, 
  "type": "response", 
  "name": "CreateMirrorResponse", 
  "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+", "ignorable": true, 
      "about": "The error message, or null if there was no error." } 

  ] 

}

AddTopicsToMirror

The AddTopicsToMirror API adds topics to a specified mirror. The broker validates that all target topic partitions are in either UNKNOWN or STOPPED state; otherwise, the request is rejected with an INVALID_REQUEST error. Once validated, the request is forwarded to the controller, which sets the mirror.name topic config to the specified mirror name.

AddTopicsToMirrorRequest

{
  "apiKey":TBD,
  "type": "request",
  "listeners": ["broker", "controller"],
  "name": "AddTopicsToMirrorRequest",
  "validVersions": "0",
  "flexibleVersions": "0+",
  "fields": [
    { "name": "Topics", "type": "[]TopicState", "versions": "0+", "about": "The topic state.",
      "fields": [
        { "name": "TopicId", "type": "uuid", "versions": "0+", "about": "The unique topic ID."},
        { "name": "TopicName", "type": "string", "versions": "0+", "mapKey": true, "entityType": "topicName",
          "about": "The topic name." },
        { "name": "MirrorName", "type": "string", "versions": "0+", "nullableVersions": "0+",
          "about": "The mirror name."}
      ]}
  ]
}

...

{
  "apiKey":TBD,
  "type": "response",
  "name": "AddTopicsToMirrorResponse",
  "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": "TopicId", "type": "uuid", "versions": "0+", "about": "The unique topic ID."},
    { "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+", "ignorable": true,
      "about": "The error message, or null if there was no error." }
  ]
}

RemoveTopicsFromMirror

RemoveTopicsFromMirrorRequest

The RemoveTopicsFromMirror API allows users to detach topics from their associated mirror. 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.

RemoveTopicsFromMirrorRequest

{
  "apiKey": TBD,
  "type": "request",
  "listeners": ["broker", "controller"],
  "name": "RemoveTopicsFromMirrorRequest",
  "validVersions": "0",
  "flexibleVersions": "0+",
  "fields": [
    { "name": "MirrorName", "type": "string", "versions": "0+", "ignorable": true,
      "about": "The cluster mirror name." },
    { "name": "Topics", "type": "[]TopicState", "versions": "0+", "about": "The topic state.",
      "fields": [
        { "name": "TopicId", "type": "uuid", "versions": "0+", "about": "The unique topic ID."},
        { "name": "TopicName", "type": "string", "versions": "0+", "mapKey": true, "entityType": "topicName",
          "about": "The topic name." }
      ]}
  ]
}

...

{
  "apiKey": TBD,
  "type": "response",
  "name": "RemoveTopicsFromMirrorResponse",
  "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": "TopicId", "type": "uuid", "versions": "0+", "about": "The unique topic ID."},
    { "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+", "ignorable": true,
      "about": "The error message, or null if there was no error." }
  ]
}

LastMirroredOffset

ListMirroredOffsetsRequest

The LastMirroredOffset API allows destination cluster partition leaders in PREPARING state to query the last mirrored offset from the source cluster. If the source cluster has no record of this offset in its internal topic, it returns 0, meaning the log must be truncated to the beginning and mirroring starts from scratch. This is particularly important during failback. The last mirrored offset identifies where mirrored data ends and un-mirrored data begins. Records beyond this offset must be truncated before mirroring new data from the new source cluster; otherwise, the two clusters would contain inconsistent data.

ListMirroredOffsetsRequest

{
  "apiKey": TBD,
  "type": "request",
  "listeners": ["broker", "

{
  "apiKey": TBD,
  "type": "request",
  "listeners": ["broker", "controller"],
  "name": "LastMirroredOffsetsRequest",
  "validVersions": "0",
  "flexibleVersions": "0+",
  "fields": [
    { "name": "MirrorName", "type": "string", "versions": "0+", "about": "The mirror name." },
    { "name": "Topics", "type": "[]TopicState", "versions": "0",
      "about": "The responses per topic.", "fields": [
      { "name": "Name", "type": "string", "versions": "0", "entityType": "topicName",
        "about": "The topic name." },
      { "name": "Partitions", "type": "[]PartitionState", "versions": "0",
        "about": "The responses per partition.", "fields": [
        { "name": "PartitionIndex", "type": "int32", "versions": "0",
          "about": "The partition index." }
      ]}
    ]}
  ]
}

...

{
  "apiKey": TBD,
  "type": "response",
  "name": "LastMirroredOffsetsResponse",
  "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": "[]OffsetResponseTopic", "versions": "0",
      "about": "The responses per topic.", "fields": [
      { "name": "Name", "type": "string", "versions": "0", "entityType": "topicName",
        "about": "The topic name." },
      { "name": "Partitions", "type": "[]OffsetResponsePartition", "versions": "0",
        "about": "The responses per partition.", "fields": [
        { "name": "PartitionIndex", "type": "int32", "versions": "0",
          "about": "The partition index." },
        { "name": "LastMirroredOffset", "type": "int64", "versions": "0",
          "about": "The last mirrored record offset." },
        { "name": "ErrorCode", "type": "int16", "versions": "0",
          "about": "The error code, or 0 if there was no error." }
      ]}
    ]}
  ]
}

ListMirrors

The ListMirrors API returns the current mirror names and their associated topic counts in the cluster.

ListMirrorsRequest

{
  "apiKey": TBD,
  "type": "request",
  "listeners": ["broker"],
  "name": "ListMirrorsRequest",
  // Version 0 is the initial version.
  "validVersions": "0",
  "flexibleVersions": "0+",
  "fields": []
}

...

{
  "apiKey": TBD,
  "type": "response",
  "name": "ListMirrorsResponse",
  // Version 0 is the initial version.
  "validVersions": "0",
  "flexibleVersions": "0+",
  "fields": [
    { "name": "ThrottleTimeMs", "type": "int32", "versions": "0+", "ignorable": true,
      "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": "Mirrors", "type": "[]ListedMirror", "versions": "0+",
      "about": "Each mirror in the response.", "fields": [
      { "name": "MirrorName", "type": "string", "versions": "0+",
        "about": "The mirror name." },
      { "name": "SourceBootstrap", "type": "string", "versions": "0+",
        "about": "The source cluster bootstrap servers." },
      { "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

The DescribeMirrors API is to retrieve the information about the mirror names, including the partition state, lag, source offset and destination offset.

DescribeMirrorsRequest

{
  "apiKey": TBD,
  "type": "request",
  "listeners": ["broker"],
  "name": "DescribeMirrorsRequest",
  // Version 0 is the initial version.
  "validVersions": "0",
  "flexibleVersions": "0+",
  "fields": [
    { "name": "MirrorNames", "type": "[]string", "versions": "0+",
      "about": "The names of the mirrors to describe. Null or empty array means all mirrors." },
    { "name": "IncludeAuthorizedOperations", "type": "bool", "versions": "0+", "default": "false",
      "about": "Whether to include authorized operations." }
  ]
}

...

{
  "apiKey": TBD,
  "type": "response",
  "name": "DescribeMirrorsResponse",
  // 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": "Mirrors", "type": "[]DescribedMirror", "versions": "0+",
      "about": "Each described mirror.", "fields": [
      { "name": "ErrorCode", "type": "int16", "versions": "0+",
        "about": "The error code, or 0 if there was no error." },
      { "name": "MirrorName", "type": "string", "versions": "0+",
        "about": "The mirror name." },
      { "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 detail.", "fields": [
          { "name": "PartitionIndex", "type": "int32", "versions": "0+",
            "about": "The partition index." },
          { "name": "SourceOffset", "type": "int64", "versions": "0+",
            "about": "The high watermark offset from the source cluster leader." },
          { "name": "DestinationOffset", "type": "int64", "versions": "0+",
            "about": "The log end offset on the destination cluster." },
          { "name": "Lag", "type": "int64", "versions": "0+",
            "about": "The lag (source offset - destination offset)." },
          { "name": "State", "type": "string", "versions": "0+",
            "about": "The partition state (INITIALIZING, PREPARING, MIRRORING, STOPPING, STOPPED, FAILED)." }
        ]}
      ]},
      { "name": "AuthorizedOperations", "type": "int32", "versions": "0+", "default": "-2147483648",
        "about": "32-bit bitfield to represent authorized operations for this mirror." }
    ]}
  ]
}

ReadMirrorStates

ReadMirrorStatesRequest

The ReadMirrorStates RPC reads mirror states from the coordinator broker when it resides on a different node than the requesting broker.

ReadMirrorStatesRequest

{

{
  "apiKey": TBD,
  "type": "request",
  "listeners": ["broker", "controller"],
  "name": "ReadMirrorStatesRequest",
  "validVersions": "0",
  "flexibleVersions": "0+",
  "fields": [
    { "name": "MirrorName", "type": "string", "versions": "0+", "about": "The mirror name." },
    { "name": "Topics", "type": "[]TopicState", "versions": "0",
      "about": "The responses per topic.", "fields": [
      { "name": "Name", "type": "string", "versions": "0", "entityType": "topicName",
        "about": "The topic name." },
      { "name": "Partitions", "type": "[]PartitionState", "versions": "0",
        "about": "The responses per partition.", "fields": [
        { "name": "PartitionIndex", "type": "int32", "versions": "0",
          "about": "The partition index." }
        ]}
      ]},
    { "name": "NeedPartitionStates", "type": "bool", "versions": "0+", "default": "true",
      "about": "Need the partition states or only topics states needed." }
  ]
}

...

{
  "apiKey": TBD,
  "type": "response",
  "name": "ReadMirrorStatesResponse",
  "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": "[]TopicState", "versions": "0",
      "about": "The responses per topic.", "fields": [
      { "name": "Name", "type": "string", "versions": "0", "entityType": "topicName",
        "about": "The topic name." },
      { "name": "Partitions", "type": "[]PartitionState", "versions": "0",
        "about": "The responses per partition.", "fields": [
        { "name": "PartitionIndex", "type": "int32", "versions": "0",
          "about": "The partition index." },
        { "name": "LastMirroredOffset", "type": "int64", "versions": "0",
          "about": "The last mirrored record offset." },
        { "name": "state", "type": "int8", "versions": "0+",
          "about": "The mirror partition state." },
        { "name": "ErrorCode", "type": "int16", "versions": "0",
          "about": "The error code, or 0 if there was no error." }
      ]}
    ]}
  ]
}

WriteMirrorStates

The WriteMirrorStates RPC writes mirror state updates to the coordinator broker when it resides on a different node than the requesting broker.

WriteMirrorStatesRequest

{
  "apiKey": TBD,
  "type": "request",
  "listeners": ["broker", "controller"],
  "name": "WriteMirrorStatesRequest",
  "validVersions": "0",
  "flexibleVersions": "0+",
  "fields": [
    { "name": "MirrorName", "type": "string", "versions": "0+", "about": "The mirror name." },
    { "name": "TopicsUpdated", "type": "[]TopicState", "versions": "0",
      "about": "The topics to be updated.", "fields": [
      { "name": "Name", "type": "string", "versions": "0", "entityType": "topicName",
        "about": "The topic name." },
      { "name": "Partitions", "type": "[]PartitionState", "versions": "0",
        "about": "The responses per partition.", "fields": [
        { "name": "PartitionIndex", "type": "int32", "versions": "0",
          "about": "The partition index." },
        { "name": "LastMirroredOffset", "type": "int64", "versions": "0",
          "about": "The last mirrored record offset." },
        { "name": "state", "type": "int8", "versions": "0+",
          "about": "The mirror partition state." }
      ]}
    ]},
    { "name": "RemovedTopics", "type": "[]string", "versions": "0+", "about": "The topic names to be removed." }
  ]
}

...

public enum CoordinatorType { 
    GROUP((byte) 0), 

    TRANSACTION((byte) 1), 

    MIRROR((byte) 2); // New type
}

...

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


{
  "apiKey": 2,
  "type": "coordinator-value",
  "name": "MirrorPartitionStateValue",
  "validVersions": "0",
  "flexibleVersions": "0+",
  "fields": [
    { "name": "TopicName", "type": "string", "versions": "0",
      "about": "The topic name."},
    { "name": "Partition", "type": "int32", "versions": "0",
      "about": "The partition index."},
    { "name": "State", "type": "int8", "versions": "0+",
      "about": "The mirror partition state." }
  ]
}

...

...

...

...

...

When Cluster Mirroring reaches general availability, the feature will be enabled by default when clusters reach the corresponding production metadata version. All new APIs will become stable production APIs with all unstable markers removed from their definition. No special configuration flags or explicit feature enablement will be required beyond setting an appropriate metadata version, and the feature will be fully supported for mission-critical production workloads under Kafka's standard compatibility guarantees. Clusters using Cluster Mirroring in preview can upgrade seamlessly to GA releases without migration steps. Downgrade is also supported, but it would require manual cleanup of the internal topic.

Migration

...

From MirrorMaker 2

Cluster Mirror is not compatible with MirrorMaker 2 (MM2). This is a critical consideration for users planning to migrate from MirrorMaker 2 to Cluster Mirroring.

...

...

• Separate Thread Pools: Cross-cluster fetcher threads run in a dedicated thread pool, which is independent from the intra-cluster fetcher thread pool. This separation ensures that cross-cluster replication latency does not impact local replica synchronization.
• Network I/O Overhead: Read-only leaders perform additional network I/O to fetch from source clusters. This overhead is proportional to the number of mirror partitions and the replication throughput. Brokers with many mirror partitions may experience increased CPU usage for network processing and data serialization.
• Memory Footprint: Each mirror fetcher thread maintains its own fetch session state, partition state map, and response buffers. With default configuration, memory overhead is comparable to standard replica fetchers. The metadata manager maintains connection pools and metadata caches, adding minimal memory overhead.
• Bandwidth Consumption: Cross-cluster traffic between source and destination clusters consumes WAN bandwidth. For large-scale deployments, administrators should provision adequate inter-datacenter connectivity or configure throttling.State Management: Mirror partition state management is evenly distributed to available brokers to avoid any hot spot, especially during rolling update or restart eventsscale deployments, administrators should provision adequate inter-datacenter connectivity or configure throttling.
• State Management: Mirror partition state management is evenly distributed to available brokers to avoid any hot spot, especially during rolling update or restart events.

Future Work

Sync mirroring: Currently, mirroring is asynchronous. The source cluster acknowledges the producer without waiting for the destination to replicate the data. Sync mirroring would guarantee that records are replicated to the destination cluster before the source acknowledges the produce request, providing stronger durability guarantees at the cost of higher latency. This would be useful for workloads where zero data loss across clusters is a strict requirement.

Tiered storage: Mirror topics in the destination cluster currently only replicate data from local storage on the source broker. Integrating with tiered storage would allow mirroring to handle data that has been offloaded to remote storage (e.g., S3, HDFS), enabling full replication of topics with long retention periods without requiring all data to reside in local broker storage.

Diskless topics: Diskless topics store data exclusively in tiered storage, with no local log segments on brokers. Supporting mirroring for diskless topics requires adapting the fetch and replication mechanisms to work without local storage, which introduces changes to how mirror offsets are tracked and how truncation is handled during failover.

Source cluster throttling: The current replication throttling mechanism was designed for intra-cluster replica reassignment and may not be well suited for cross-cluster mirroring traffic. A dedicated throttling method would allow operators to control the bandwidth consumed by mirror fetch requests on the source cluster independently, preventing mirroring from competing with client traffic or internal replication without the limitations of repurposing the existing quota framework.

Test Plan

Unit Tests

Unit tests will cover individual component behavior:

...

Keep using MirrorMaker 2:

This KIP is introduces native cluster mirroring to address the drawbacks existing limitations of MirrorMaker 2 as described in the motivation section. 

Support unclean leader election

...

source cluster
leader for foo-0 contains this data:
offset 0, epoch: 0, value: A
offset 1, epoch: 1, value: B

Suppose we mirror everything from the source into destination cluster, including the leader epoch in batches:
target cluster
leader for foo-0 contains this data:
offset 0, epoch: 0, value: A
offset 1, epoch: 1, value: B
===

This could happen:

1. leadership change in the source cluster, bumping the leader epoch to 2
2. New records appended to source cluster: offset=2, epoch: =2, value: =C
3. Before target cluster fetch from the source to identify the leader epoch update, source cluster down, failover happened
4. Topics in destination cluster becomes writable, new records appended from producer:   offset=2, epoch: =2, value: =D
5. Keep using MirrorMaker 2:
   This KIP is to address the drawbacks existing MirrorMaker 2 as described in the motivation section. 

...

The issue above can be resolved by the the LastMirroredOffset API we did in this KIP. The flow will be like this:

1. leadership Leadership change in the source cluster, bumping the leader epoch to 2
2. New records appended to source cluster: offset=2, epoch: =2, value: =C
3. Before target cluster fetch from the source to identify the update, source cluster down, failover happened
4. When failover, the destination cluster will store the current last mirrored offset (1 in this case) into internal topic.
5. Topics in destination cluster becomes writable, new records appended from producer:   offset=2, epoch: =2, value: =D
6. When the old source cluster wants to reverse mirroring to the new source cluster, it'll firstly ask for the last mirrored offset, which is 1 in 1 in this case. Then, truncate data to offset 1.
7. Then, start fetch from offset 1.

It works well, but when unclean leader election comes into the play, it'll become complicated:
 

1. unclean Unclean leader election happened and leadership change in the source cluster, bumping the leader epoch to 2.
2. New leader has empty log in disk.
3. New records appended to source cluster: offset=0, epoch: =2, value: =C.
4. Before target cluster fetch from the source to identify the update, source cluster down, failover happened
5. When failover, the destination cluster will store the current last mirrored offset (1 in this case) into internal topic.
6. Topics in destination cluster becomes writable, new records appended from producer:   offset=2, epoch: =2, value: =D.
7. When the old source cluster wants to reverse mirroring to the new source cluster, it'll firstly ask for the last mirrored offset, which is 1 in 1 in this case. Then, truncate data to offset 1.

...