# 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)

$ bin/kafka-cluster-mirrors.sh --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.                      
--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.                     
--delete                                Delete a cluster mirror.              
--describe                              Describe a cluster mirror including   
                                          partition lag and state.            
--exclude <String: exclude patterns>    Comma-separated list of topic names or
                                          regex patterns to exclude from      
                                          mirroring. Only valid with --start. 
--help                                  Print usage information.              
--json                                  Output description in JSON format     
--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 topics matching   
                                          the given patterns.                 
--resume                                Resume mirroring for previously paused
                                          topics matching the given patterns. 
--start                                 Start mirroring topics matching the   
                                          given patterns.                     
--stop                                  Stop mirroring 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 Kafka version.

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

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

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

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

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

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

$ bin/kafka-cluster-mirrors.sh --bootstrap-server :9094 --describe
MIRROR                         TOPIC                                    PARTITION  SOURCE-OFFSET   DESTINATION-OFFSET LAG      STATE       
my-mirror                      bar                                      0          -               -                  -        STOPPED   
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          -               -                  -        PAUSED   
new-mirror                     baz                                      1          -               -                  -        PAUSED

/**
 * 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 CreateMirrorResultCreateClusterMirrorResult
 */
CreateMirrorResultCreateClusterMirrorResult createMirrorcreateClusterMirror(String mirrorName, Map<String, String> configs, CreateMirrorOptionsCreateClusterMirrorOptions 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 CreateMirrorResultCreateClusterMirrorResult
 */
CreateMirrorResultCreateClusterMirrorResult createMirrorcreateClusterMirror(String mirrorName, Map<String, String> configs, CreateMirrorOptionsCreateClusterMirrorOptions 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) {
        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 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);

/**
 * Options for {@link Admin#stopMirrorTopics(String, Set, StopMirrorTopicsOptions)}.
 */
public class StopMirrorTopicsOptions extends 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 DeleteMirrorResultDeleteClusterMirrorResult
 */
DeleteMirrorResultDeleteClusterMirrorResult deleteMirrordeleteClusterMirror(String mirrorName, DeleteMirrorOptionsDeleteClusterMirrorOptions options);

/**
 * List the cluster mirrors available in the cluster.
 *
 * @param options The options to use when listing the mirrors.
 * @return The ListMirrorsResultListClusterMirrorsResult.
 */
ListMirrorsResultListClusterMirrorsResult listMirrorslistClusterMirrors(ListMirrorsOptionsListClusterMirrorsOptions 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 DescribeMirrorsResultDescribeClusterMirrorsResult
 */
DescribeMirrorsResultDescribeClusterMirrorsResult describeMirrorsdescribeClusterMirrors(Collection<String> mirrorNames, DescribeMirrorsOptionsDescribeClusterMirrorsOptions options);

{
  "apiKey": TBD,
  "type": "request",
  "listeners": ["broker", "controller"],
  "name": "CreateMirrorRequestCreateClusterMirrorRequest",
  // 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": "[]MirrorConfigClusterMirrorConfig", "versions": "0+",
      "about": "The cluster mirror configurations.",  "fields": [
      { "name": "Name", "type": "string", "versions": "0+", "mapKey": true,
        "about": "The configuration key name." },
      { "name": "Value", "type": "string", "versions": "0+", "nullableVersions": "0+",
        "about": "The value to set for the configuration key."}
    ]}
  ]
}

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

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

{
  "apiKey": TBD,
  "type": "response",
  "name": "DescribeMirrorsResponseDescribeClusterMirrorsResponse",
  // 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": "AuthorizedOperations", "type": "int32", "versions": "0+", "default": "-2147483648",
        "about": "32-bit bitfield to represent authorized operations for this mirror." },
      { "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." },
          { "name": "", "type": "int32", "versions": "0+", "default": "-1",
            "about": "The last mirror leader epoch, or -1 if not available." } 
        ]}
      ]}
    ]}
  ]
}