...

A cluster mirror is a named, unidirectional replication channel from a remote source cluster to the local destination cluster. It is created by specifying a unique mirror name along with the source cluster's bootstrap servers and security credentials. Once a mirror is created, individual topics on the source cluster can be started, stopped, or paused for replication within it. Each mirror is a first-class entity managed through the Admin API and the kafka-cluster-mirrors.sh CLI tool, with its state persisted in a coordinator that manages cross-cluster replication.

Image Modified

The architecture consists of three main components that work together to provide automatic metadata synchronization and data replication. The following diagram illustrates how these components are wired together. For the sake of clarity, some internal APIs are excluded.

Image RemovedImage Added

The mirror name is stored as a topic-level internal configuration called mirror.name that has the same validation rules of topic names, and propagates through Kafka's metadata log as configuration change records. When topics are added to a mirror, the quorum controller generates configuration metadata records that are replicated to all brokers through the standard metadata update mechanism. Brokers monitor these configuration changes to detect when partitions they lead belong to a mirror, triggering the creation of mirror fetchers and enforcement of read-only semantics. This design ensures that mirror associations are visible, auditable, and manageable through standard Kafka tools while maintaining strict control over how mirroring relationships are established and modified.

Main Components

...

ClusterMirrorCoordinator

The ClusterMirrorCoordinator (MCCMC) manages Cluster Mirroring state using a partitioned coordinator partitioned coordinator pattern similar to the group and transaction coordinators.

...

• State Management: Mirror partition states and control records are stored in __mirror_state internal topic. The coordinator loads these metadata on startup and partition leadership changes.
• Partition Assignment: Mirror partitions are distributed evenly to coordinators across all brokers and allows for horizontal scaling. The number of coordinator partitions is configurable via via mirror.state.topic.num.partitions.
• Leader Election: When a broker becomes the leader for a __mirror_state partition, it loads the mirror metadata for all mirrors assigned to that partition and begins coordinating those mirrors. On resignation, it clears its in-memory state to avoid stale metadata.
• Metadata Refresh Scheduling: The coordinator schedules periodic metadata refresh operations by invoking a metadata manager every 30 seconds by default. This ensures that configuration updates and group offset commits in the source cluster are continuously propagated to the destination cluster. The refresh interval is configurable via mirror.metadata.refresh.interval.ms.
• State Transitions: The coordinator manages asynchronous state transitions for mirror partitions. Each partition is an independent replication unit with its own state. When the coordinator is the leader for a mirror partition, it writes the state updates directly to __mirror_state topic, otherwise it reads and writes the state via new internal RPCs, enabling distributed coordination across the cluster.

Image RemovedImage Added

State descriptions:

• 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.
• PREPARINGLOG_TRUNCATION: The coordinator for this partition detects via onMetadataUpdate that it leads a mirror partition. It fetches last mirror epoch from the source cluster and truncates 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.
• EPOCH_FENCING: The destination leader epoch needs to be bumped. A BumpLeaderEpochs request is sent to the controller. On success, the partition transitions to MIRRORING.
• PAUSING: Triggered by the pause operation. The system removes fetchers for the affected partitions.
• PAUSED: Fetchers have been removed. The partition stays read-only with no active fetchers and no metadata synchronization. On resume, transitions directly to MIRRORING.
• STOPPING: The mirror fetcher is removed, the leader epoch is bumped, ABORT markers are appended for all ongoing transactions, last mirror epochs are persisted, and a MIRROR_PID_RESET record appended.
• 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 during mirroring or stopping. The operator that wants to restart a failed mirror partition can restart topic mirroring. More sophisticated recovery strategies can be added later with a follow-up KIPcoordinator tries to recover the mirror partition and then fail permanently. The operator can still use the CLI tool or admin API to manually recover.

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 synchronization for that mirror is halted and an error is logged. This prevents silent data corruption in case of misconfiguration or unintended source cluster replacement.

...

...

...

In a follow-up KIP we will add source-side throttling that allows source cluster leaders to limit bandwidth served to all mirror fetchers, similar to how leader.replication.throttled.rate controls intra-cluster replication. This provides independent control over mirror catch-up traffic without impacting local replication or consumer workloads. Combined with destination-side throttling, operators gain complete bidirectional bandwidth control for mirror traffic.

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. During the LOG_TRUNCATION state, the mirror truncates the local log to the LME. This operation does not support tiered storage on the destination cluster because LME may be moved to remote segments.

When tiered storage is enabled locally for a mirror topic, its partitions transition to FAILED state. This limitation applies only to the destination cluster during log truncation phase, and will be removed once tiered storage truncation will be fully supported. A detailed design of the metadata synchronization protocol, API schema, and state management will be provided in a follow-up KIP.

...

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. Instead, bidirectional mirroring is supported, but only when mirroring different topics between clusters, allowing records produced to either cluster to be consumed from both. Unlike MirrorMaker 2, Cluster Mirroring does not need special cycle detection or prevention logic because the read-only enforcement inherently blocks the conditions that would create infinite replication loops.

Synchronous Mirroring

Streams Applications

Stateful Kafka Streams rely on internal topics (changelogs, repartition topics, offset tracking) that are tightly coupled through atomic transactions. A single EOS transaction spans input offset commits, state store mutations written to changelog topics, and intermediate records written to repartition topics. Cluster mirroring replicates topics asynchronously and independently, so it cannot preserve these transactional boundaries. This means internal topics on the destination cluster can end up at inconsistent points in time relative to each other, making state store recovery produce incorrect results. The synchronous mirroring extension would preserve these guarantees. For this reason, mirroring of Kafka Streams internal topics is not  supported.

Synchronous Mirroring

Currently, mirroring is Currently, mirroring is asynchronous. The source cluster acknowledges the producer without waiting for the destination to replicate the data. Synchronous 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.

...

# 9091 (source) -----> 9094 (destination)
# in case of disaster, the operator can failover by running the following command
bin/kafka-cluster-mirrormirrors.sh --bootstrap-server :9094 --stop --topic .* --mirror my-mirror
# 9091 (source) --x--> 9094 (destination)
# now all mirror topics are detached from the source cluster and accept writes (the two clusters are allowed to diverge)

...

Before transitioning a mirror partition from PREPARING to LOG_TRUNCATION 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 reverse mirroring may cause the data loss if there are records that didn't get mirrored to the old destination cluster.

# 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-cluster-mirrors.sh --bootstrap-server :9091 --create --mirror my-mirror --mirror-config /tmp/my-mirror.properties
bin/kafka-cluster-mirrors.sh --bootstrap-server :"9091 --start --topic .* --mirror my-mirror
# 9091 (destination) <----- 9094 (source)

Create Mirror

1. The user sends CreateMirror CreateClusterMirror 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.

...

1. User sends StartMirrorTopicsRequest with mirror name, topics, and optional include/exclude patterns.
2. Controller persists include/exclude patterns as ConfigRecord entries on the MIRROR resource in the metadata log.
3. For each topic, the controller creates it on the destination if it does not already exist, and sets mirror.name=<mirrorName> on the TOPIC resource config. Both operations are written in a single metadata record batch.
4. Brokers receive the metadata update. The MirrorMetadataManager detects the new mirror.name config (without .stopped or .paused suffix) and queries the coordinator for the current partition state.
5. Partitions transition from UNKNOWN to PREPARINGLOG_TRUNCATION. During PREPARINGthis state, LME truncation runs and waits for all ISR members (or all replicas if ULE is enabled).
6. Partitions transition to MIRRORING. A MirrorFetcherThread is created and begins fetching from the source cluster.
7. Partition state is persisted to __mirror_state.
8. On subsequent metadata refresh cycles, the MirrorMetadataManager discovers new source topics matching the persisted include/exclude patterns and repeats steps 3-7 for each.

Stop Mirror Topics

1. User sends StopMirrorTopicsRequest StopMirrorTopics request with mirror name, topics, and optional patterns.
2. If patterns are provided, the controller removes matching entries from mirror.topics.include and adds them to mirror.topics.exclude on the MIRROR resource in the metadata log. Any currently mirroring topic that matches the updated exclude is also stopped.
3. For each topic, the controller writes a ConfigRecord updating mirror.name=<mirrorName>.stopped on the TOPIC resource.
4. Brokers receive the metadata update. The MirrorMetadataManager detects the .stopped suffix and transitions partitions to STOPPING.
5. During STOPPING, the following operations execute sequentially:
   
   1. Fetcher threads are removed for the affected partitions stopping replication.
   2. The current leader epoch is collected and persisted as LME in __mirror_state.
   3. The partition's leader epoch is bumped to draw a boundary between mirrored and locally produced records.

   4. ABORT markers are appended for all ongoing transactions using the new leader epoch.

   5. A MIRROR_PID_RESET control record is written to expire all producer state entries.

6. Partitions transition to STOPPED and becomes writable.

...

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 updates the .paused suffix to the mirror name config, e.g. to mirror.name=my-mirror<mirrorName>.paused, generating a ConfigRecord.
3. When the MirrorMetadataManager in the partition leader node gets notified, it detects the .paused suffix . It and 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 paused for the paused these topics.
6. The partition state change is persisted to the __mirror_state topic.

...

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, and 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 DeleteClusterMirror 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. The mirror state records in __mirror_state internal topic are also tombstoned.
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.

List Mirrors

1. The user sends ListMirrorsRequest ListClusterMirrors request 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 DescribeClusterMirrors request 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, current state, and LME.
5. No metadata records are written. This is a read only operation.

...

$ bin/kafka-dump-log.sh --mirror-state-decoder --files /tmp/server*/data/__mirror_state-1/00000000000000000000.log
Dumping /home/fvaleri/Documents/kafka/build/test/server4/data/__mirror_state-0/00000000000000000000.log
Log starting offset: 0
baseOffset: 0 lastOffset: 0 count: 1 baseSequence: -1 lastSequence: -1 producerId: -1 producerEpoch: -1 partitionLeaderEpoch: 0 isTransactional: false isControl: false deleteHorizonMs: OptionalLong.empty position: 0 CreateTime: 1771239071191 size: 98 magic: 2 compresscodec: none crc: 3314855113 isvalid: true
| offset: 0 CreateTime: 1771239071191 keySize: 13 valueSize: 17 sequence: -1 headerKeys: [] key: {"type":"2","data":{"mirrorName":"my-mirror"}} payload: {"version":"0","data":{"topicName":"my-topic","partition":0,"state":0}}
baseOffset: 1 lastOffset: 1 count: 1 baseSequence: -1 lastSequence: -1 producerId: -1 producerEpoch: -1 partitionLeaderEpoch: 0 isTransactional: false isControl: false deleteHorizonMs: OptionalLong.empty position: 98 CreateTime: 1771239071219 size: 98 magic: 2 compresscodec: none crc: 3968746657 isvalid: true
| offset: 1 CreateTime: 1771239071219 keySize: 13 valueSize: 17 sequence: -1 headerKeys: [] key: {"type":"2","data":{"mirrorName":"my-mirror"}} payload: {"version":"0","data":{"topicName":"my-topic","partition":0,"state":1}}

