...

Cluster Mirroring supports comprehensive security controls through both authorization and authentication mechanisms. On the destination cluster, mirror-related operations (creating mirrors, adding/removing topics from mirrors, managing mirror configurations) require the {{CLUSTER_ACTION}} permission on the cluster resource.  This ensures that only authorized principals can establish and manage cluster mirrors. When configuring a mirror, operators specify operators specify ACLs that should be synchronized from the source cluster, and these ACLs are periodically replicated to the destination cluster to maintain consistent access control policies across both environments.

For connecting to the source cluster, Cluster Mirroring requires only the bootstrap server address and appropriate credentials, no other sensitive cluster information is exposed or required. The destination cluster's mirror configuration supports all standard Kafka authentication mechanisms including TLS/SSL for encrypted transport and SASL for client authentication. Each mirror can be configured with its own security settings, allowing different mirrors to connect to source clusters with varying security requirements. This enables secure cross-cluster replication even when source and destination clusters use different authentication protocols or when connecting across security boundaries such as on-premises to cloud environments. All credentials are stored in the destination cluster's mirror configuration and used exclusively for establishing authenticated connections to the source cluster.

Idempotent Producer

The idempotent producers rely on producer IDs to detect duplicate writes and ensure idempotent production. To avoid conflicts with the destination cluster's producer ID space, we rewrite source producer IDs to occupy the unused negative space by applying the formula: 

destinationProducerId = -(sourceProducerId + 2)

The rationale of this formula is to keep the existing semantic of NO_PRODUCER_ID (-1) but still have a way to avoid the conflict. The CRC checksum is automatically recalculated after the producer ID changes to maintain batch integrity. Producer epochs from the source cluster are preserved exactly as they appear in the source batches. This ensures the last stable offset is correctly reflected because the producer state is updated after each append.

When a mirror topic becomes writable during failover, records with transformed producer IDs (<= -2) remain in the log with their original sequence numbers and epochs. Applications that reconnect to the destination cluster receive new producer IDs (>=0) from the destination's transaction coordinator, allowing them to continue producing.

Exactly-Once Semantics

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

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

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

Transactional Consumer Guarantees

Kafka consumers with isolation.level=read_committed determine transaction visibility using only the Last Stable Offset (LSO), which is computed from COMMIT/ABORT control markers in the log. Consumers never interact with the transaction coordinator or validate producer IDs. This separation between log-level markers (replicated) and coordinator state (not replicated) is why transactional consumers work correctly on mirror topics without mirroring coordinator state. The LSO truncation during failover ensures all remaining transactions have mirrored markers, maintaining this guarantee.

Example Scenario

Consider this source cluster log:

If replication reaches offset 4 and the source cluster fails, the destination cluster contains data records for transaction 4002 (offset 2) without the abort marker (offset 4). This creates a hanging transaction that can never be committed or aborted on the destination cluster.

Note that this approach causes data loss for any in-flight transactions or non-mirrored completed transactions when we experiencing a lag during the failover and may result in already-processed records being lost if consumers on the destination cluster read uncommitted data.

Bandwidth Control

Cluster Mirroring adopts a dual-sided throttling mechanism that extends Kafka's existing bandwidth control capabilities to work across cluster boundaries.

1. Destination Cluster Throttling: To avoid conflicts with intra-cluster replication controls, mirror-specific throttling configurations operate independently from standard replication throttling. The system provides two configuration levels: a broker-level rate limit (mirror.replication.throttled.rate) that sets the overall bandwidth ceiling for mirror replication traffic, and a topic-level replica list (mirror.replication.throttled.replicas) that specifies which partition-broker combinations should be throttled using the standard partition-index:broker-id notation. Operators can dynamically adjust throttling rates at runtime without restarting brokers, first setting a cluster-wide default rate, then fine-tuning specific topic partitions as mirroring progresses. This allows gradual bandwidth allocation as mirror relationships are established.
2. Source Cluster Throttling: The source cluster side requires a different approach because mirror fetch requests operate as consumer traffic rather than replication traffic. This design is intentional since the mirroring must fetch only up to the LSO to maintain transactional consistency, which is a consumer-level guarantee not available through the replication protocol. Consequently, standard leader replication throttling mechanisms cannot apply to mirror traffic. Instead, the source cluster leverages Kafka's client quota system. Each mirror fetcher thread presents itself with a deterministic client identifier that encodes the broker ID, fetcher thread number, and mirror name. Operators can apply per-client byte rate quotas to these identifiers, effectively throttling the outbound mirror traffic from the source cluster. This approach integrates seamlessly with Kafka's existing quota enforcement infrastructure.

