Page History

...

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

...

Start Mirror Topics

1. User sends StartMirrorTopics request StartMirrorTopicsRequest with mirror name, topics and mirror name.

2. The broker forwards to the active controller.

3. The controller validates that each topic exists and is not already in a mirror. It then sets the topic config mirror.name=<mirrorName> for each topic, generating a ConfigRecord per topic into the metadata log.
4. Response is sent back to clients with per topic results.
5. , and optional include/exclude patterns.
6. Controller persists include/exclude patterns as ConfigRecord entries on the MIRROR resource in the metadata log.
7. 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.
8. Brokers receive the metadata update. The MirrorMetadataManager detects the new mirror.name config (without When the MirrorMetadataManager in the partition leader node gets notified about the topic config update, it detects that mirror.name is not empty and has no .stopped or .paused suffix. It then ) and queries the coordinator for the current mirror partition state from the coordinator. The coordinator could be located on a different broker node, so a ReadMirrorStates inter broker RPC may be needed.Based on the current mirror partition state, the state machine transitions the partition. In most cases,
9. Partitions transition from UNKNOWN to PREPARING. During PREPARING, the mirror fetcher performs Last Mirror Epoch ( LME ) truncation. The LME is the greatest leader epoch that the source cluster recognizes from the destination. If the source has no LME knowledge (first time mirroring), it returns -1 and the destination truncates everything and replicates from scratch. Otherwise, the destination truncates at the start offset of the first epoch beyond the LME. It then waits until truncation runs and waits for all ISR members (or all replicas if mirror.support.unclean.leader.election=true) complete the truncation.ULE is enabled).
10. Partitions transition Once all ISR members have completed truncation, the state transitions from PREPARING to MIRRORING. A MirrorFetcherThread is created and starts sending consumer Fetch requests (not follower requests) to the source cluster to replicate data. The Fetch protocol handles any offset level divergence by truncating to the exact offset where the source epoch ends. The fetched batch retains its original leader epoch from the source.begins fetching from the source cluster.
11. Partition The partition state is persisted to the __mirror_state topic on each state change via local append or WriteMirrorStates (when coordinator is remote) as MirrorPartitionStateKey/MirrorPartitionStateValue records, distributed by hash(mirrorName, topicId, partition) % numPartitions..
12. On subsequent metadata refresh cycles, the MirrorMetadataManager discovers new source topics matching the persisted include/exclude patterns and repeats steps 3-7 for eachThe MirrorMetadataManager also periodically synchronizes topic configs, consumer group offsets, and ACLs from the source cluster.

Stop Mirror Topics

1. User sends StopMirrorTopics request StopMirrorTopicsRequest with mirror name, topics, and mirror nameoptional patterns.
2. The controller validates each topic belongs to the specified mirror and is in MIRRORING state. It then updates the topic config by appending the .stopped suffix, e.g. mirror.name=my-mirror.stopped, generating a ConfigRecord.
3. 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.
4. For each topic, the controller writes a ConfigRecord updating mirror.name=<mirrorName>.stopped on the TOPIC resource.
5. Brokers receive the metadata update. The MirrorMetadataManager detects the .stopped suffix and transitions partitions When the MirrorMetadataManager gets notified, it detects the .stopped suffix on mirror.name. It queries the current mirror partition state from the coordinator, and transitions the mirror partition to STOPPING.
6. During STOPPING, the following operations execute sequentially:
   
   1. The MirrorFetcherManager removes all fetcher threads Fetcher threads are removed for the affected partitions, stopping replication.
   2. The LME is recorded as LastMirrorEpochsKey/LastMirrorEpochsValue records into the __mirror_state topic for potential future failback.
   3. The partitions leader epoch is bumped for the partitions to ensure monotonically increasing epochs for new records.

   4. ABORT markers are appended for all ongoing transactions. For each partition, ProducerStateManager provides the set of in-flight transaction entries and an EndTransactionMarker(ABORT) is appended for each one. This resolves hanging transactions without truncating committed data.

   5. A MIRROR_PID_RESET control record is written to the each partition log, which expires all ProducerStateManager entries so that new producers get fresh PIDs with (no collision risk of collisions).