A new command-line tool kafka-mirrors.sh provides administrative operations for managing cluster mirrorsAlternatively, you can use the new message formatter to consume from the internal metadata topic:

$ bin/kafka-console-mirrorsconsumer.sh --bootstrap-server :9094 --topic __mirror_state --from-beginning \
  --formatter org.apache.kafka.tools.consumer.MirrorStateMessageFormatter
{"key":{"type":2,"data":{"mirrorName":"my-mirror"}},"value":{"version":0,"data":{"topicName":"my-topic","partition":0,"state":0}}}
{"key":{"type":2,"data":{"mirrorName":"my-mirror"}},"value":{"version":0,"data":{"topicName":"my-topic","partition":0,"state":2}}}

A new command-line tool kafka-cluster-mirrors.sh provides administrative operations for managing cluster mirrors:

$ bin/kafka-cluster-mirrors.sh --help
Create cluster mirrors and manage mirror topics.
Optionhelp
Create cluster mirrors and manage mirrored topics.
Option                                  Description                           
------                                  -----------                                               
--bootstrap-server <String: server to   REQUIRED: The destination Kafka server
  connect to>                             to connect to.                      
--command-config <String: command       Property file containing configs to be
  config property file>                   passed to Admin Client.             
--create      Description                          Create a new cluster mirror from a 
------         
                         -----------                 source cluster.                     
--delete        
--bootstrap-server <String: server to   REQUIRED: The destination Kafka server
  connect to>           Delete a cluster mirror.              
--describe to connect to.                      
--command-config <String: command   Describe a cluster mirror includingProperty file containing 
configs to be
  config property file>                   passed to Admin   Client.            partition lag and state.            
--excludecreate <String: exclude patterns>    Comma-separated list of topic names or
                    Create a new cluster mirror from a    
            regex patterns to exclude from      
                          source cluster.                mirroring. Only valid with --start. 
--helpdelete                                Delete a Printcluster usage informationmirror.              
--jsondescribe                              Describe a cluster mirror including  Output description
 in JSON format     
--list                                  Listpartition alllag clusterand mirrorsstate.             
--mirrorexclude <String: mirror>exclude patterns>    Comma-separated list of topic names or
     The name of the cluster mirror.       
--mirror-config <String: mirror config  Property file containing source       
  property file>       regex patterns to exclude from      
         cluster configs for mirroring.      
--pause                        mirroring. Only valid with --start. 
--help    Pause mirroring for topics matching   
                       Print usage information.              
--json   the given patterns.                 
--resume                 Output description in JSON format     
--list                      Resume mirroring for previously paused
        List all cluster mirrors.             
--mirror <String: mirror>               The topicsname matchingof the givencluster patternsmirror.       
--startmirror-config <String: mirror config  Property file containing source       
  property file>                  Start mirroring topics matching the   
 cluster configs for mirroring.      
--pause                                given patterns.Pause mirroring for topics matching   
              
--stop                            the given patterns.    Stop mirroring topics matching the    
     
--resume                                Resume mirroring for previously paused
 given patterns.                     
