DUE TO SPAM, SIGN-UP IS DISABLED. Goto Selfserve wiki signup and request an account.
...
A new command-line tool kafka-mirrors.sh provides administrative operations for managing cluster mirrors:
| Code Block | ||
|---|---|---|
| ||
$ 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 | ||
|---|---|---|
| ||
$ 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 | ||
|---|---|---|
| ||
/** * 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); |
...