Tiered Storage

Tiered Storage is not initially supported, but a detailed design of the metadata synchronization protocol, API schema, and state management will be provided in a follow-up KIP. Before this is supported, the destination cluster will treat all the data as in local storage, and mirrors every record, regardless of where it was stored originally.

Share Group

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

Share consumer groups use a different offset management model based on Share-Partition Start Offset (SPSO) and Share-Partition End Offset (SPEO) rather than traditional committed offsets. First we retrieve the current SPSO for each share group using the DescribeShareGroupOffsets API from the source cluster, and then we update the SPSO in the destination cluster using the AlterShareGroupOffsets API, which also initializes the group state in both the group coordinator and share coordinator. This means the API can initialize a share group in the destination cluster even if it doesn't exist yet, eliminating the need for pre-creation or complex state management.

Kafka enforces that consumer group and share group names must be unique within a single cluster. This creates a potential conflict scenario during mirroring. When such conflicts occur, the offset commit operation will fail with GroupIdNotFoundException. Users must resolve these conflicts manually by either deleting the conflicting group in the destination cluster before mirroring begins, or excluding the conflicting groups from offset synchronization. These conflicts affect only offset synchronization and do not impact data mirroring itself. The topic data continues to replicate normally, and only the automatic offset synchronization for the conflicting groups is blocked.

Diskless Topics

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

Active-Active Writes

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

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.

Public Interfaces

Command-Line

A new dump flag allows to decode cluster mirroring metadata for debugging purpose:

$ 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}}

Source cluster permissions (mirror principal):

Destination cluster permissions:

An operator can grant ClusterMirror:*:CREATE,ALTER,DESCRIBE for full mirror management, or scope it to specific mirrors like ClusterMirror:prod-dr:DESCRIBE for read-only monitoring of a single mirror, without granting any broker-level privileges.

MMM issues CreatePartitions, OffsetCommit, CreateAcls, DeleteAcls locally using the inter-broker principal, bypassing normal ACL checks. This is by design but means the mirror feature implicitly holds ALTER on topics, groups, and ACLs within the destination cluster.

Idempotent Producer

The idempotent producers rely on producer IDs to detect duplicate writes and ensure idempotent production. To avoid conflicts with the destination cluster's producer ID space, we rewrite source producer IDs to occupy the unused negative space by applying the formula: 

destinationProducerId = -(sourceProducerId + 2)

The rationale of this formula is to keep the existing semantic of NO_PRODUCER_ID (-1) but still have a way to avoid the conflict. The CRC checksum is automatically recalculated after the producer ID changes to maintain batch integrity. Producer epochs from the source cluster are preserved exactly as they appear in the source batches. This ensures the last stable offset is correctly reflected because the producer state is updated after each append.

When a mirror topic becomes writable during failover, records with transformed producer IDs (<= -2) remain in the log with their original sequence numbers and epochs. Applications that reconnect to the destination cluster receive new producer IDs (>=0) from the destination's transaction coordinator, allowing them to continue producing.

Exactly-Once Semantics

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

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

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

Transactional Consumer Guarantees

Kafka consumers with isolation.level=read_committed determine transaction visibility using only the Last Stable Offset (LSO), which is computed from COMMIT/ABORT control markers in the log. Consumers never interact with the transaction coordinator or validate producer IDs. This separation between log-level markers (replicated) and coordinator state (not replicated) is why transactional consumers work correctly on mirror topics without mirroring coordinator state. The LSO truncation during failover ensures all remaining transactions have mirrored markers, maintaining this guarantee.

Example Scenario

Consider this source cluster log:

If replication reaches offset 4 and the source cluster fails, the destination cluster contains data records for transaction 4002 (offset 2) without the abort marker (offset 4). This creates a hanging transaction that can never be committed or aborted on the destination cluster.

Note that this approach causes data loss for any in-flight transactions or non-mirrored completed transactions when we experiencing a lag during the failover and may result in already-processed records being lost if consumers on the destination cluster read uncommitted data.

Bandwidth Control

Cluster Mirroring adopts a dual-sided throttling mechanism that extends Kafka's existing bandwidth control capabilities to work across cluster boundaries.

1. Destination Cluster Throttling: To avoid conflicts with intra-cluster replication controls, mirror-specific throttling configurations operate independently from standard replication throttling. The system provides two configuration levels: a broker-level rate limit (mirror.replication.throttled.rate) that sets the overall bandwidth ceiling for mirror replication traffic, and a topic-level replica list (mirror.replication.throttled.replicas) that specifies which partition-broker combinations should be throttled using the standard partition-index:broker-id notation. Operators can dynamically adjust throttling rates at runtime without restarting brokers, first setting a cluster-wide default rate, then fine-tuning specific topic partitions as mirroring progresses. This allows gradual bandwidth allocation as mirror relationships are established.
2. Source Cluster Throttling: The source cluster side requires a different approach because mirror fetch requests operate as consumer traffic rather than replication traffic. This design is intentional since the mirroring must fetch only up to the LSO to maintain transactional consistency, which is a consumer-level guarantee not available through the replication protocol. Consequently, standard leader replication throttling mechanisms cannot apply to mirror traffic. Instead, the source cluster leverages Kafka's client quota system. Each mirror fetcher thread presents itself with a deterministic client identifier that encodes the broker ID, fetcher thread number, and mirror name. Operators can apply per-client byte rate quotas to these identifiers, effectively throttling the outbound mirror traffic from the source cluster. This approach integrates seamlessly with Kafka's existing quota enforcement infrastructure.

Tiered Storage

Tiered Storage is not initially supported, but a detailed design of the metadata synchronization protocol, API schema, and state management will be provided in a follow-up KIP. Before this is supported, the destination cluster will treat all the data as in local storage, and mirrors every record, regardless of where it was stored originally.

Share Group

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

Share consumer groups use a different offset management model based on Share-Partition Start Offset (SPSO) and Share-Partition End Offset (SPEO) rather than traditional committed offsets. First we retrieve the current SPSO for each share group using the DescribeShareGroupOffsets API from the source cluster, and then we update the SPSO in the destination cluster using the AlterShareGroupOffsets API, which also initializes the group state in both the group coordinator and share coordinator. This means the API can initialize a share group in the destination cluster even if it doesn't exist yet, eliminating the need for pre-creation or complex state management.

Kafka enforces that consumer group and share group names must be unique within a single cluster. This creates a potential conflict scenario during mirroring. When such conflicts occur, the offset commit operation will fail with GroupIdNotFoundException. Users must resolve these conflicts manually by either deleting the conflicting group in the destination cluster before mirroring begins, or excluding the conflicting groups from offset synchronization. These conflicts affect only offset synchronization and do not impact data mirroring itself. The topic data continues to replicate normally, and only the automatic offset synchronization for the conflicting groups is blocked.

Diskless Topics

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

Active-Active Writes

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

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.

Public Interfaces

Command-Line

A new dump flag allows to decode cluster mirroring metadata for debugging purpose:

$ 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 mirrors:

$ bin/kafka-mirrors.sh --help
This tool helps to create cluster mirrors and add topics to them.
Option                                  Description                           
------                                  -----------                           
--add                                   Add topic(s) to an existing cluster   
                                          mirror (supports regex).            
--alter                                 Alter the configuration of an existing
                                          cluster mirror.                     
--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                                Create a new cluster mirror from a    
                                          source cluster.                     
--describe                              Describe a cluster mirror including   
                                          partition lag and state.            
--help                                  Print usage information.              
--list                                  List all cluster mirrors.             
--mirror <String: mirror>               The name of the cluster mirror.       
--mirror-config <String: mirror config  Property file containing source       
  property file>                          cluster configs for mirroring.      