--topics <String: topics>               Comma-separated list oftopics topicmatching namesthe or
given patterns. 
--start                                 Start mirroring topics matching the   regex
 patterns (e.g., 'my-topic,    
                                   given patterns.      orders-.*,payments').               
--versionstop                                  DisplayStop mirroring Kafka version.

Create a new cluster mirror in the destination cluster (forbidden suffixes: .stopped, .paused):

$ echo "bootstrap.servers=localhost:9092" >/tmp/mirror.properties
$ bin/kafka-mirror.sh --bootstrap-server :9094 --create --mirror my-mirror --mirror-config /tmp/mirror.properties
Created mirror my-mirror

Start mirroring a topic or set of topics (the topic flag accepts regex expression):

$ bin/kafka-mirrors.sh --bootstrap-server :9094 --start \topics matching the    
                                          given patterns.                     
--topics <String: topics>               Comma-separated list of topic names or
                                          regex patterns (e.g., 'my-topic,    
                                          orders-.*,payments').               
--version                               Display        
  --topics 'orders-.*' --exclude 'orders-internal' --mirror my-mirror
Started 2 mirror topic(s) in mirror my-mirror: [orders-us, orders-eu]Kafka version.

Create a new cluster mirror in the destination cluster (forbidden suffixes: .stopped, .pausedStop mirroring a topic or set of topics (failover; topics become writable):

$ echo "bootstrap.servers=localhost:9092" >/tmp/mirror.properties
$ bin/kafka-cluster-mirrors.sh --bootstrap-server :9094 --stopcreate --topicsmirror 'ordersmy-us'mirror --mirror my-mirror
Stopped mirroring for 1 topic(s) in -config /tmp/mirror.properties
Created mirror my-mirror: [orders-us]

Start mirroring a topic or set of topics (the --topics flag accepts regex expressionDelete a mirror including its configuration (the mirror must be empty or include only stopped partitions):

$ bin/kafka-cluster-mirrors.sh --bootstrap-server :9094 --delete --mirror my-mirror
Deleted mirror my-mirror

Pause mirroring for a specific topic or set of topics (topics remain read-only):

$ bin/kafka-mirrors.sh --bootstrap-server :9094 --pause --topic my-topic --mirror my-mirror
Paused mirroring for 1 topic(s) in mirror my-mirror: [my-topic]

Resume mirroring for a specific topic or set of topics:

$ bin/kafka-mirrors.sh --bootstrap-server :9094 --resume --topic my-topic --mirror my-mirror
Resumed mirroring for 1 topic(s) in mirror my-mirror: [my-topic]

...

$ bin/kafka-mirrors.sh --bootstrap-server :9094 --list
MIRRORstart \                                                                           TOPICS     CLUSTER-ID                 BOOTSTRAP-SERVER
my-mirror                      2          lBq12jYZRp-9wF3M9MPopg     localhost:9091,localhost:9092
new-mirror                     1          lBq12jYZRp-9wF3M9MPopg     localhost:9091,localhost:9092

Describe configured mirrors to check their lag compared to their source topics (use --mirror flag to filter other partitions):

$ bin/kafka-mirrors.sh --bootstrap-server :9094 --describe
MIRROR                         TOPIC                           
  --topics 'orders-.*' --exclude 'orders-internal' --mirror my-mirror
Started 2 mirror topic(s) in mirror my-mirror: [orders-us, orders-eu]

Stop mirroring a topic or set of topics (failover; topics become writable):

$ bin/kafka-cluster-mirrors.sh --bootstrap-server :9094 --stop --topics 'orders-us' --mirror my-mirror
Stopped mirroring for 1 topic(s) in mirror my-mirror: [orders-us]

Delete a mirror including its configuration (the mirror must be empty or include only stopped partitions):

$ bin/kafka-cluster-mirrors.sh --bootstrap-server :9094 --delete --mirror my-mirror
Deleted mirror my-mirror

Pause mirroring for a specific topic or set of topics (topics remain read-only):

$ bin/kafka-cluster-mirrors.sh --bootstrap-server :9094 --pause --topics my-topic --mirror my-mirror
Paused mirroring for 1 topic(s) in mirror my-mirror: [my-topic]

Resume mirroring for a specific topic or set of topics:

$ bin/kafka-cluster-mirrors.sh --bootstrap-server :9094 --resume --topics my-topic --mirror my-mirror
Resumed mirroring for 1 topic(s) in mirror my-mirror: [my-topic]

List configured mirrors with additional information:

$ bin/kafka-cluster-mirrors.sh --bootstrap-server :9094 --list
MIRROR    PARTITION  SOURCE-OFFSET   DESTINATION-OFFSET LAG      STATE       
my-mirror                      bar                                      0          -               -  TOPICS     CLUSTER-ID           -        STOPPED   BOOTSTRAP-SERVER
my-mirror                      2 foo         lBq12jYZRp-9wF3M9MPopg     localhost:9091,localhost:9092
new-mirror                     1   0       lBq12jYZRp-9wF3M9MPopg     localhost:9091,localhost:9092

Describe configured mirrors to check their lag compared to their source topics (use --mirror flag to only show partitions from a specific mirror):

$ bin/kafka-cluster-mirrors.sh --bootstrap-server :9094 --describe
MIRROR69              66                 3TOPIC        MIRRORING   
my-mirror                      foo   PARTITION  SOURCE-OFFSET   DESTINATION-OFFSET LAG      STATE       
my-mirror                1      bar    94              84                 10   0    MIRRORING   
my-mirror   -               -    foo              -        STOPPED   
my-mirror             2         foo 94              90                 4      0  MIRRORING   
new-mirror     69              66       baz          3        MIRRORING   
my-mirror                 0     foo     -               -                  -      1  PAUSED   
new-mirror     94              84  baz               10       MIRRORING   
my-mirror             1         foo -               -                  -        PAUSED

Alter mirror configuration (any valid configuration triggers a reconnection):

$ bin/kafka-configs.sh --bootstrap-server :9094 --entity-type mirrors --entity-name my-mirror \
    --alter --add-config 'bootstrap.servers=localhost:9092'
Completed updating config for mirror my-mirror.

Throttling on the destination cluster:

$ bin/kafka-configs.sh --bootstrap-server :9094 --entity-type brokers --entity-name 4 \
  --alter --add-config mirror.replication.throttled.rate=100000000
Completed updating config for broker 4.

$ bin/kafka-configs.sh --bootstrap-server :9094 --entity-type topics --entity-name my-topic \
  --alter --add-config mirror.replication.throttled.replicas=[0:4]
Completed updating config for topic my-topic.

Throttling on the source cluster:

$ bin/kafka-configs.sh --bootstrap-server :9091 --alter --add-config 'consumer_byte_rate=1024' \
  --entity-type clients --entity-name broker-4-fetcher-0-mirror-my-mirror
Completed updating config for client broker-4-fetcher-0-mirror-my-mirror.

Grant mirror admin full access to a specific mirror:

$ bin/kafka-acls.sh --bootstrap-server :9094 --add \
  --cluster-mirror my-mirror \
  --operation Create --operation Alter --operation Describe --operation Delete \
  --operation AlterConfigs --operation DescribeConfigs \
  --allow-principal User:mirror-admin
Adding ACLs for resource `ResourcePattern(resourceType=CLUSTER_MIRROR, name=my-mirror, patternType=LITERAL)`:
      (principal=User:mirror-admin, host=*, operation=CREATE, permissionType=ALLOW)
      (principal=User:mirror-admin, host=*, operation=ALTER, permissionType=ALLOW)
      (principal=User:mirror-admin, host=*, operation=DESCRIBE, permissionType=ALLOW)
      (principal=User:mirror-admin, host=*, operation=DELETE, permissionType=ALLOW)
      (principal=User:mirror-admin, host=*, operation=ALTER_CONFIGS, permissionType=ALLOW)
      (principal=User:mirror-admin, host=*, operation=DESCRIBE_CONFIGS, permissionType=ALLOW)

Grant read-only monitoring access to all mirrors:

$ bin/kafka-acls.sh --bootstrap-server :9094 --add \
  --cluster-mirror '*' \
  --operation Describe --operation DescribeConfigs \
  --allow-principal User:monitor
Adding ACLs for resource `ResourcePattern(resourceType=CLUSTER_MIRROR, name=*, patternType=LITERAL)`:
      (principal=User:monitor, host=*, operation=DESCRIBE, permissionType=ALLOW)
      (principal=User:monitor, host=*, operation=DESCRIBE_CONFIGS, permissionType=ALLOW)

List ACLs for a specific mirror:

$ bin/kafka-acls.sh --bootstrap-server :9094 --list --cluster-mirror my-mirror
Current ACLs for resource `ResourcePattern(resourceType=CLUSTER_MIRROR, name=my-mirror, patternType=LITERAL)`: 2          94              90                 4        MIRRORING   
new-mirror                     baz                                      0          -               -                  -        PAUSED   
new-mirror                     baz                                      1          -               -                  -        PAUSED

Alter mirror configuration (any valid configuration triggers a reconnection):

$ bin/kafka-configs.sh --bootstrap-server :9094 --entity-type cluster-mirrors --entity-name my-mirror \
    --alter --add-config 'bootstrap.servers=localhost:9092'
Completed updating config for mirror my-mirror.

Throttling on the destination cluster:

$ bin/kafka-configs.sh --bootstrap-server :9094 --entity-type brokers --entity-name 4 \
  --alter --add-config mirror.replication.throttled.rate=100000000
Completed updating config for broker 4.

$ bin/kafka-configs.sh --bootstrap-server :9094 --entity-type topics --entity-name my-topic \
  --alter --add-config mirror.replication.throttled.replicas=[0:4]
Completed updating config for topic my-topic.

Throttling on the source cluster:

$ bin/kafka-configs.sh --bootstrap-server :9091 --alter --add-config 'consumer_byte_rate=1024' \
  --entity-type clients --entity-name broker-4-fetcher-0-mirror-my-mirror
Completed updating config for client broker-4-fetcher-0-mirror-my-mirror.

Grant mirror admin full access to a specific mirror:

$ bin/kafka-acls.sh --bootstrap-server :9094 --add \
  --cluster-mirror my-mirror \
  --operation Create --operation Alter --operation Describe --operation Delete \
  --operation AlterConfigs --operation DescribeConfigs \
  --allow-principal User:mirror-admin
Adding ACLs for resource `ResourcePattern(resourceType=CLUSTER_MIRROR, name=my-mirror, patternType=LITERAL)`:                                                                                            
      (principal=User:mirror-admin, host=*, operation=CREATE, permissionType=ALLOW)
      (principal=User:mirror-admin, host=*, operation=ALTER, permissionType=ALLOW)
      (principal=User:mirror-admin, host=*, operation=DESCRIBE, permissionType=ALLOW)
      (principal=User:mirror-admin, host=*, operation=DELETE, permissionType=ALLOW)
      (principal=User:mirror-admin, host=*, operation=ALTER_CONFIGS, permissionType=ALLOW)
      (principal=User:mirror-admin, host=*,                          operation=DESCRIBE_CONFIGS, permissionType=ALLOW)

Grant read-only monitoring access to all mirrors:

$ bin/kafka-acls.sh --bootstrap-server :9094 --add \
  --cluster-mirror '*' \
  --operation Describe --operation DescribeConfigs \
  --allow-principal User:monitor
Adding ACLs for resource `ResourcePattern(resourceType=CLUSTER_MIRROR, name=*, patternType=LITERAL)`:
      (principal=User:monitor, host=*, operation=DESCRIBE, permissionType=ALLOW)
      (principal=User:monitor, host=*, operation=DESCRIBE_CONFIGS, permissionType=ALLOW)

List ACLs for a specific mirror:

$ bin/kafka-acls.sh --bootstrap-server :9094 --list --cluster-mirror my-mirror
Current ACLs for resource `ResourcePattern(resourceType=CLUSTER_MIRROR, name=my-mirror, patternType=LITERAL)`:                                                                                                   
      (principal                                                                  
      (principal=User:mirror-admin, host=*, operation=ALTERCREATE, permissionType=ALLOW)                                                                                                                                                                                                       
      (principal=User:mirror-admin, host=*, operation=DESCRIBEALTER, permissionType=ALLOW)                                                                                                                                                                                                       
      (principal=User:mirror-admin, host=*, operation=DELETEDESCRIBE, permissionType=ALLOW)                                                                                                                                                                                                      
      (principal=User:mirror-admin, host=*, operation=ALTER_CONFIGSDELETE, permissionType=ALLOW)                                                                                                                                                                                                      
      (principal=User:mirror-admin, host=*, operation=DESCRIBEALTER_CONFIGS, permissionType=ALLOW)    

Admin Client

New methods are added to the Admin interface for programmatic cluster mirror management, along with their supporting classes:

/**
 * Create a new cluster mirror.
 *
 * @param mirrorName The name of the cluster mirror
 * @param configs Configuration for the cluster mirror, including bootstrap servers and security settings
 * @param options Options for the create mirror operation
 * @return The CreateMirrorResult
 */
CreateMirrorResult createMirror(String mirrorName, Map<String, String> configs, CreateMirrorOptions options);/**
 * Create a new cluster mirror.
 *
 * @param mirrorName The name of the cluster mirror
 * @param configs Configuration for the cluster mirror, including bootstrap servers and security settings
 * @param options Options for the create mirror operation
 * @return The CreateMirrorResult
 */
CreateMirrorResult createMirror(String mirrorName, Map<String, String> configs, CreateMirrorOptions options);

/**
 * Start mirroring for the specified topics.
 *
 * When topics are started in a mirror, they become read-only on the destination cluster and start
 * replicating data from the source cluster. This operation marks the specified topics with the
 * mirror name, preventing local writes and enabling the MirrorFetcherThread to begin replication.
 *
 * @param mirrorName The cluster mirror name
 * @param topics Set of topic names to start mirroring
 * @param options Options for the start mirror topics operation
 * @return The StartMirrorTopicsResult containing
 futures for each topic
 */
StartMirrorTopicsResult (principal=User:mirror-admin, host=*, operation=DESCRIBE_CONFIGS, permissionType=ALLOW)   

Admin Client

New methods are added to the Admin interface for programmatic cluster mirror management, along with their supporting classes:

startMirrorTopics(String mirrorName, Set<String> topics, StartMirrorTopicsOptions options);

/**
 * OptionsCreate fora {@link Admin#startMirrorTopics(String, Set, StartMirrorTopicsOptions)}new cluster mirror.
 */
public class StartMirrorTopicsOptions extends AbstractOptions<StartMirrorTopicsOptions> {
    private List<String> includePatterns = List.of();
    private List<String> excludePatterns = List.of();
    private Map<String, StartMirrorTopicsRequestData.TopicData> topicMetadata = Map.of();

    public StartMirrorTopicsOptions includePatterns(List<String> patterns) {
        this.includePatterns = patterns;
        return this;
    }

    public StartMirrorTopicsOptions excludePatterns(List<String> patterns) {
        this.excludePatterns = patterns;
        return this;
    }

    public StartMirrorTopicsOptions topicMetadata(Map<String, StartMirrorTopicsRequestData.TopicData> metadata) {
        this.topicMetadata = metadata;
        return this;
    }

    public List<String> includePatterns() {
        return includePatterns;
    }

    public List<String> excludePatterns() {
        return excludePatterns;
    }

    public Map<String, StartMirrorTopicsRequestData.TopicData> topicMetadata() {
        return topicMetadata;
    }
}

/**
 * Stop mirroring for the specified topics.
 *
 * This operation is typically used during failover scenarios when the destination cluster needs to
 * be promoted from passive (read-only mirror) to active (accepting writes). Stopping mirror topics
 * clears the mirrorName field from partition metadata, which allows producers to write
 * to these partitions.
 *
 *  * @param mirrorName The name of the cluster mirror
 * @param configs Configuration for the cluster mirror, including bootstrap servers and security settings
 * @param options Options for the create mirror operation
 * @return The CreateClusterMirrorResult
 */
CreateClusterMirrorResult createClusterMirror(String mirrorName, Map<String, String> configs, CreateClusterMirrorOptions options);/**
 * Create a new cluster mirror.
 *
 * @param mirrorName The name of the cluster mirror
 * @param configs Configuration for the cluster mirror, including bootstrap servers and security settings
 * @param options Options for the create mirror operation
 * @return The CreateClusterMirrorResult
 */
CreateClusterMirrorResult createClusterMirror(String mirrorName, Map<String, String> configs, CreateClusterMirrorOptions options);

/**
 * Start mirroring for the specified topics.
 *
 * When topics are started in a mirror, they become read-only on the destination cluster and start
 * replicating data from the source cluster. This operation marks the specified topics with the
 * mirror name, preventing local writes and enabling the MirrorFetcherThread to begin replication.
 *
 * @param mirrorName The cluster mirror name
 * @param topics Set of topic names to stopstart mirroring
 * @param options Options for the stopstart mirror topics operation
 * @return The StopMirrorTopicsResultStartMirrorTopicsResult containing futures for each topic
 */
StopMirrorTopicsResultStartMirrorTopicsResult stopMirrorTopicsstartMirrorTopics(String mirrorName, Set<String> topics, StopMirrorTopicsOptionsStartMirrorTopicsOptions options);

/**
 * Options for {@link Admin#stopMirrorTopicsAdmin#startMirrorTopics(String, Set, StopMirrorTopicsOptionsStartMirrorTopicsOptions)}.
 */
public class StopMirrorTopicsOptionsStartMirrorTopicsOptions extends AbstractOptions<StopMirrorTopicsOptions>AbstractOptions<StartMirrorTopicsOptions> {
    private List<String> patternsincludePatterns = List.of();
    private List<String> excludePatterns = List.of();
    private Map<String, StartMirrorTopicsRequestData.TopicData> topicMetadata = Map.of();

    public StopMirrorTopicsOptionsStartMirrorTopicsOptions patternsincludePatterns(List<String> patterns) {
        this.patternsincludePatterns = patterns;
        return this;
    }

    public StartMirrorTopicsOptions excludePatterns(List<String> patterns() {
        returnthis.excludePatterns = patterns;
    }
}

/**
 * Pause mirroring for the specified topics.
 *
 * Paused topics remain read-only on the destination cluster but stop fetching new data from the
 * source cluster. The mirror fetcher threads are removed for these partitions, preserving the
 * current replicated state. Mirroring can be resumed later with {@link #resumeMirrorTopics}.
 *
 * @param mirrorName The cluster mirror name
 * @param topics Set of topic names to pause mirroring
 * @param options Options for the pause mirror topics operation
 * @return The PauseMirrorTopicsResult containing futures for each topic
 */
PauseMirrorTopicsResult pauseMirrorTopics(String mirrorName, Set<String> topics, PauseMirrorTopicsOptions options);    return this;
    }

    public StartMirrorTopicsOptions topicMetadata(Map<String, StartMirrorTopicsRequestData.TopicData> metadata) {
        this.topicMetadata = metadata;
        return this;
    }

    public List<String> includePatterns() {
        return includePatterns;
    }

    public List<String> excludePatterns() {
        return excludePatterns;
    }

    public Map<String, StartMirrorTopicsRequestData.TopicData> topicMetadata() {
        return topicMetadata;
    }
}

/**
 * ResumeStop mirroring for previouslythe pausedspecified topics.
 *
 * Resumed topics restart fetching data from This operation is typically used during failover scenarios when the sourcedestination cluster, picking up from where they
 * left off. New mirror fetcher threads are created and the partitions transition back to the
 * MIRRORING state needs to
 * be promoted from passive (read-only mirror) to active (accepting writes). Stopping mirror topics
 * clears the mirrorName field from partition metadata, which allows producers to write
 * to these partitions.
 *
 * @param mirrorName The cluster mirror name
 * @param topics Set of topic names to resumestop mirroring
 * @param options Options for the resumestop mirror topics operation
 * @return The ResumeMirrorTopicsResultStopMirrorTopicsResult containing futures for each topic
 */
ResumeMirrorTopicsResultStopMirrorTopicsResult resumeMirrorTopicsstopMirrorTopics(String mirrorName, Set<String> topics, ResumeMirrorTopicsOptionsStopMirrorTopicsOptions options);

/**
 * DeleteOptions afor cluster mirror including its configuration{@link Admin#stopMirrorTopics(String, Set, StopMirrorTopicsOptions)}.
 */
public *class TheStopMirrorTopicsOptions mirrorextends mustAbstractOptions<StopMirrorTopicsOptions> be{
 empty   (noprivate topics)List<String> orpatterns all its topics must have been removed (in STOPPED
 * state). After deletion, all mirror metadata are tombstoned and failback is no longer possible.
 *
 * @param mirrorName The cluster mirror name
 * @param options Options for the delete mirror operation
 * @return The DeleteMirrorResult
 */
DeleteMirrorResult deleteMirror(String mirrorName, DeleteMirrorOptions options);= List.of();

    public StopMirrorTopicsOptions patterns(List<String> patterns) {
        this.patterns = patterns;
        return this;
    }

    public List<String> patterns() {
        return patterns;
    }
}

/**
 * ListPause themirroring clusterfor mirrorsthe available in the clusterspecified topics.
 *
 * @paramPaused optionstopics Theremain optionsread-only toon usethe whendestination listingcluster thebut mirrors.
stop *fetching @returnnew Thedata ListMirrorsResult.
 */
ListMirrorsResult listMirrors(ListMirrorsOptions options);

/**
from the
 * Describesource cluster mirrors.
 *
The *mirror Thisfetcher operationthreads retrievesare detailedremoved informationfor aboutthese clusterpartitions, mirrorspreserving including:the
 * -current Topicsreplicated beingstate. mirrored
Mirroring *can -be Partition-levelresumed laglater informationwith (source offset vs destination offset){@link #resumeMirrorTopics}.
 *
 * -@param MirroringmirrorName stateThe forcluster each partition (INITIALIZING, PREPARING, MIRRORING, etc.)mirror name
 *
 * @param mirrorNamestopics TheSet names of thetopic mirrorsnames to pause describemirroring
 * @param options TheOptions optionsfor tothe usepause whenmirror describingtopics mirrorsoperation
 * @return The DescribeMirrorsResult PauseMirrorTopicsResult containing futures for each topic
 */
DescribeMirrorsResultPauseMirrorTopicsResult describeMirrors(Collection<String> mirrorNames, DescribeMirrorsOptions options);

Protocol Changes

This section describes all protocol level changes.

CreateTopic

The CreateTopic API request is extended to add information required for mirror topic creation.

{ "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).

Fetch

The Fetch API is extended to add the MirrorLeaderEpoch field used by the destination cluster internal replication.

// new request field in FetchPartition type
{ "name": "MirrorLeaderEpoch", "type": "int32", "versions": "19+", "default": "-1", "taggedVersions": "19+", "tag": 2, "ignorable": true,
  "about": "The latest known mirror leader epoch." }

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

CreateMirror

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

pauseMirrorTopics(String mirrorName, Set<String> topics, PauseMirrorTopicsOptions options);

/**
 * Resume mirroring for previously paused topics.
 *
 * Resumed topics restart fetching data from the source cluster, picking up from where they
 * left off. New mirror fetcher threads are created and the partitions transition back to the
 * MIRRORING state.
 *
 * @param mirrorName The cluster mirror name
 * @param topics Set of topic names to resume mirroring
 * @param options Options for the resume mirror topics operation
 * @return The ResumeMirrorTopicsResult containing futures for each topic
 */
ResumeMirrorTopicsResult resumeMirrorTopics(String mirrorName, Set<String> topics, ResumeMirrorTopicsOptions options);

/**
 * Delete a cluster mirror including its configuration.
 *
 * The mirror must be empty (no topics) or all its topics must have been removed (in STOPPED
 * state). After deletion, all mirror metadata are tombstoned and failback is no longer possible.
 *
 * @param mirrorName The cluster mirror name
 * @param options Options for the delete mirror operation
 * @return The DeleteClusterMirrorResult
 */
DeleteClusterMirrorResult deleteClusterMirror(String mirrorName, DeleteClusterMirrorOptions options);

/**
 * List the cluster mirrors available in the cluster.
 *
 * @param options The options to use when listing the mirrors.
 * @return The ListClusterMirrorsResult.
 */
ListClusterMirrorsResult listClusterMirrors(ListClusterMirrorsOptions options);

/**
 * Describe cluster mirrors.
 *
 * This operation retrieves detailed information about cluster mirrors including:
 * - Topics being mirrored
 * - Partition-level lag information
 * - Mirroring state for each partition
 *
 * @param mirrorNames The names of the mirrors to describe
 * @param options The options to use when describing mirrors
 * @return The DescribeClusterMirrorsResult
 */
DescribeClusterMirrorsResult describeClusterMirrors(Collection<String> mirrorNames, DescribeClusterMirrorsOptions options);

Protocol Changes

This section describes all protocol level changes.

CreateTopic

The CreateTopic API request is extended to add information required for mirror topic creation.

{ "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

{
  "apiKey": TBD,
  "type": "request",
  "listeners": ["broker", "controller"],
  "name": "CreateMirrorRequest",
  // Version 0 is the initial version.
  "validVersions": "0",
  "flexibleVersions": "0+",
  "fields": [
    { "name": "MirrorName", "type": "string", "versions": "0+", "entityType": "mirrorName",
      "about": "The cluster mirror name."},
    { "name": "Config", "type": "[]MirrorConfig", "versions": "0+",
      "about": "The cluster mirror configurations.",  "fields": [
      { "name": "Name", "type": "stringuuid", "versions": "08+", "mapKey": true,
    
        "about": "The configuration key nametopic ID from the source cluster." },
      { "name": "Value", "type": "string", "versions": "0+", "nullableVersions": "0+",
        "about": "The value to set for the configuration key."}
    ]}
  ]
}

{
  "apiKey": TBD,
  "type": "response",
  "name": "CreateMirrorResponse",
  // Version 0 is the initial version.
  "validVersions": "0",
  "flexibleVersions": "0+",
  "fields": [
    
]}

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

Fetch

The Fetch API is extended to add the MirrorLeaderEpoch field used by the destination cluster internal replication.

// new request field in FetchPartition type
{ "name": "ThrottleTimeMsMirrorLeaderEpoch", "type": "int32", "versions": "019+", "default": "-1", "taggedVersions": "19+",
 "tag": 2, "ignorable": true,
  "about": "The durationlatest inknown millisecondsmirror for which the request was throttled due to a quota violation, or zero if the request did not violate any quotaleader epoch." },
    

// new response field in PartitionData type
{ "name": "ErrorCodeMirrorLeaderEpoch", "type": "int16int32", "versions": "019+",
      "aboutdefault": "The error code, or 0 if there was no error." },
    { "name-1", "taggedVersions": "ErrorMessage19+", "typetag": "string"3, "versionsignorable": "0+", "nullableVersions": "0+",true,
      "about": "The errorlatest message,known ormirror null if there was no errorleader epoch." },
  ]
}

StartMirrorTopics

CreateClusterMirror

Allows users to create a mirror and supply its configurationStart 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 the mirror name is not already in use, contains only permitted characters, and does not end with .stopped or .paused suffix. Once validated, the request is forwarded to the controller, which sets the mirror.name topic config to the specified mirror namepersists the configuration in the metadata log.

{
  "apiKey": TBD,
  "type": "request",
  "listeners": ["broker", "controller"],
  "name": "StartMirrorTopicsRequestCreateClusterMirrorRequest",
  // Version 0 is the initial version.
  "validVersions": "0",
  "flexibleVersions": "0+",
  "fields": [
    { "name": "MirrorName", "type": "string", "versions": "0+", "entityType": "mirrorName",
      "about": "The cluster mirror name." },
    { "name": "TopicsConfig", "type": "[]TopicDataClusterMirrorConfig", "versions": "0+",
      "about": "The datacluster formirror the topicsconfigurations.",
      "fields": [
        { "name": "TopicIdName", "type": "uuidstring", "versions": "0+", "mapKey": true,
        "about": "The uniqueconfiguration topickey IDname." },
        { "name": "TopicNameValue", "type": "string", "versions": "0+", "mapKeynullableVersions": true, "entityType": "topicName"0+",
          "about": "The topic name value to set for the configuration key." },
    ]}
    {]
}

{
  "nameapiKey": "NumPartitions"TBD,
  "type": "int32response",
  "versionsname": "0+CreateClusterMirrorResponse",
  // Version 0 is the initial version.
  "aboutvalidVersions": "0"The,
 number of partitions for the topic. Must match the source topic." }
      ]},
    { "name": "IncludePatterns"flexibleVersions": "0+",
  "fields": [
    { "name": "ThrottleTimeMs", "type": "[]stringint32", "versions": "0+",
      "about": "Regex patterns to add to mirror.topics.includeThe 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": "ExcludePatternsErrorCode", "type": "[]stringint16", "versions": "0+",
      "about": "RegexThe patternserror tocode, addor to mirror.topics.exclude." }
  ]
}

{
  "apiKey": TBD0 if there was no error." },
  "type": "response",
 { "name": "StartMirrorTopicsResponseErrorMessage",
  // Version 0 is the initial version.
  "validVersions"type": "0string",
  "flexibleVersionsversions": "0+",
  "fields": [
    { "name": "ThrottleTimeMs", "type": "int32", "versions": ""nullableVersions": "0+",
      "about": "The durationerror inmessage, millisecondsor fornull whichif the requestthere was throttledno 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": "request",
  "listeners": ["broker", "controller"],
  "name": "StartMirrorTopicsRequest",
  // Version 0 is the initial version.
  "validVersions": "0",
  "flexibleVersions": "0+",
  "fields": [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": "ErrorMessageMirrorName", "type": "string", "versions": "0+", "nullableVersionsentityType": "0+mirrorName", "default": "null",
      
      "about": "The top-levelcluster error message, or null if there was no errormirror name." },
    { "name": "MirrorNameTopics", "type": "string[]TopicData", "versions": "0+", "entityTypeabout": "mirrorNameThe data for the topics.",
      "aboutfields": "The[
 cluster mirror name." },
    { "name": "TopicsTopicId", "type": "[]TopicResultuuid", "versions": "0+",
      "about": "The resultsunique fortopic the topicsID."},
 "fields": [
      { "name": "NameTopicName", "type": "string", "versions": "0+", "mapKey": true, "entityType": "topicName",
          "about": "The topic name." },
        { "name": "ErrorCodeNumPartitions", "type": "int16int32", "versions": "0+",
          "about": "The error code, or 0 if there was no errornumber of partitions for the topic. Must match the source topic." }
      ]},
    { "name": "IncludePatterns", "type": "[]
}

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 .stopped suffix to the mirror.name topic config to mark the topics as no longer mirrored.

{
  "apiKey": TBD,
  "type": "request",
  "listeners": ["broker", "controller"]string", "versions": "0+",
      "about": "Regex patterns to add to mirror.topics.include." },
    { "name": "ExcludePatterns", "type": "[]string", "versions": "0+",
      "about": "Regex patterns to add to mirror.topics.exclude." }
  ]
}

{
  "apiKey": TBD,
  "type": "response",
  "name": "StopMirrorTopicsRequestStartMirrorTopicsResponse",
  // Version 0 is the initial version.
  "validVersions": "0",
  "flexibleVersions": "0+",
  "fields": [
    { "name": "MirrorNameThrottleTimeMs", "type": "stringint32", "versions": "0+",
      "about": "The cluster mirror nameduration in milliseconds for which the request was throttled due to a quota violation, or zero if the request did not violate any quota." },
    { "name": "TopicsErrorCode", "type": "[]TopicDataint16", "versions": "0+",
      "about": "The dataerror forcode, the topics.",
      "fields": [
  or 0 if there was no error." },
      { "name": "TopicIdErrorMessage", "type": "uuidstring", "versions": "0+", "nullableVersions": "0+", "default": "null",
      "about": "The unique topic ID."},
     top-level error message, or null if there was no error." },
    { "name": "TopicNameMirrorName", "type": "string", "versions": "0+", "mapKey": true, "entityType": "topicNamemirrorName",
          "about": "The cluster topicmirror name." }
      ]},
    { "name": "PatternsTopics", "type": "[]stringTopicResult", "versions": "0+",
      "about": "PatternsThe toresults updatefor inthe mirror.topics.include/exclude." }
  ]
}

{
  "apiKey": TBD,
  ", "fields": [
      { "name": "Name", "type": "responsestring",
  "versions": "0", "nameentityType": "StopMirrorTopicsResponsetopicName",
  // Version 0 is the initial version.
  "validVersionsabout": "0",
The topic "flexibleVersionsname.": "0+"},
  "fields": [
    { "name": "ThrottleTimeMsErrorCode", "type": "int32int16", "versions": "0+",
        "about": "The durationerror incode, millisecondsor for0 whichif the requestthere was throttledno error." }
    ]}
  ]
}

StopMirrorTopics

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

{
  "apiKey": TBD,
  "type": "request",
  "listeners": ["broker", "controller"],
  "name": "StopMirrorTopicsRequest",
  // Version 0 is the initial version.
  "validVersions": "0",
  "flexibleVersions": "0+",
  "fields": [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": "ErrorMessageMirrorName", "type": "string", "versions": "0+", "nullableVersionsentityType": "0+mirrorName", "default": "null",
      
      "about": "The top-levelcluster error message, or null if there was no errormirror name." },
    { "name": "MirrorNameTopics", "type": "string[]TopicData", "versions": "0+", "entityTypeabout": "mirrorNameThe data for the topics.",
      "aboutfields": "The[
 cluster mirror name." },
    { "name": "TopicsTopicId", "type": "[]TopicResultuuid", "versions": "0+",
      "about": "The resultsunique fortopic the topicsID."},
 "fields": [
      { "name": "NameTopicName", "type": "string", "versions": "0+", "mapKey": true, "entityType": "topicName",
          "about": "The topic name." },
      ]},
    { "name": "ErrorCodePatterns", "type": "int16[]string", "versions": "0+",
        "about": "ThePatterns errorto code,update or 0 if there was no errorin mirror.topics.include/exclude." }
    ]
}

{
  ]
}

PauseMirrorTopics

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

{
  "apiKey""apiKey": TBD,
  "type": "requestresponse",
  "listeners": ["broker", "controller"],
  "name": "PauseMirrorTopicsRequestStopMirrorTopicsResponse",
  // Version 0 is the initial version.
  "validVersions": "0",
  "flexibleVersions": "0+",
  "fields": [
    { "name": "MirrorNameThrottleTimeMs", "type": "stringint32", "versions": "0+",
      "about": "The mirrorduration in namemilliseconds tofor pausewhich the topicsrequest for." },
    { "name": "Topics", "type": "[]TopicData", "versions": "0+", "about": "The data for the topics.",
      "fields": [was throttled due to a quota violation, or zero if the request did not violate any quota." },
        { "name": "TopicIdErrorCode", "type": "uuidint16", "versions": "0+",
      "about": "The unique topic ID." error code, or 0 if there was no error." },
        { "name": "TopicNameErrorMessage", "type": "string", "versions": "0+", "mapKeynullableVersions": true"0+", "entityTypedefault": "topicNamenull",
          ""about": "The topic name." }
      ]}
  ]
}

{
  "apiKey": TBD,
  top-level error message, or null if there was no error." },
    { "name": "MirrorName", "type": "responsestring",
  "nameversions": "PauseMirrorTopicsResponse0+",
  // Version 0 is the initial version.
  "validVersionsentityType": "0mirrorName",
      "flexibleVersionsabout": "0+",
  "fields": [The cluster mirror name." },
    { "name": "ThrottleTimeMsTopics", "type": "int32[]TopicResult", "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 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.

{
  "nameapiKey": "ErrorMessage"TBD,
  "type": "stringrequest",
  "versionslisteners": ["0+broker", "nullableVersionscontroller": "0+",],
  "defaultname": "nullPauseMirrorTopicsRequest",
  // Version 0 is the initial version.
  "aboutvalidVersions": "0"The,
 top-level error message, or null if there was no error." }, "flexibleVersions": "0+",
  "fields": [
    { "name": "MirrorName", "type": "string", "versions": "0+", "entityType": "mirrorName",
         "about": "The cluster mirror name to pause the topics for." },
    { "name": "Topics", "type": "[]TopicResultTopicData", "versions": "0+",
      "about": "The resultsdata for the topics.",
      "fields": [
        { "name": "NameTopicId", "type": "stringuuid", "versions": "0", "entityType": "topicName+",
        "about": "The unique topic nameID." },
        { "name": "ErrorCodeTopicName", "type": "int16string", "versions": "0+",
 "mapKey": true, "entityType": "topicName",
          "about": "The error code, or 0 if there was no errortopic name." }
      ]}
  ]
}

ResumeMirrorTopics

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

{
  "apiKey": TBD,
  "type": "request",
  "listeners": ["broker", "controller"],response",
  "name": "ResumeMirrorTopicsRequestPauseMirrorTopicsResponse",
  // Version 0 is the initial version.
  "validVersions": "0",
  "flexibleVersions": "0+",
  "fields": [
    { "name": "MirrorNameThrottleTimeMs", "type": "stringint32", "versions": "0+",
      "about": "The cluster mirror name 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": "TopicsErrorCode", "type": "[]TopicDataint16", "versions": "0+",
      "about": "The dataerror forcode, the topics.",
      "fields": [or 0 if there was no error." },
        { "name": "TopicIdErrorMessage", "type": "uuidstring", "versions": "0+", "aboutnullableVersions": "The unique topic ID."},
        0+", "default": "null",
      "about": "The top-level error message, or null if there was no error." },
    { "name": "TopicNameMirrorName", "type": "string", "versions": "0+", "mapKey": true, "entityType": "topicNamemirrorName",
          "about": "The cluster topicmirror name." },
      ]}
  ]
}

{
  "apiKey": TBD,
     { "name": "Topics", "type": "response[]TopicResult",
  "nameversions": "ResumeMirrorTopicsResponse0",
  // Version 0 is the initial version.
  "validVersionsabout": "0",
  "flexibleVersions": "0+",
  The results for the topics.", "fields": [
      { "name": "ThrottleTimeMsName", "type": "int32string", "versions": "0+", "entityType": "topicName",
        "about": "The durationtopic 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+",
      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.

{
  "nameapiKey": "ErrorMessage"TBD,
  "type": "stringrequest",
  "versionslisteners": ["0+broker", "nullableVersionscontroller": "0+",],
  "defaultname": "nullResumeMirrorTopicsRequest",
  // Version 0 is the "aboutinitial version.
  "validVersions": "The top-level error message, or null if there was no error." },0",
  "flexibleVersions": "0+",
  "fields": [
    { "name": "MirrorName", "type": "string", "versions": "0+", "entityType": "mirrorName",
         "about": "The cluster mirror name." },
    { "name": "Topics", "type": "[]TopicResultTopicData", "versions": "0+",
      "about": "The resultsdata for the topics.",
      "fields": [
      {  { "name": "NameTopicId", "type": "stringuuid", "versions": "0", "entityType": "topicName",
        +", "about": "The unique topic nameID." },
        { "name": "ErrorCodeTopicName", "type": "int16string", "versions": "0+", "mapKey": true, "entityType": "topicName",
          "about": "The error code, or 0 if there was no errortopic name." }
      ]}
  ]
}

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": "requestresponse",
  "listeners": ["broker", "controller"],
  "name": "DeleteMirrorRequestResumeMirrorTopicsResponse",
  // Version 0 is the initial version.
  "validVersions": "0",
  "flexibleVersions": "0+",
  "fields": [
    { "name": "MirrorNameThrottleTimeMs", "type": "stringint32", "versions": "0+", "entityType": "mirrorName",
      "about": "The clusterduration mirrorin namemilliseconds to delete."}
  ]
}

{
  "apiKey": TBD,
  "type": "response",
  "name": "DeleteMirrorResponse",
  // Version 0 is the initial version.
  "validVersions": "0",
  "flexibleVersions": "0+",
  "fields": [
  for which the request was throttled due to a quota violation, or zero if the request did not violate any quota." },
    { "name": "ThrottleTimeMsErrorCode", "type": "int32int16", "versions": "0+",
      "about": "The duration in milliseconds for which the request was throttled due to a quota violationerror code, or zero0 if thethere requestwas did not violate any quotano error." },
    { "name": "ErrorCodeErrorMessage", "type": "int16string", "versions": "0+", "nullableVersions": "0+", "default": "null",
      "about": "The top-level error codemessage, or 0null if there was no error." },
    { "name": "ErrorMessageMirrorName", "type": "string", "versions": "0+", "nullableVersionsentityType": "0+mirrorName",
      "about": "The errorcluster message, or null if there was no error." }
  ]
}

ListMirrors

Returns the current mirror names and their associated topic counts in the cluster. It also includes source cluster ID and bootstrap server.

{
  "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": [
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": "ThrottleTimeMsErrorCode", "type": "int32int16", "versions": "0+",
        "about": "The durationerror incode, millisecondsor for0 whichif the requestthere was throttledno 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." },error." }
    ]}
  ]
}

DeleteClusterMirror

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": "request",
  "listeners": ["broker", "controller"],
  "name": "DeleteClusterMirrorRequest",
  // Version 0 is the initial version.
  "validVersions": "0",
  "flexibleVersions": "0+",
  "fields": [
    { "name": "ErrorMessageMirrorName", "type": "string", "versions": "0+", "nullableVersionsentityType": "0+", "default": "nullmirrorName",
      "about": "The top-levelcluster errormirror message,name or null if there was no error." },
    { "name": "Mirrors", to delete."}
  ]
}

{
  "apiKey": TBD,
  "type": "[]ListedMirrorresponse",
  "versionsname": "0+DeleteClusterMirrorResponse",
  // Version 0 is the initial version.
  "aboutvalidVersions": "0"Each,
 mirror in the response."flexibleVersions": "0+",
  "fields": [
      { "name": "MirrorNameThrottleTimeMs", "type": "stringint32", "versions": "0+",
      "entityTypeabout": "mirrorName",
        "about": "The cluster mirror nameThe 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": "SourceBootstrapErrorCode", "type": "stringint16", "versions": "0+",
        "about": "The source cluster bootstrap servers error code, or 0 if there was no error." },
      { "name": "SourceClusterIdErrorMessage", "type": "string", "versions": "0+", "defaultnullableVersions": "0+",
        "about": "The sourceerror cluster IDmessage, or emptynull if notthere was yetno resolvederror." },
  ]
}

ListClusterMirrors

Returns the current mirror names and their associated topic counts in the cluster. It also includes source cluster ID and bootstrap server.

{
  "apiKey": TBD,
      { "name": "TopicCount", "type": "int32request",
  "versionslisteners": ["0+broker"],
  "defaultname": "0ListClusterMirrorsRequest",
  // Version 0 is the initial version.
  "aboutvalidVersions": "0"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. Allows destination cluster partition leaders to query the LME from the source cluster.

"flexibleVersions": "0+",
  "fields": []
}

{
  "apiKey": TBD,
  "type": "request",
  "listeners": ["broker"]response",
  "name": "DescribeMirrorsRequestListClusterMirrorsResponse",
  // Version 0 is the initial version.
  "validVersions": "0",
  "flexibleVersions": "0+",
  "fields": [
    { "name": "MirrorNamesThrottleTimeMs", "type": "[]stringint32", "versions": "0+", "entityType": "mirrorName",
      "about": "The namesduration ofin the mirrors to describe. Null or empty array means all mirrorsmilliseconds for which the request was throttled due to a quota violation, or zero if the request did not violate any quota." },
    { "name": "IncludeAuthorizedOperationsErrorCode", "type": "boolint16", "versions": "0+", "default": "false",
      "about": "Whether to include authorized operationsThe error code, or 0 if there was no error." },
    ]
}

{
  "apiKeyname": TBD"ErrorMessage",
  "type": "responsestring",
  "nameversions": "DescribeMirrorsResponse0+",
  // Version 0 is the initial version.
  "validVersions"nullableVersions": "0+", "default": "0null",
      "flexibleVersionsabout": "0+",
  "fields": [The top-level error message, or null if there was no error." },
    { "name": "ThrottleTimeMsMirrors", "type": "int32[]ListedMirror", "versions": "0+",
      "about": "TheEach durationmirror 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+the response.", "fields": [
      { "name": "MirrorName", "type": "string", "versions": "0+", "entityType": "mirrorName",
        "about": "The errorcluster code, or 0 if there was no errormirror name." },
      { "name": "ErrorMessageSourceBootstrap", "type": "string", "versions": "0+",
 "nullableVersions": "0+", "default": "null",
      "about": "The top-levelsource errorcluster message, or null if there was no errorbootstrap servers." },
        { "name": "MirrorsSourceClusterId", "type": "[]DescribedMirrorstring", "versions": "0+", "default": "",
        "about": "EachThe source describedcluster mirror."ID, or "fields": [empty if not yet resolved." },
      { "name": "ErrorCodeTopicCount", "type": "int16int32", "versions": "0+", "default": "0",
        "about": "The errornumber code,of ortopics 0configured iffor therethis was mirror. 0 indicates an empty mirror with no errortopics." },
    ]}
  ]
}

DescribeClusterMirrors

Returns the current mirroring status, state, and configuration for the specified mirror topics on the destination cluster. Allows destination cluster partition leaders to query the LME from the source cluster.

{
  "nameapiKey": "MirrorName"TBD,
  "type": "stringrequest",
  "versionslisteners": ["0+broker"],
  "entityTypename": "mirrorNameDescribeClusterMirrorsRequest",
  // Version 0 is the initial version.
  "aboutvalidVersions": "0"The,
 cluster mirror name." }"flexibleVersions": "0+",
  "fields": [
    { "name": "AuthorizedOperationsMirrorNames", "type": "int32[]string", "versions": "0+", "defaultentityType": "-2147483648mirrorName",
        "about": "32-bit bitfieldThe names of the mirrors to represent authorized operations for this mirrordescribe. Null or empty array means all mirrors." },
       { "name": "TopicsIncludeAuthorizedOperations", "type": "[]TopicPartitionsbool", "versions": "0+",
 "default": "false",
      "about": "EachWhether topicto ininclude theauthorized mirroroperations.", "fields": [
        { "name": "TopicName",  }
  ]
}

{
  "apiKey": TBD,
  "type": "stringresponse",
  "versionsname": "0+DescribeClusterMirrorsResponse",
  // Version 0 is the initial version.
  "aboutvalidVersions": "0"The,
 topic name."flexibleVersions": }"0+",
    "fields": [
    { "name": "PartitionsThrottleTimeMs", "type": "[]PartitionDetailint32", "versions": "0+",
          "about": "Each partition detail.", "fields": [
          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": "PartitionIndexErrorCode", "type": "int32int16", "versions": "0+",
            "about": "The error partition index." },
      code, or 0 if there was no error." },
    { "name": "SourceOffsetErrorMessage", "type": "int64string", "versions": "0+", "nullableVersions": "0+", "default": "-1null",
            "about": "The hightop-level watermark offset from the source cluster leadererror message, or -1null if there notwas yetno availableerror." },
             { "name": "DestinationOffsetMirrors", "type": "int64[]DescribedMirror", "versions": "0+", "default": "-1",
            "about": "TheEach log end offset on the destination cluster, or -1 if not yet available." },
    described mirror.", "fields": [
      { "name": "LagErrorCode", "type": "int64int16", "versions": "0+", "default": "-1",
            "about": "The lag (source offset - destination offset)error code, or -10 if notthere was yetno availableerror." },
           { "name": "StateMirrorName", "type": "string", "versions": "0+",
 "entityType": "mirrorName",
          "about": "The partitioncluster mirror statename." },
          { "name": "AuthorizedOperations", "type": "int32", "versions": "0+", "default": "-12147483648",
            "about": "The32-bit lastbitfield mirrorto leaderrepresent epoch,authorized oroperations -1for ifthis not availablemirror." } ,
        ]}{ "name": "Topics", "type": "[]TopicPartitions", "versions": "0+",
      ]}
  "about": "Each ]}
topic  ]
}

BumpLeaderEpochs

Internal API that sets a minimum leader epoch on the specified partitions. The controller increments each partition's leader epoch to at least the requested value.

{
  "apiKey": TBD,
  "type": "request",
  "listeners": ["broker", "controller"],
  "name": "BumpLeaderEpochsRequestin the mirror.", "fields": [
        { "name": "TopicName", "type": "string", "versions": "0+",
  // Version 0 is the initial version.
  "validVersionsabout": "0",
The topic "flexibleVersionsname.": "0+"},
  "fields": [
      { "name": "TopicsPartitions", "type": "[]TopicStatePartitionDetail", "versions": "0+",
          "about": "TheEach topic and partitions statepartition detail.",
      "fields": [
          { "name": "TopicIdPartitionIndex", "type": "uuidint32", "versions": "0+",
            "about": "The uniquepartition topic IDindex." },
          { "name": "PartitionsSourceOffset", "type": "[]LeaderEpochStateint64", "versions": "0+", "aboutdefault": "The partition leader epochs.-1",
            "fieldsabout": [
"The high watermark offset from the source      {"name": "partitionIndex", "type": "int32", "versions": "0+", "about": "The partition index."cluster leader, or -1 if not yet available." },
           { {"name": "minLeaderEpochDestinationOffset", "type": "int32int64", "versions": "0+", "default": "-1",
            "about": "The minimumlog leaderend epochoffset thaton the destination cluster should bump to."}
   , or -1 if not yet available." },
       ]}
      ]}
  ]
}

{
  "apiKeyname": TBD"Lag",
  "type": "responseint64",
 "versions": "name0+", "default": "BumpLeaderEpochsResponse-1",
   // Version 0 is the initial version.
   "validVersionsabout": "0",
  "flexibleVersions": "0+",
  "fields": [
   The lag (source offset - destination offset), or -1 if not yet available." },
          { "name": "ThrottleTimeMsState", "type": "int32string", "versions": "0+",
            "about": "The durationpartition in milliseconds for which the request was throttled due to a quota violation, or zero if the request did not violate any quota." },
state." },
          { "name": "ErrorCode", "type": "int16int32", "versions": "0+", "default": "-1",
            "about": "The last mirror errorleader codeepoch, or 0-1 if there was no errornot available." }, 
        { "name": "Topics", "type": "[]TopicPartitions", "versions": "0+",]}
      ]}
    ]}
  ]
}

BumpLeaderEpochs

Internal API that sets a minimum leader epoch on the specified partitions. The controller increments each partition's leader epoch to at least the requested value.

{
  "apiKey": TBD,
  "type": "request",
  "listeners": ["broker", "controller"],
  "name": "BumpLeaderEpochsRequest",
  // Version 0 is the initial version.
  "validVersions": "0",
  "flexibleVersions": "0+",
  "fields": [
      "about": "Each topic in the mirror.", "fields": [
      { "name": "TopicName", "type": "string", "versions": "0+",
        "about": "The topic name." },
      { "name": "PartitionsTopics", "type": "[]PartitionDetailTopicState", "versions": "0+",
        "about": "Each partitionThe topic and partitions state.",
      "fields": [
        { "name": "PartitionIndexTopicId", "type": "int32uuid", "versions": "0+",
          "about": "The unique partitiontopic indexID." },
        { "name": "ErrorCodePartitions", "type": "int16[]LeaderEpochState", "versions": "0+",
          "about": "The partition errorleader codeepochs.",
  or 0 if there was no error." }   "fields": [
      ]}
    ]}
  ]
}

ReadMirrorStates

Internal API that reads the current mirror partition states from the internal __mirror_state topic on the destination cluster.

{
  "apiKeyname": TBD"partitionIndex",
  "type": "requestint32",
  "listenersversions": ["broker0+", "controller"],
  "nameabout": "ReadMirrorStatesRequest",
The  // Version 0 is the initial version.partition index."},
  "validVersions": "0",
  "flexibleVersions": "0+",
  "fields": [
    { "name": "MirrorNameminLeaderEpoch", "type": "stringint32", "versions": "0+", "entityTypedefault": -1, "mirrorNameabout",
: "The minimum leader epoch that "about": "Thethe destination cluster mirrorshould bump nameto."}
          ]},
      ]}
  ]
}

{
  "nameapiKey": "Topics"TBD,
  "type": "response"[]TopicData",
  "versionsname": "0BumpLeaderEpochsResponse",
  // Version 0 is the initial version.
  "aboutvalidVersions": "The0",
 data for the topics."flexibleVersions": "0+",
  "fields": [
      { "name": "NameThrottleTimeMs", "type": "stringint32", "versions": "0+",
      "entityTypeabout": "topicName"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 topic name error code, or 0 if there was no error." },
      { "name": "PartitionsTopics", "type": "[]PartitionDataTopicPartitions", "versions": "0+",
        "about": "TheEach datatopic forin the partitionsmirror.", "fields": [
        { "name": "PartitionIndexTopicName", "type": "int32string", "versions": "0+",
          "about": "The partitiontopic indexname." },
        ]}
      ]}
  ]
}

{
  "apiKey": TBD,
  { "name": "Partitions", "type": "response[]PartitionDetail",
  "nameversions": "ReadMirrorStatesResponse0+",
  // Version 0 is the initial version.
  "validVersionsabout": "0",
Each  "flexibleVersions": "0+partition state.",
  "fields": [
        { "name": "ThrottleTimeMsPartitionIndex", "type": "int32", "versions": "0+",
          "about": "The durationpartition in milliseconds for which the request was throttled due to a quota violation, or zero if the request did not violate any quota." },
    index." },
        { "name": "ErrorCode", "type": "int16", "versions": "0+",
          "about": "The error code, or 0 if there was no error." }
      ]},
    ]}
  ]
}

