Versions Compared

Key

  • This line was added.
  • This line was removed.
  • Formatting was changed.
Comment: Add usage information for kafka-mirrors.sh

...

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

Code Block
languagebash
$ 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 configuration in the destination cluster:

...

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

Code Block
languagebash
$ 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                      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             189                0        MIRRORING   
new-mirror                     baz                                      1          859             859                0        MIRRORING

...

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

Code Block
languagejava
    /**
     * 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);

    /**
     * Add topics to an existing cluster mirror for cross-cluster replication.
     *
     * When topics are added to 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 topicToMirrorName Map of topic names to mirror names, allowing multiple topics to be
     *                          added to potentially different mirrors in a single operation
     * @param options Options for the add topics to mirror operation
     * @return The AddTopicsToMirrorResult containing futures for each topic addition
     */
    AddTopicsToMirrorResult addTopicsToMirror(Map<String, String> topicToMirrorName, AddTopicsToMirrorOptions options);

    /**
     * Remove topics from cluster mirror, making them writable on the destination cluster.
     *
     * 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). Removing topics from
     * the mirror clears the mirrorName field from partition metadata, which allows producers to write
     * to these partitions.
     *
     * @param topics Set of topic names to remove from mirroring
     * @param options Options for the remove topics from mirror operation
     * @return The RemoveTopicsFromMirrorResult containing futures for each topic removal
     */
    RemoveTopicsFromMirrorResult removeTopicsFromMirror(Set<String> topics, RemoveTopicsFromMirrorOptions 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 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(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 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(Set<String> topics, ResumeMirrorTopicsOptions 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);

...