--pause                                 Pause mirroring for topic(s) matching 
                                          the pattern (supports regex).       
--remove                                Remove topic(s) from an existing      
                                          cluster mirror (supports regex).    
--replication-factor <Short:            The replication factor to use for the 
  replication-factor>                     mirror topic. If not specified, uses
                                          the destination cluster's default.  
--resume                                Resume mirroring for previously paused
                                          topic(s) matching the pattern       
                                          (supports regex).                   
--topic <String: topic>                 Topic name or regex pattern to match  
                                          topics (e.g., 'my-topic' or 'test-. 
                                          *').                                
--version                               Display Kafka version.

Create a new cluster mirror in the destination cluster (forbidden suffixes: .removed, .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

Add a topic or set of topics to an existing cluster mirror (start mirroring; the topic flag accepts regex expression):

$ bin/kafka-mirror.sh --bootstrap-server :9094 --add --topic my-topic --mirror my-mirror
Added 1 topic(s) to mirror my-mirror: [my-topic]

Remove a specific topic or set of topics from a mirror (failover; topics become writable)A new command-line tool kafka-mirrors.sh provides administrative operations for managing cluster mirrors:

$ bin/kafka-mirrorsmirror.sh --bootstrap-server :9094 --remove --topic my-topic --mirror my-mirror
Removed 1 topic(s) from mirror my-mirror: [my-topic]

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

TODO

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]

List configured mirrors with additional information:

$ bin/kafka-mirrors.sh --bootstrap-server :9094 --list
MIRRORhelp
This tool helps to create cluster mirrors and add topics to them.
Option                                  Description                           
------                                  -----------                           
--add                                   Add topic(s) to an existing cluster   
          TOPICS     CLUSTER-ID                 BOOTSTRAP-SERVER
my-mirror          mirror (supports regex).          2  
--alter        lBq12jYZRp-9wF3M9MPopg     localhost:9091,localhost:9092
new-mirror                    Alter the1 configuration of an existing
      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  cluster mirror.                     
--bootstrap-server <String: server to   REQUIRED: The destination Kafka server
  connectPARTITION to> SOURCE-OFFSET   DESTINATION-OFFSET LAG      STATE         
my-mirror         to connect to.                bar      
--command-config <String: command       Property file containing configs to be
  config property file>              0     passed to Admin Client.  2324           
--create 2324               0        MIRRORING   
my-mirror     Create a new cluster mirror from a    
       foo                                   source cluster.  0          69         
--describe     66                 3        DescribeMIRRORING a cluster 
my-mirror including   
                  foo                        partition lag and state.           1 
--help         94              84           Print usage information.    10       MIRRORING   
my--listmirror                      foo            List all cluster mirrors.             
--mirror <String: mirror>        2       The name of the94 cluster mirror.       
--mirror-config <String: mirror config  Property90 file containing source       
  property file>    4        MIRRORING   
new-mirror           cluster configs for mirroring.      
--pause baz                                Pause mirroring for topic(s) matching 
 0          189             189                0  the pattern (supports regex).    MIRRORING   
--removenew-mirror                     baz              Remove topic(s) from an existing      
              1          859             859     cluster mirror (supports regex).    
--replication-factor <Short:   0        MIRRORING

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)`: The replication factor to use for the 
  replication-factor>                     mirror topic. If not specified, uses
                                          the destination cluster's default.  
--resume                                Resume mirroring for previously paused
                                          topic(s) matching the pattern       
                                          (supports regex).                   
--topic <String: topic>                 Topic name or regex pattern to match  
                                          topics (e.g., 'my-topic' or 'test-. 
                                          *').                                
--version    
      (principal=User:mirror-admin, host=*, operation=CREATE, permissionType=ALLOW)                  Display Kafka version.

Create a new cluster mirror in the destination cluster (forbidden suffixes: .removed, .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

Add a topic or set of topics to an existing cluster mirror (start mirroring; the topic flag accepts regex expression):

$ bin/kafka-mirror.sh --bootstrap-server :9094 --add --topic my-topic --mirror my-mirror
Added 1 topic(s) to mirror my-mirror: [my-topic]

Remove a specific topic or set of topics from a mirror (failover; topics become writable):

$ bin/kafka-mirror.sh --bootstrap-server :9094 --remove --topic my-topic --mirror my-mirror
Removed 1 topic(s) from mirror my-mirror: [my-topic]

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

TODO

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
MIRROR                                                                                                       TOPICS     CLUSTER-ID                 BOOTSTRAP-SERVER
my-mirror                      2          lBq12jYZRp-9wF3M9MPopg     localhost:9091,localhost:9092
new-mirror                    
 1     (principal=User:mirror-admin, host=*, operation=ALTER, permissionType=ALLOW)  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                                    PARTITION  SOURCE-OFFSET   DESTINATION-OFFSET LAG      STATE       
my-mirror                      bar                                      0          2324            2324               0        MIRRORING   
my-mirror      (principal=User:mirror-admin, host=*, operation=DESCRIBE, permissionType=ALLOW)             foo                                      0          69              66                 3        MIRRORING   
my-mirror                      foo                                      1          94              84         
      (principal=User:mirror-admin,  10 host=*, operation=DELETE, permissionType=ALLOW)      MIRRORING   
my-mirror                      foo                                      2          94              90                 4        MIRRORING   
new-mirror                     baz                                      0          189        
     189     (principal=User:mirror-admin, host=*, operation=ALTER_CONFIGS, permissionType=ALLOW)            0        MIRRORING   
new-mirror                     baz                                      1          859             859                0        MIRRORING

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

...

{
  "apiKey": TBD,
  "type": "response",
  "name": "DescribeMirrorsResponse",
  // Version 0 is the initial version.
  "validVersions": "0",
  "flexibleVersions": "0+",
  "fields": [
    { "name": "ThrottleTimeMs", "type": "int32", "versions": "0+",
      "about": "The duration in milliseconds for which the request was throttled due to a quota violation, or zero if the request did not violate any quota." },
    { "name": "ErrorCode", "type": "int16", "versions": "0+",
      "about": "The error code, or 0 if there was no error." },
    { "name": "ErrorMessage", "type": "string", "versions": "0+", "nullableVersions": "0+", "default": "null",
      "about": "The top-level error message, or null if there was no error." },
    { "name": "Mirrors", "type": "[]DescribedMirror", "versions": "0+",
      "about": "Each described mirror.", "fields": [
      { "name": "ErrorCode", "type": "int16", "versions": "0+",
        "about": "The error code, or 0 if there was no error." },
      { "name": "MirrorName", "type": "string", "versions": "0+", "entityType": "mirrorName",
        "about": "The cluster mirror name." },
      { "name": "Topics", "type": "[]TopicPartitions", "versions": "0+",
        "about": "Each topic in the mirror.", "fields": [
        { "name": "TopicName", "type": "string", "versions": "0+",
          "about": "The topic name." },
        { "name": "Partitions", "type": "[]PartitionDetail", "versions": "0+",
          "about": "Each partition detail.", "fields": [
          { "name": "PartitionIndex", "type": "int32", "versions": "0+",
            "about": "The partition index." },
          { "name": "SourceOffset", "type": "int64", "versions": "0+", "default": "-1",
            "about": "The high watermark offset from the source cluster leader, or -1 if not yet available." },
          { "name": "DestinationOffset", "type": "int64", "versions": "0+", "default": "-1",
            "about": "The log end offset on the destination cluster, or -1 if not yet available." },
          { "name": "Lag", "type": "int64", "versions": "0+", "default": "-1",
            "about": "The lag (source offset - destination offset), or -1 if not yet available." },
          { "name": "State", "type": "string", "versions": "0+",
            "about": "The partition state (INITIALIZING, PREPARING, MIRRORING, STOPPING, STOPPED, FAILED)." }
        ]}
      ]},
      { "name": "AuthorizedOperations", "type": "int32", "versions": "0+", "default": "-2147483648",
        "about": "32-bit bitfield to represent authorized operations for this mirror." }
    ]}
  ]
}

LastMirroredOffset

...