ReadMirrorStates

Internal API that reads the current mirror partition states from the internal __mirror_state topic on the destination cluster.

{
  "nameapiKey": "ErrorMessage"TBD,
  "type": "stringrequest",
  "versionslisteners": "0+", "nullableVersions["broker", "controller"],
  "name": "ReadMirrorStatesRequest",
  // Version 0 is the initial version.
  "validVersions": "0+",
  "defaultflexibleVersions": "null0+",
  "fields": [
    "about{ "name": "MirrorName", "type": "The top-level error message, or null if there was no error"string", "versions": "0+", "entityType": "mirrorName",
      "about": "The cluster mirror name." },
      { "name": "Topics", "type": "[]TopicResultTopicData", "versions": "0",
      "about": "The readdata results for the topics.", "fields": [
      { "name": "Name", "type": "string", "versions": "0", "entityType": "topicName",
        "about": "The topic name." },
      { "name": "Partitions", "type": "[]PartitionResultPartitionData", "versions": "0",
        "about": "The resultsdata for the partitions.", "fields": [
        { "name": "PartitionIndex", "type": "int32", "versions": "0",
          "about": "The partition index." },
        ]}
      ]}
  ]
}

{
  "nameapiKey": "LastMirrorEpoch"TBD,
  "type": "int32response",
 "versions": "0", defaultname": "-1ReadMirrorStatesResponse",
  // Version 0 is the initial version.
  "aboutvalidVersions": "0"The,
 last mirror leader epoch, or -1 if not available." },
       "flexibleVersions": "0+",
  "fields": [
    { "name": "StateThrottleTimeMs", "type": "int8int32", "versions": "0+",
          "about": "The mirror partition state 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." }
      ]},
    ]}
  ]
}