7. The state transitions from STOPPING Partitions transition to STOPPED. The read-only flag is cleared and the topic becomes writable. New producers can start producing with fresh PIDs starting at sequence 0 and a higher leader epochis cleared and the topic becomes writable on the destination.

Pause Mirror Topics

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

...

$ bin/kafka-mirrors.sh --mirrors.sh --help
This tool helps to create cluster mirrors and manage mirrored topics.
Option help
Create cluster mirrors and manage mirrored topics.
Option                                  Description                           
------                                  -----------                                               
--bootstrap-server <String: server to   REQUIRED: The destination Kafka server
  connect to>                             to connect to.  Description                    
--command-config <String: command       
------    Property file containing configs to be
  config property file>                   passed to Admin Client.        -----------     
--create                             
--alter   Create a new cluster mirror from a    
                    Alter the configuration of an existing
                 source cluster.                     
--delete   cluster mirror.                     
--bootstrap-server <String: server to   REQUIRED: TheDelete destinationa Kafkacluster server
mirror.  connect to>           
--describe                  to connect to.          Describe a cluster mirror including   
     
--command-config <String: command       Property file containing configs to be
  config property file>                   passedpartition tolag Adminand Client.state.            
--exclude <String: exclude patterns>     
Comma--createseparated list of topic names or
                           Create a new cluster mirror from a    
     regex patterns to exclude from      
                           source cluster.              mirroring. Only valid    with --start. 
--deletehelp                                 Delete aPrint clusterusage mirrorinformation.              
--describejson                              Describe a cluster mirror includingOutput description in 
JSON format     
--list                                  List all partition lag and state.cluster mirrors.             
--helpmirror <String: mirror>               The name of the cluster mirror.       
--mirror-config <String: mirror config  PrintProperty usagefile information.containing source             
--json  property file>                          cluster configs for mirroring.  Output description in JSON format     
--listpause                                 Pause Listmirroring allfor clustertopics mirrors.matching   
          
--mirror <String: mirror>               The name of the cluster mirror.       
--mirror-config <String: mirror configthe given Property filepatterns. containing source       
  property file>     
--resume                     cluster configs for mirroring.      
--pause  Resume mirroring for previously paused
                           Pause  mirroring for topic(s) matching 
         topics matching the given patterns. 
--start                            the pattern (supports regex).  Start mirroring topics matching 
--resumethe   
                             Resume mirroring for previously paused
         given patterns.                     
--stop           topic(s) matching the pattern       
             Stop mirroring topics matching the    
                     (supports regex).                   
