Page History

# 9091 (source) -----> 9094 (destination)
# in case of disaster, the operator can failover by running the following command
bin/kafka-mirror.sh --bootstrap-server :9094 --removestop --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)

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

$ bin/kafka-mirrors.sh --help
This tool helps to create cluster mirrors and addmanage mirrored 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.                     
--describedelete                                DescribeDelete a cluster mirror. including   
          
--describe                              Describe a partitioncluster lagmirror andincluding state.  
          
--help                                partition lag Printand usage informationstate.              
--listhelp                                  ListPrint allusage clusterinformation. mirrors.             
--mirrorjson   <String: mirror>               The name of the cluster mirror.       
--mirror-config <String: mirror config  Property file containing source     Output description in JSON format     
--list     
  property file>                          clusterList configsall forcluster mirroringmirrors.      
--pause       
--mirror <String: mirror>               The name of the cluster mirror.    Pause mirroring for topic(s) matching 
       
--mirror-config <String: mirror config  Property file containing source       
  property file>                          thecluster patternconfigs (supportsfor regex)mirroring.       
--removepause                                 Pause mirroring Removefor topic(s) from an existing      matching 
                                          clusterthe mirrorpattern (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).                   
--start                                 Start mirroring topic(s) in an        
                                          existing cluster mirror (supports   
                                          regex).                             
--stop                                  Stop mirroring topic(s) in an existing
                              (supports regex).           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.

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

/**
 * 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
 * Add@param topicsconfigs toConfiguration anfor existingthe 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 cross-clusterspecified replicationtopics.
 *
 * When topics are addedstarted toin 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 to add the topics to
 * @param topics Set of topic names to add tostart mirroring
  * @param options Options for the add topics tostart mirror topics operation
 * @return The AddTopicsToMirrorResultStartMirrorTopicsResult containing futures for each topic addition
 */
AddTopicsToMirrorResultStartMirrorTopicsResult addTopicsToMirrorstartMirrorTopics(String mirrorName, Set<String> topics, AddTopicsToMirrorOptionsStartMirrorTopicsOptions options);

/**
 * RemoveStop topics from cluster mirror, making them writable on the destination clustermirroring 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 Removingmirror topics from
 * the mirror clears the mirrorName field from partition metadata, which allows producers to write
 * to these partitions.
 *
 * @param mirrorName The cluster mirror name to remove the topics from
 
 * @param topics Set of topic names to removestop from mirroring
 * @param options Options for the stop removemirror topics from mirror operation
 * @return The RemoveTopicsFromMirrorResultStopMirrorTopicsResult containing futures for each topic removal
 */
RemoveTopicsFromMirrorResultStopMirrorTopicsResult removeTopicsFromMirrorstopMirrorTopics(String mirrorName, Set<String> topics, RemoveTopicsFromMirrorOptionsStopMirrorTopicsOptions options);

/**
 * 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 to pause the topics for
 
 * @param topics Set of topic names to pause mirroring for
 * @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 to resume the topics for
 * @param topics Set of topic names to resume mirroring for
 * @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 name of the cluster mirror to deletename
  * @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);