WriteMirrorStates

Internal API that persists mirror partition state transitions to the internal __mirror_state topic on the destination cluster.

{
  "apiKeyname": TBD"ErrorMessage",
  "type": "requeststring",
  "listenersversions": ["broker0+", "controller"],
  "namenullableVersions": "0+", "default": "WriteMirrorStatesRequestnull",
  // Version 0 is the initial version.
  "validVersions"about": "0",
  "flexibleVersions": "0+",
  "fields": [
    { "name": "MirrorName", "type": "string", "versions": "0+", "entityType": "mirrorName",
      "about": "The mirror nameThe top-level error message, or null if there was no error." },
    { "name": "Topics", "type": "[]TopicDataTopicResult", "versions": "0",
      "about": "The dataread results for the topics.", "fields": [
      { "name": "Name", "type": "string", "versions": "0", "entityType": "topicName",
        "about": "The topic name." },
      { "name": "Partitions", "type": "[]PartitionDataPartitionResult", "versions": "0",
        "about": "The dataresults for the partitions.", "fields": [
        { "name": "PartitionIndex", "type": "int32", "versions": "0",
          "about": "The partition index." },
        { "name": "LastMirrorEpoch", "type": "int32", "versions": "0", "default": "-1",
          "about": "The last mirrormirrored leader epoch, or -1 if not available." },
            { "name": "State", "type": "int8", "versions": "0+",
          "about": "The mirror partition state." },
      ]}
    ]},
    { "name": "StoppedTopicsErrorCode", "type": "[]stringint16", "versions": "0+",
          "about": "The topic names to be stopped error code, or 0 if there was no error." }
  ]
}