--start given patterns.                     
--topics <String: topics>        Start mirroring topic(s) in an   Comma-separated list of topic names or
                                          regex existing cluster mirror (supportspatterns (e.g., 'my-topic,    
                                          regexorders-.*,payments').               
--version              
--stop                 Display 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 \   Stop mirroring topic(s) in an existing
                                          cluster mirror (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: .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-mirror.sh --bootstrap-server :9094 --start --topic my-topic            
  --topics 'orders-.*' --exclude 'orders-internal' --mirror my-mirror
Started mirroring2 formirror 1 topic(s) in mirror my-mirror: [my-topicorders-us, orders-eu]

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

$ bin/kafka-mirrormirrors.sh --bootstrap-server :9094 --stop --topictopics my'orders-topicus' --mirror my-mirror
Stopped mirroring for 1 topic(s) in mirror my-mirror: [myorders-topicus]

Delete a mirror including its configuration (the mirror must be empty or include only stopped 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-        MIRRORINGSTOPPED   
my-mirror                      foo                                      0          69              66                 3        MIRRORING   
my-mirror                      foo                                      1          94              84                 10       MIRRORING   
my-mirror                      foo                                      2          94              90                 4        MIRRORING   
new-mirror                     baz                                      0          -   189              -                  -        PAUSED   
new-mirror               189      baz          0        MIRRORING   
new-mirror                 1    baz      -               -                  -        PAUSED

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

$ bin/kafka-configs.sh --bootstrap-server :9094 --entity-type mirrors --entity-name my-mirror \
 1   --alter --add-config 'bootstrap.servers=localhost:9092'
Completed updating config for  859             859                0        MIRRORINGmirror 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 clusterAlter mirror configuration (any valid configuration triggers a reconnection):

$ bin/kafka-configs.sh --bootstrap-server :90949091 --entity-type mirrorsalter --entityadd-name my-mirrorconfig 'consumer_byte_rate=1024' \
  --entity-type clients --entity-altername broker--add-config bootstrap.servers=localhost:90924-fetcher-0-mirror-my-mirror
Completed updating config for client mirror broker-4-fetcher-0-mirror-my-mirror.

Throttling on the destination clusterGrant mirror admin full access to a specific mirror:

$ bin/kafka-configsacls.sh --bootstrap-server :9094 --add \
  --cluster-mirror my-mirror \
  --operation Create --operation Alter --operation Describe --operation Delete \
  --entity-typeoperation brokersAlterConfigs --entity-nameoperation 4DescribeConfigs \
  --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.

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 mirrorsGrant 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-adminmonitor
Adding ACLs for resource `ResourcePattern(resourceType=CLUSTER_MIRROR, name=my-mirror*, patternType=LITERAL)`:
      (principal=User:mirror-adminmonitor, host=*, operation=CREATEDESCRIBE, permissionType=ALLOW)
      (principal=User:mirror-adminmonitor, host=*, operation=ALTERDESCRIBE_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,  (principal=User:mirror-admin, host=*, operation=DESCRIBE, permissionType=ALLOW)
name=my-mirror, patternType=LITERAL)`:                                                                                                                     (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:monitormirror-admin, host=*, operation=DESCRIBE_CONFIGSCREATE, 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=User:mirror-admin, host=*, operation=CREATEALTER, permissionType=ALLOW)                                                                                                                                                                                                       
      (principal=User:mirror-admin, host=*, operation=ALTERDESCRIBE, permissionType=ALLOW)                                                                                                                                                                                                       
      (principal=User:mirror-admin, host=*, operation=DESCRIBEDELETE, permissionType=ALLOW)                                                                                                                                                                                                      
      (principal=User:mirror-admin, host=*, operation=ALTER_CONFIGS, permissionType=ALLOW)                                                                                 
      (principal=User:mirror-admin, host=*, operation=DELETE, permissionType=ALLOW)                                                                                                     
      (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:

/**
 * 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 (principal=User:mirror-admin, host=*, operation=ALTER_CONFIGS, permissionType=ALLOW)                                                                                                                                                             
 * @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 startMirrorTopics(String mirrorName, Set<String> topics, StartMirrorTopicsOptions options);

/**
 * Options for {@link Admin#startMirrorTopics(String, Set, StartMirrorTopicsOptions)}.
 */
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)    {
      (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:

/**
 * 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); 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;
    }
}

/**
 * StartStop mirroring for the specified topics.
 *
 * WhenThis topicsoperation areis startedtypically inused a mirror, they become read-only onduring failover scenarios when the destination cluster andneeds startto
 * replicatingbe datapromoted from the source cluster. This operation marks the specified topics with the
 * mirror name, preventing local writes and enabling the MirrorFetcherThread to begin replicationpassive (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 startstop mirroring
 * @param options Options for the startstop mirror topics operation
 * @return The StartMirrorTopicsResultStopMirrorTopicsResult containing futures for each topic
 */
StartMirrorTopicsResultStopMirrorTopicsResult startMirrorTopicsstopMirrorTopics(String mirrorName, Set<String> topics, StartMirrorTopicsOptionsStopMirrorTopicsOptions options);

/**
 * StopOptions mirroringfor for the specified topics{@link Admin#stopMirrorTopics(String, Set, StopMirrorTopicsOptions)}.
 */
public *class ThisStopMirrorTopicsOptions operationextends 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 cluster mirror name
 * @param topics Set of topic names to stop mirroring
 * @param options Options for the stop mirror topics operation
 * @return The StopMirrorTopicsResult containing futures for each topic
 */
StopMirrorTopicsResult stopMirrorTopics(String mirrorName, Set<String> topics, StopMirrorTopicsOptions options);AbstractOptions<StopMirrorTopicsOptions> {
    private List<String> patterns = List.of();

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

    public List<String> patterns() {
        return 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);

/**
 * 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 DeleteMirrorResult
 */
DeleteMirrorResult deleteMirror(String mirrorName, DeleteMirrorOptions options);

/**
 * List the cluster mirrors available in the cluster.
 *
 * @param options The options to use when listing the mirrors.
 * @return The ListMirrorsResult.
 */
ListMirrorsResult listMirrors(ListMirrorsOptions options);

/**
 * Describe cluster mirrors.
 *
 * This operation retrieves detailed information about cluster mirrors including:
 * - Topics being mirrored
 * - Partition-level lag information (source offset vs destination offset)
 * - Mirroring state for each partition (INITIALIZING, PREPARING, MIRRORING, etc.)
 *
 * @param mirrorNames The names of the mirrors to describe
 * @param options The options to use when describing mirrors
 * @return The DescribeMirrorsResult
 */
DescribeMirrorsResult describeMirrors(Collection<String> mirrorNames, DescribeMirrorsOptions options);

...

StartMirrorTopicsRequest

{
  "apiKey": TBD,
  {
  "apiKey": TBD,
  "type": "request",
  "listeners": ["broker", "controller"],
  "name": "StartMirrorTopicsRequest",
  // Version 0 is the initial version.
  "validVersions": "0",
  "flexibleVersions": "0+",
  "fields": [
    { "name": "MirrorName", "type": "requeststring",
  "versions": "0+", "listenersentityType": ["brokermirrorName",
 "controller"]     "about": "The cluster mirror name." },
    { "name": "StartMirrorTopicsRequestTopics",
  // Version 0 is the initial version.
  "validVersions": "0",
  "flexibleVersions"type": "[]TopicData", "versions": "0+", "about": "The data for the topics.",
      "fields": [
        { "name": "TopicId", "type": "uuid", "versions": "0+",
  "fieldsabout": [ "The unique topic ID."},
        { "name": "MirrorNameTopicName", "type": "string", "versions": "0+", "mapKey": true, "entityType": "mirrorNametopicName",
          "about": "The cluster mirrortopic name." },
        { "name": "TopicsNumPartitions", "type": "[]TopicDataint32", "versions": "0+",
          "about": "The number of datapartitions for the topicstopic.",
 Must match the source  "fieldstopic.": [}
      ]},
    { "name": "TopicIdIncludePatterns", "type": "uuid[]string", "versions": "0+",
      "about": "The unique topic ID."Regex patterns to add to mirror.topics.include." },
        { "name": "TopicNameExcludePatterns", "type": "[]string", "versions": "0+", "mapKey": true, "entityType": "topicName",
          "about": "TheRegex topicpatterns name." }
      ]to add to mirror.topics.exclude." }
  ]
}

StartMirrorTopicsResponse

...

{
  "apiKey": TBD,
  "type": "request",
  "listeners": ["broker", "controller"],
  "name": "StopMirrorTopicsRequest",
  // Version 0 is the initial version.
  "validVersions": "0",
  "flexibleVersions": "0+",
  "fields": [
    { "name": "MirrorName", "type": "string", "versions": "0+",
      "about": "The cluster mirror name." },
    { "name": "Topics", "type": "[]TopicData", "versions": "0+", "about": "The data for the topics.",
      "fields": [
        { "name": "TopicId", "type": "uuid", "versions": "0+", "about": "The unique topic ID."},
        { "name": "TopicName", "type": "string", "versions": "0+", "mapKey": true, "entityType": "topicName",
          "about": "The topic name." }
      ]},
    { "name": "TopicNamePatterns", "type": "[]string", "versions": "0+", "mapKey": true, "entityType": "topicName",
          "about": "The topic namePatterns to update in mirror.topics.include/exclude." }
      ]}
  ]
}

StopMirrorTopicsResponse

...

...