{
  "apiKey": TBD,
 ,
        { "name": "PreviousState", "type": "responseint8",
  "nametaggedVersions": "WriteMirrorStatesResponse0+",
  // Version "tag": 0, is the initial version.
"default": 16,
          "validVersionsabout": "0",
  "flexibleVersions": "0+"The mirror partition state before the last transition; UNKNOWN if not recorded." },
  "fields": [
      { "name": "ThrottleTimeMsRetryAttempt", "type": "int32int16", "versionstaggedVersions": "0+", "tag": 1, "default": 0,
          "about": "The durationnumber inof millisecondsautomatic forretry whichattempts thewhile requestin wasFAILED state." }
      ]}
    ]}
  ]
}

WriteMirrorStates

Internal API that persists mirror partition state transitions to the internal __mirror_state topic on the destination cluster.

{
  "apiKey": TBD,
  "type": "request",
  "listeners": ["broker", "controller"],
  "name": "WriteMirrorStatesRequest",
  // Version 0 is the initial version.
  "validVersions": "0",
  "flexibleVersions": "0+",
  "fields": [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": "ErrorMessageMirrorName", "type": "string", "versions": "0+", "nullableVersionsentityType": "0+mirrorName", "default": "null",
      "about": "The top-level error message, or null if there was no errormirror name." },
      { "name": "Topics", "type": "[]TopicResultTopicData", "versions": "0",
      "about": "The write resultsdata for the topics.", "fields": [
      { "name": "Name", "type": "string", "versions": "0", "entityType": "topicName",
        "about": "The topic name." },
      { "name": "Partitions", "type": "[]PartitionResultPartitionData", "versions": "0",
        "about": "The resultsdata for the partitions.", "fields": [
        { "name": "PartitionIndex", "type": "int32", "versions": "0",
          "about": "The partition index." },
        { "name": "ErrorCode", "type": "int16int32", "versions": "0", "default": "-1",
          "about": "The last mirror errorleader codeepoch, or -1 if not available." },
        { "name": "State", "type": "int8", "versions": "0 if there was no error+",
          "about": "The mirror partition state." }
      ]}
    ]},
   ]
}

Cluster Metadata Records

This section describes records written to the KRaft metadata log by the active controller as part of Cluster Mirroring operations.

PartitionChangeRecord

Written by the controller when processing a BumpLeaderEpochs request. The record carries a minLeaderEpoch field that sets a floor for the partition's leader epoch.

{
  "apiKey": 5,                                                                                                                                                                                   { "name": "StoppedTopics", "type": "[]string", "versions": "0+", "about": "The topic names to be stopped." }
  ]
}

{
  "apiKey": TBD,
  "type": "response",
  "name": "WriteMirrorStatesResponse",
  // 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": "Topics", "type": "[]TopicResult", "versions": "0",
      "about": "The write results for the topics.", "fields": [
      { "name": "Name", "type": "string", "versions": "0", "entityType": "topicName",
        "about": "The topic name." },
      { "name": "Partitions", "type": "[]PartitionResult", "versions": "0",
        "about": "The results for the partitions.", "fields": [
        { "name": "PartitionIndex", "type": "int32", "versions": "0",
          "about": "The partition index." },
        {               
  "name": "ErrorCode", "type": "metadataint16", "versions": "0",
          "about": "The error code, or 0 if there was no error." }
      ]}
    ]}
  ]
}

Cluster Metadata Records

This section describes records written to the KRaft metadata log by the active controller as part of Cluster Mirroring operations.

PartitionChangeRecord

Written by the controller when processing a BumpLeaderEpochs request. The record carries a minLeaderEpoch field that sets a floor for the partition's leader epoch.

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

MirrorPidResetRecord

A control record (type MIRROR_PID_RESET) written to each partition's data log during the STOPPING transition. 

{
  "type": "data",
  "name": "MirrorPidResetRecord",
  "validVersions": "0",
  "flexibleVersions": "0+",
  "fields": [
    { "name": "Version", "type": "int16", "versions": "0",
                                                   "about": "The version of the mirror PID reset record."},
    { "name": "SourceClusterId", "type": "string", "versions": "0",
      "about": "The source cluster UUID for verification."}
  ]
}

Mirror Metadata Records

This section describes records written to the __mirror_state internal topic by the MirrorCoordinator to track mirror state and synchronization points across brokers.

LastMirrorEpochs

Written during the STOPPING transition to record the last mirrored leader epoch for each partition before the destination becomes writable.

{
  "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 mirror 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 mirror epochs."flexibleVersions": "0+",
  "fields": [
    // ... existing fields { ...
    {"name": "PartitionIndexMinLeaderEpoch", "type": "int32", "versions": "03+",
    "default": -1,
      "about": "The minimum leader partitionepoch indexrequested." },
        { "name
  ]
}

Cluster Control Records

This section describes control records written to data log as part of Cluster Mirroring operations.

MirrorPidResetRecord

A control record (type MIRROR_PID_RESET) written to each partition's data log during the STOPPING transition. 

{
  "type": "data",
  "typename": "int32MirrorPidResetRecord",
  "versionsvalidVersions": "0+",
          "about"flexibleVersions": "The last mirror leader epoch for this partition." }0+",
      ]}"fields": [
    ]}
  ]
}

MirrorPartitionState

Written on every state transition. Tracks the current state of each mirrored partition.

{
  "apiKey": 2,
  "typename": "coordinator-keyVersion",
   "nametype": "MirrorPartitionStateKeyint16",
   "validVersionsversions": "0",
  "flexibleVersions      "about": "none",
  "fields": [
    The version of the mirror PID reset record."},
    { "name": "MirrorNameSourceClusterId", "type": "string", "versions": "0",
            "about": "The source cluster UUID mirrorfor nameverification."}
    ]
}

Mirror Metadata Records

This section describes records written to the __mirror_state internal topic by the MirrorCoordinator to track mirror state and synchronization points across brokers.

LastMirrorEpochs

Written during the STOPPING transition to record the last mirrored leader epoch for each partition before the destination becomes writable.

{
  

{
  "apiKey": 21,
    "type": "coordinator-valuekey",
    "name": "MirrorPartitionStateValueLastMirrorEpochsKey",
    "validVersions": "0",
    "flexibleVersions": "0+none",
    "fields": [
        { "name": "TopicNameMirrorName", "type": "string", "versions": "0", "entityType": "mirrorName",
            "about": "The topiccluster mirror name."}
  ]
}

{
  "apiKey": 1,
    {  "nametype": "Partitioncoordinator-value",
  "typename": "int32LastMirrorEpochsValue",
  "versionsvalidVersions": "0",
        "aboutflexibleVersions": "0+"The,
 partition index."},
    "fields": [
    { "name": "StateTopics", "type": "int8[]Topic", "versions": "0+",
            "about": "The mirror partition state." }
  ]
}

Type Enumerations

This section lists the new values added to existing Kafka type enumerations to support Cluster Mirroring

Configuration

Represents a mirror as a configurable resource in cluster metadata. Mirror-level properties such as source cluster bootstrap servers, security credentials, mirror.topics.include, and mirror.topics.exclude are stored under this type, keyed by mirror name.

public enum Type {
    // existing types unchanged 
    MIRROR((byte) 64, "mirror");
}

Entity

 A schema-level annotation for MirrorName string fields in protocol messages. The message generator uses it to validate that mirror name fields across all request/response schemas conform to the expected type.

public enum EntityType {
    // existing types unchanged
    @JsonProperty("mirrorName")
    MIRROR_NAME(FieldType.StringFieldType.INSTANCE);
}

Resource

An ACL resource type that represents a cluster mirror as a securable object. Authorization checks use this type with the mirror name as the resource name.

public enum ResourceType {
    // existing types unchanged
    CLUSTER_MIRROR((byte) 8);

Coordinator

A coordinator type for locating the broker responsible for a given mirror name. The coordinator partition is determined by hashing the mirror name across __mirror_state topic partitions.

public enum CoordinatorType { 
    // existing types unchanged
    MIRROR((byte) 3);
}

Configuration

This section describes new configurations introduced by Cluster Mirroring.

Broker

 topics for which we want to store the last mirror 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 mirror 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." }
      ]}
    ]}
  ]
}

MirrorPartitionState

Written on every state transition. Tracks the current state of each mirror partition.

{
  "apiKey": 2,
  "type": "coordinator-key",
  "name": "MirrorPartitionStateKey",
  "validVersions": "0",
  "flexibleVersions": "none",
  "fields": [
    { "name": "MirrorName", "type": "string", "versions": "0", "entityType": "mirrorName",
      "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." },
    { "name": "PreviousState", "type": "int8",  "versions": "0+", "default": 16,
      "about": "The mirror partition state before this transition; UNKNOWN if not recorded." },
    { "name": "RetryAttempt", "type": "int16", "versions": "0+", "default": "0",
      "about": "The number of automatic retry attempts while in FAILED state." }, 
  ]
}

Type Enumerations

This section lists the new values added to existing Kafka type enumerations to support Cluster Mirroring

Config

A CLI/user-facing entity type string used by kafka-configs.sh.

public enum ConfigType {
    // existing types unchanged 
    CLUSTER_MIRRORS("cluster-mirrors");
}

Config Resource

The resource type that represents the cluster mirror configuration in the metadata log. This is used in Admin API requests (DescribeConfigs, IncrementalAlterConfigs, etc.).

public final class ConfigResource
    // ...
    public enum Type {
        // existing types unchanged 
        CLUSTER_MIRROR((byte) 64);
    }

Schema Field

The schema-level annotation for MirrorName string fields in protocol messages. The message generator uses it to validate that mirror name fields across all request/response schemas conform to the expected type.

public enum EntityType {
    // existing types unchanged
    @JsonProperty("mirrorName")
    MIRROR_NAME(FieldType.StringFieldType.INSTANCE);
}

ACL Resource

An ACL resource type that represents a cluster mirror as a securable object. Authorization checks use this type with the mirror name as the resource name.

public enum ResourceType {
    // existing types unchanged
    CLUSTER_MIRROR((byte) 8);

Coordinator

The coordinator type for locating the broker responsible for a given mirror name. The coordinator partition is determined by hashing the mirror name across __mirror_state topic partitions.

public enum CoordinatorType { 
    // existing types unchanged
    CLUSTER_MIRROR((byte) 3);
}

Configuration

This section describes new configurations introduced by the Cluster Mirroring feature.

Broker

Set via broker Set via broker config. Stored in server.properties or dynamic broker config.

Mirror

Set via CreateMirror CreateClusterMirror or IncrementalAlterConfigs. Stored in cluster metadata records.

...

Errors

List of protocol-level errors returned by the new RPCs:

Errors

List of protocol-level errors returned by the new RPCs:

Compatibility,

...

CreateMirror, StartMirrorTopics, StopMirrorTopics, PauseMirrorTopics, ResumeMirrorTopics, DeleteMirror

Compatibility, Deprecation, and Migration Plan

Cluster Mirroring will be introduced through a phased rollout across multiple Kafka releases to ensure stability and gather community feedback. 

Release Phases

Early access

Cluster Mirroring is introduced as an early access feature, disabled by default to prevent accidental production usage. To enable it, all cluster nodes (controllers and brokers) must explicitly enable unstable API versions (unstable.api.versions.enable=true) and unstable feature versions (unstable.feature.versions.enable=true) in all configuration files. After starting the cluster with a minimum metadata version, operators can dynamically enable the mirror version feature to activate Cluster Mirroring (bin/kafka-features.sh --bootstrap-server :9092 upgrade --feature mirror.version=1). This stage is intended for testing and evaluation in non-production environments only, as the new APIs and metadata record formats may change in subsequent releases without backward compatibility guarantees.

Preview

In a future release, Cluster Mirroring will transition to preview status with frozen protocol and metadata schemas. The feature will still require explicit enablement via dynamic feature upgrades but will no longer require the unstable API and feature configuration. The feature remains disabled by default to ensure operators consciously opt-in, but the upgrade path from early access clusters will be officially supported with compatibility guarantees. This stage is suitable for pre-production testing and pilot deployments where API stability is required but production-grade maturity is not yet needed.

General availability

...

Deprecation, and Migration Plan

Cluster Mirroring will be introduced through a phased rollout across multiple Kafka releases to ensure stability and gather community feedback. 

Release Phases

Early Access

Cluster Mirroring is introduced as an early access feature, disabled by default to prevent accidental production usage. To enable it, all cluster nodes (controllers and brokers) must explicitly enable unstable API versions (unstable.api.versions.enable=true) and unstable feature versions (unstable.feature.versions.enable=true) in all configuration files. After starting the cluster with a minimum metadata version, operators can dynamically enable the mirror version feature to activate Cluster Mirroring (bin/kafka-features.sh --bootstrap-server :9092 upgrade --feature mirror.version=1). This stage is intended for testing and evaluation in non-production environments only, as the new APIs and metadata record formats may change in subsequent releases without backward compatibility guarantees.

Preview

In a future release, Cluster Mirroring will transition to preview status with frozen protocol and metadata schemas. The feature will still require explicit enablement via dynamic feature upgrades but will no longer require the unstable API and feature configuration. The feature remains disabled by default to ensure operators consciously opt-in, but the upgrade path from early access clusters will be officially supported with compatibility guarantees. This stage is suitable for pre-production testing and pilot deployments where API stability is required but production-grade maturity is not yet needed.

General Availability

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.

Compatibility Matrix

Note that some features require support from the source cluster.

Additional notes:

1. Sources older than Kafka 2.7 (pre-KIP-595) do not support Fetch API v12+, which introduced the lastFetchedEpoch field for truncation-on-fetch. When mirroring from these sources, the fetcher omits lastFetchedEpoch from fetch requests once the negotiated API version is known. Truncation-on-fetch mode remains enabled to avoid falling back to OffsetsForLeaderEpoch, which does not work cross-cluster. This means divergence detection is effectively disabled for pre-2.7 sources, but data replication works correctly.
2. Sources older than Kafka 2.8 (pre-KIP-516) do not support topic IDs and return ZERO_UUID in metadata responses. In this case, the destination controller assigns a new topic ID, so topic identity is not preserved across clusters. Topic matching falls back to name-based lookup. Partition scaling and topic creation work correctly, but the source and destination will have different topic IDs for the same topic.
3. Share groups were introduced in Kafka 4.0 (KIP-932). When mirroring from older sources, share group offset sync is automatically skipped because the source Admin client does not support the share group listing API. A warning is logged but does not affect other sync operations.

Migration From MirrorMaker 2

...

MM2 and Cluster Mirroring use different internal topic structures and naming conventions for storing metadata and offsets. The two systems track and store consumer offsets differently, making it impossible to seamlessly transition between them.

Follow this process to switch from MirrorMaker 2 to Cluster Mirroring:

1. Stop MM2 replication
2. Delete mirror topics on destination cluster, including MM2 internal topics
3. Start fresh with Cluster Mirroring

Compatibility Matrix

Note that some features require support from the source cluster.

...

Feature

...

Source Cluster Requirement

...

Destination Cluster Requirement

...

Notes

...

Core mirroring and failover

...

2.1

...

4.x

...

Kafka 4 is compatible with old clients versions up to 2.1 included.

...

Failback (reverse mirroring)

...

4.x

...

4.x

...

Requires LME tracking on both sides, otherwise it will fallback and truncate to zero, effectively mirroring from scratch.

...

Share Groups

...

4.x

...

4.y

...

and store consumer offsets differently, making it impossible to seamlessly transition between them.

Follow this process to switch from MirrorMaker 2 to Cluster Mirroring:

1. Stop MM2 replication.
2. Delete mirror topics on destination cluster, including MM2 internal topics.
3. Start fresh with Cluster Mirroring.

Performance Considerations

...

• CLI Workflow: Create mirror with kafka-cluster-mirrors.sh, start mirroring, verify log convergence
• Basic Replication: Create mirror via API, start mirroring, verify log convergence
• Metadata Sync: Modify topic config in source, verify automatic sync to destination
• Partition Expansion: Add partitions to source topic, verify destination expands
• Consumer Groups: Commit offsets in source, verify replication to destination
• ACL Replication: Create ACL in source, verify creation in destination
• Leader Changes: Trigger leader election in source, verify fetcher reconnects
• Broker Failures: Stop destination broker, verify replication continues after recovery

...

System tests will validate behavior under realistic production conditions:

• Log Convergence: Check log convergence after a series of leader elections in source and destination clusters.
• Failover Test: Simulate source cluster failure, measure consumer recovery time.
• Security Validation: Test all authentication mechanisms (SASL PLAIN, SCRAM, Kerberos, mTLS).
• Migration Test: Test migration from older Kafka versionsPerformance Benchmark: Measure replication throughput and latency across WAN.
• Scalability Test: Replicate 1000 topics with 100,000 partitions across clusters.
• Failover Test: Simulate source cluster failure, measure consumer recovery time.
• Long-Running Stability: Run continuous replication for 7 days, verify no memory leaks or performance degradation.
• Security Validation: Test all authentication mechanisms (SASL PLAIN, SCRAM, Kerberos, mTLS) via kafka-mirrors.sh config filesPerformance Benchmark: Measure replication throughput and latency across WAN.

Rejected Alternatives

Keep Using MirrorMaker 2

...