Versions Compared

Old Version 241

changes.mady.by.user Federico Valeri

Saved on

compared with

New Version Current

changes.mady.by.user Federico Valeri

Saved on

Key

  • This line was added.
  • This line was removed.
  • Formatting was changed.

...

A cluster mirror is a named, unidirectional replication channel from a remote source cluster to the local destination cluster. It is created by specifying a unique mirror name along with the source cluster's bootstrap servers and security credentials. Once a mirror is created, individual topics on the source cluster can be started, stopped, or paused for replication within it. Each mirror is a first-class entity managed through the Admin API and the kafka-cluster-mirrors.sh CLI tool, with its state persisted in a coordinator that manages cross-cluster replication.

Image Modified

The architecture consists of three main components that work together to provide automatic metadata synchronization and data replication. The following diagram illustrates how these components are wired together. For the sake of clarity, some internal APIs are excluded.

Image RemovedImage Added

The mirror name is stored as a topic-level internal configuration called mirror.name that has the same validation rules of topic names, and propagates through Kafka's metadata log as configuration change records. When topics are added to a mirror, the quorum controller generates configuration metadata records that are replicated to all brokers through the standard metadata update mechanism. Brokers monitor these configuration changes to detect when partitions they lead belong to a mirror, triggering the creation of mirror fetchers and enforcement of read-only semantics. This design ensures that mirror associations are visible, auditable, and manageable through standard Kafka tools while maintaining strict control over how mirroring relationships are established and modified.

Main Components

...

ClusterMirrorCoordinator

The ClusterMirrorCoordinator (MCCMC) manages Cluster Mirroring state using a partitioned coordinator partitioned coordinator pattern similar to the group and transaction coordinators.

...

  • State Management: Mirror partition states and control records are stored in __mirror_state internal topic. The coordinator loads these metadata on startup and partition leadership changes.
  • Partition Assignment: Mirror partitions are distributed evenly to coordinators across all brokers and allows for horizontal scaling. The number of coordinator partitions is configurable via via mirror.state.topic.num.partitions.
  • Leader Election: When a broker becomes the leader for a __mirror_state partition, it loads the mirror metadata for all mirrors assigned to that partition and begins coordinating those mirrors. On resignation, it clears its in-memory state to avoid stale metadata.
  • Metadata Refresh Scheduling: The coordinator schedules periodic metadata refresh operations by invoking a metadata manager every 30 seconds by default. This ensures that configuration updates and group offset commits in the source cluster are continuously propagated to the destination cluster. The refresh interval is configurable via mirror.metadata.refresh.interval.ms.
  • State Transitions: The coordinator manages asynchronous state transitions for mirror partitions. Each partition is an independent replication unit with its own state. When the coordinator is the leader for a mirror partition, it writes the state updates directly to __mirror_state topic, otherwise it reads and writes the state via new internal RPCs, enabling distributed coordination across the cluster.

Image RemovedImage Added

State descriptions:

...

RPC

Component

ACL Operation

ACL Resource

Purpose

FetchMFTReadTopicData replication
MetadataMMMDescribeTopicTopic discovery and leader tracking
DescribeConfigsMMMDescribeConfigsTopicTopic configuration sync
ListGroupsMMMDescribeGroupConsumer group offset sync
OffsetFetchMMMDescribeGroupConsumer group offset sync
DescribeAclsMMMDescribeClusterACL synchronization
DescribeMirrorsDescribeClusterMirrorsMCReadClusterClusterMirrorLog truncation when preparing
ApiVersionsMMM

Feature negotiation
ListOffsetsMFTDescribeTopicOffset bounds discovery
OffsetsForLeaderEpochMFTDescribeTopicLeader epoch validation for truncation

...

RPC

Component

ACL Operation

ACL Resource

Purpose

CreateMirrorCreateClusterMirrorControllerCreateClusterMirrorNew cluster mirror creation
StartMirrorTopicsControllerAlterClusterMirrorMirror topics creation
StartMirrorTopicsControllerAlterConfigsTopicMirror topics creation
StopMirrorTopicsControllerAlterClusterMirrorMirror topics removal (failover)
StopMirrorTopicsControllerAlterConfigsTopicMirror topics removal (failover)
PauseMirrorTopicsControllerAlterClusterMirrorMirror topics pause
PauseMirrorTopicsControllerAlterConfigsTopicMirror topics pause
ResumeMirrorTopicsControllerAlterClusterMirrorMirror topics resume
ResumeMirrorTopicsControllerAlterConfigsTopicMirror topics resume
DeleteMirrorDeleteClusterMirrorControllerAlterClusterMirrorDelete a cluster mirror
ListMirrorsListClusterMirrorsBrokerDescribeClusterMirrorMirror topic listing
DescribeMirrorsDescribeClusterMirrorsBrokerDescribeClusterMirrorMirror topic describe (state, lag)
DescribeConfigsBrokerDescribeConfigsClusterMirrorMirror configuration describe
WriteMirrorStatesMCClusterActionClusterMirror partition state write
ReadMirrorStatesMCClusterActionClusterMirror partition state read
BumpLeaderEpochsMCClusterActionClusterLeader epoch bump when stopping
FindCoordinatorBrokerClusterActionClusterMirror coordinator location
CreateTopicsMMMCreateTopicTopic creation
CreatePartitionsMMMAlterTopicPartitions scaling
IncrementalAlterConfigsMMMAlterConfigsClusterMirrorMirror configuration update
OffsetCommitMMMReadTopicSource CG offsets commit
OffsetCommitMMMReadGroupSource CG offsets commit
CreateAclsMMMAlterClusterSource ACLs creation
DeleteAclsMMMAlterClusterSource ACLs removal

...

Integrating with tiered storage would allow mirroring to handle data that has been offloaded to remote storage (e.g., S3, HDFS), enabling full replication of topics with long retention periods without requiring all data to reside in local broker storage. During the PREPARING LOG_TRUNCATION state, the mirror truncates the local log to the LME. This operation does not support tiered storage on the destination cluster because LME may be moved to remote segments.

When tiered storage is enabled locally for a mirror topic, its partitions transition to FAILED state. This limitation applies only to the destination cluster during PREPARING log truncation phase, and will be removed once tiered storage truncation will be fully supported. A detailed design of the metadata synchronization protocol, API schema, and state management will be provided in a follow-up KIP.

...

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

...

Before transitioning a mirror partition from PREPARING to LOG_TRUNCATION to MIRRORING, the MirrorCoordinator must ensure that all in-sync replicas in the destination cluster have truncated their logs to the correct offset. If less than min ISR are available, we will skip and retry in the following fetch. This coordination step validates that every ISR member has completed truncation before the partition is allowed to begin actively fetching from the source cluster. Without it, the mirror leader could start appending new data from the source while local followers still hold divergent log segments, causing inconsistencies within the destination cluster. After truncation, reverse mirroring begins normally. Note that the log truncation on reverse mirroring may cause the data loss if there are records that didn't get mirrored to the old destination cluster.

Code Block
languagebash
# 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)

Create Mirror

  1. The user sends CreateMirror CreateClusterMirror requests to any broker with the mirror name and mirror related properties (bootstrap servers, security settings, etc.).
  2. The broker forwards the request to the active controller.
  3. The controller saves the properties into the metadata log as ConfigRecord entries with type MIRROR.
  4. If this is the first mirror being created, the controller also auto creates the __mirror_state internal topic.
  5. All brokers receive the metadata update and the MirrorMetadataManager registers the new mirror configuration.

...

  1. User sends StartMirrorTopicsRequest with mirror name, topics, and optional include/exclude patterns.
  2. Controller persists include/exclude patterns as ConfigRecord entries on the MIRROR resource in the metadata log.
  3. 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.
  4. Brokers receive the metadata update. The MirrorMetadataManager detects the new mirror.name config (without .stopped or .paused suffix) and queries the coordinator for the current partition state.
  5. Partitions transition from UNKNOWN to PREPARINGLOG_TRUNCATION. During PREPARINGthis state, LME truncation runs and waits for all ISR members (or all replicas if ULE is enabled).
  6. Partitions transition to MIRRORING. A MirrorFetcherThread is created and begins fetching from the source cluster.
  7. Partition state is persisted to __mirror_state.
  8. On subsequent metadata refresh cycles, the MirrorMetadataManager discovers new source topics matching the persisted include/exclude patterns and repeats steps 3-7 for each.

...

Delete Mirror

  1. The user sends a DeleteMirror DeleteClusterMirror request with the mirror name.
  2. The controller validates that the mirror is empty (no topics assigned) or all its partitions are in STOPPED state.
  3. If valid, the controller tombstones the mirror configuration in the cluster metadata log, removing all ConfigRecord entries for the mirror.
  4. The mirror state records in __mirror_state internal topic are also tombstoned.
  5. Any remaining coordinator state is shut down, source cluster connections are closed, and the mirror name becomes available for reuse.
  6. After deletion, failback using this mirror configuration is no longer possible.

List Mirrors

  1. The user sends ListMirrorsRequest ListClusterMirrors request to any broker (no parameters required).

  2. The broker handler gets all configured mirror partitions from, which reads from the in memory metadata cache.
  3. For each authorized mirror, the broker returns: mirror name, source cluster ID, source bootstrap servers, and topic count.
  4. No metadata records are written. This is a read only operation against the local metadata cache.

Describe Mirrors

  1. The user sends DescribeMirrorsRequest DescribeClusterMirrors request with optional mirror names (empty means all mirrors).

  2. The broker handler queries two sources:

    1. The ReplicaManager which provides source offset, destination offset, and lag for each partition.

    2. The MirrorCoordinator which provides the current partition state from the metadata manager cache.

  3. The request is forwarded to each broker that only reports partitions for which it has lag information or is the partition leader. This avoids duplicate reporting across brokers.

  4. For each partition, the response includes: mirror name, topic name, partition ID, source offset, destination offset, lag, current state, and LME.
  5. No metadata records are written. This is a read only operation.

...

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

Code Block
languagebash
$ bin/kafka-cluster-mirrors.sh --help
Create cluster mirrors and manage mirroredmirror 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.

...

Code Block
languagebash
$ echo "bootstrap.servers=localhost:9092" >/tmp/mirror.properties
$ bin/kafka-mirrorcluster-mirrors.sh --bootstrap-server :9094 --create --mirror my-mirror --mirror-config /tmp/mirror.properties
Created mirror my-mirror

...

Code Block
languagebash
$ 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]

...

Code Block
languagebash
$ 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]

...

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

...

Code Block
languagebash
$ 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]

...

Code Block
languagebash
$ 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]

...

Code Block
languagebash
$ 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

...

Code Block
languagebash
$ 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

...

Code Block
languagebash
$ bin/kafka-configs.sh --bootstrap-server :9094 --entity-type cluster-mirrors --entity-name my-mirror \
    --alter --add-config 'bootstrap.servers=localhost:9092'
Completed updating config for mirror my-mirror.

...

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

Protocol Changes

...

Code Block
// new request field in FetchPartition type
{ "name": "MirrorLeaderEpoch", "type": "int32", "versions": "19+", "default": "-1", "taggedVersions": "19+", "tag": 2, "ignorable": true,
  "about": "The latest known mirror leader epoch." }

// new response field in PartitionData type
{ "name": "MirrorLeaderEpoch", "type": "int32", "versions": "19+", "default": "-1", "taggedVersions": "19+", "tag": 3, "ignorable": true,
  "about": "The latest known mirror leader epoch." },

...

CreateClusterMirror

Allows users to create a mirror and supply its configuration. The broker validates that the mirror name is not already in use, contains only permitted characters, and does not end with .stopped or .paused suffix. Once validated, the request is forwarded to the controller, which persists the configuration in the metadata log.

Code Block
{
  "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." }
  ]
}

...

Stop mirroring for the specified topics. The broker validates that all target topic partitions are in either PREPARING LOG_TRUNCATION or MIRRORING state. Once validated, the request is forwarded to the controller, which appends the .stopped suffix to the mirror.name topic config to mark the topics as no longer mirrored.

Code Block
{
  "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+",
 "entityType": "mirrorName",
       "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": "Patterns", "type": "[]string", "versions": "0+",
      "about": "Patterns to update in mirror.topics.include/exclude." }
  ]
}

{
  "apiKey": TBD,
  "type": "response",
  "name": "StopMirrorTopicsResponse",
  // 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": "MirrorName", "type": "string", "versions": "0+", "entityType": "mirrorName",
      "about": "The cluster mirror name." },
    { "name": "Topics", "type": "[]TopicResult", "versions": "0",
      "about": "The results for the topics.", "fields": [
      { "name": "Name", "type": "string", "versions": "0", "entityType": "topicName",
        "about": "The topic name." },
      { "name": "ErrorCode", "type": "int16", "versions": "0",
        "about": "The error code, or 0 if there was no error." }
    ]}
  ]
}

...

Code Block
{
  "apiKey": TBD,
  "type": "request",
  "listeners": ["broker", "controller"],
  "name": "PauseMirrorTopicsRequest",
  // Version 0 is the initial version.
  "validVersions": "0",
  "flexibleVersions": "0+",
  "fields": [
    { "name": "MirrorName", "type": "string", "versions": "0+",
 "entityType": "mirrorName",
       "about": "The mirror name to pause the topics for." },
    { "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." }
      ]}
  ]
}

{
  "apiKey": TBD,
  "type": "response",
  "name": "PauseMirrorTopicsResponse",
  // 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": "MirrorName", "type": "string", "versions": "0+", "entityType": "mirrorName",
      "about": "The cluster mirror name." },
    { "name": "Topics", "type": "[]TopicResult", "versions": "0",
      "about": "The results for the topics.", "fields": [
      { "name": "Name", "type": "string", "versions": "0", "entityType": "topicName",
        "about": "The topic name." },
      { "name": "ErrorCode", "type": "int16", "versions": "0",
        "about": "The error code, or 0 if there was no error." }
    ]}
  ]
}

...

Code Block
{
  "apiKey": TBD,
  "type": "request",
  "listeners": ["broker", "controller"],
  "name": "ResumeMirrorTopicsRequest",
  // 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": "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." }
      ]}
  ]
}

{
  "apiKey": TBD,
  "type": "response",
  "name": "ResumeMirrorTopicsResponse",
  // 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": "MirrorName", "type": "string", "versions": "0+", "entityType": "mirrorName",
      "about": "The cluster mirror name." },
    { "name": "Topics", "type": "[]TopicResult", "versions": "0",
      "about": "The results for the topics.", "fields": [
      { "name": "Name", "type": "string", "versions": "0", "entityType": "topicName",
        "about": "The topic name." },
      { "name": "ErrorCode", "type": "int16", "versions": "0",
        "about": "The error code, or 0 if there was no error." }
    ]}
  ]
}

...

DeleteClusterMirror

Permanently deletes a cluster mirror, including its configuration. The mirror must be empty (no topics) or all its partitions must be in STOPPED state. After deletion, all metadata are tombstoned, making failback impossible. This is an irreversible operation.

Code Block
{
  "apiKey": TBD,
  "type": "request",
  "listeners": ["broker", "controller"],
  "name": "DeleteMirrorRequestDeleteClusterMirrorRequest",
  // Version 0 is the initial version.
  "validVersions": "0",
  "flexibleVersions": "0+",
  "fields": [
    { "name": "MirrorName", "type": "string", "versions": "0+", "entityType": "mirrorName",
      "about": "The cluster mirror name to delete."}
  ]
}

{
  "apiKey": TBD,
  "type": "response",
  "name": "DeleteMirrorResponseDeleteClusterMirrorResponse",
  // 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." }
  ]
}

...

ListClusterMirrors

Returns the current mirror names and their associated topic counts in the cluster. It also includes source cluster ID and bootstrap server.

Code Block
{
  "apiKey": TBD,
  "type": "request",
  "listeners": ["broker"],
  "name": "ListMirrorsRequestListClusterMirrorsRequest",
  // Version 0 is the initial version.
  "validVersions": "0",
  "flexibleVersions": "0+",
  "fields": []
}

{
  "apiKey": TBD,
  "type": "response",
  "name": "ListMirrorsResponseListClusterMirrorsResponse",
  // 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": "[]ListedMirror", "versions": "0+",
      "about": "Each mirror in the response.", "fields": [
      { "name": "MirrorName", "type": "string", "versions": "0+", "entityType": "mirrorName",
        "about": "The cluster mirror name." },
      { "name": "SourceBootstrap", "type": "string", "versions": "0+",
        "about": "The source cluster bootstrap servers." },
      { "name": "SourceClusterId", "type": "string", "versions": "0+", "default": "",
        "about": "The source cluster ID, or empty if not yet resolved." },
      { "name": "TopicCount", "type": "int32", "versions": "0+", "default": "0",
        "about": "The number of topics configured for this mirror. 0 indicates an empty mirror with no topics." }
    ]}
  ]
}

...

DescribeClusterMirrors

Returns the current mirroring status, state, and configuration for the specified mirror topics on the destination cluster. Allows destination cluster partition leaders to query the LME from the source cluster.

Code Block
{
  "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." } 
        ]}
      ]}
    ]}
  ]
}

...

Code Block
{
  "apiKey": 1,
  "type": "coordinator-key",
  "name": "LastMirrorEpochsKey",
  "validVersions": "0",
  "flexibleVersions": "none",
  "fields": [
    { "name": "MirrorName", "type": "string", "versions": "0",
 "entityType": "mirrorName",
       "about": "The cluster mirror name."}
  ]
}

{
  "apiKey": 1,
  "type": "coordinator-value",
  "name": "LastMirrorEpochsValue",
  "validVersions": "0",
  "flexibleVersions": "0+",
  "fields": [
    { "name": "Topics", "type": "[]Topic", "versions": "0+",
      "about": "The mirror topics for which we want to store the last mirror epochs.",  "fields": [
      { "name": "Name", "type": "string", "versions": "0",
        "about": "The topic name." },
      { "name": "Partitions", "type": "[]Partition", "versions": "0+",
        "about": "Each partition to record the last mirror epochs.", "fields": [
        { "name": "PartitionIndex", "type": "int32", "versions": "0+",
          "about": "The partition index." },
        { "name": "", "type": "int32", "versions": "0+",
          "about": "The last mirror leader epoch for this partition." }
      ]}
    ]}
  ]
}

...

Code Block
{
  "apiKey": 2,
  "type": "coordinator-key",
  "name": "MirrorPartitionStateKey",
  "validVersions": "0",
  "flexibleVersions": "none",
  "fields": [
    { "name": "MirrorName", "type": "string", "versions": "0",
       "entityType": "mirrorName",
      "about": "The cluster mirror name."}
  ]
}

{
  "apiKey": 2,
  "type": "coordinator-value",
  "name": "MirrorPartitionStateValue",
  "validVersions": "0",
  "flexibleVersions": "0+",
  "fields": [
    { "name": "TopicName", "type": "string", "versions": "0",
      "about": "The topic name."},
    { "name": "Partition", "type": "int32", "versions": "0",
      "about": "The partition index."},
    { "name": "State", "type": "int8", "versions": "0+",
      "about": "The mirror partition state." },
    { "name": "PreviousState", "type": "int8",  "versions": "0+", "default": 16,
      "about": "The mirror partition state before this transition; UNKNOWN if not recorded." },
    { "name": "RetryAttempt", "type": "int16", "versions": "0+", "default": "0",
      "about": "The number of automatic retry attempts while in FAILED state." }, 
  ]
}

...

This section lists the new values added to existing Kafka type enumerations to support Cluster Mirroring

Configuration

Config

A CLI/user-facing entity type string used by kafka-configs.sh.

Code Block
languagejava
public enum ConfigType {
    // existing types unchanged 
    CLUSTER_MIRRORS("cluster-mirrors");
}

Config Resource

The resource type that represents the cluster mirror configuration in the metadata log. This is used in Admin API requests (DescribeConfigs, IncrementalAlterConfigs, etc.)Represents a mirror as a configurable resource in cluster metadata. Mirror-level properties such as source cluster bootstrap servers, security credentials are stored under this type, keyed by mirror name.

Code Block
languagejava
public final class ConfigResource
    // ...
    public enum Type {
            // existing types unchanged 
        CLUSTER_MIRROR((byte) 64, "mirror");
    }

...

Schema Field

 A The schema-level annotation for MirrorName string fields in protocol messages. The message generator uses it to validate that mirror name fields across all request/response schemas conform to the expected type.

Code Block
languagejava
public enum EntityType {
    // existing types unchanged
    @JsonProperty("mirrorName")
    MIRROR_NAME(FieldType.StringFieldType.INSTANCE);
}

ACL Resource

An ACL resource type that represents a cluster mirror as a securable object. Authorization checks use this type with the mirror name as the resource name.

Code Block
languagejava
public enum ResourceType {
    // existing types unchanged
    CLUSTER_MIRROR((byte) 8);

Coordinator

A The coordinator type for locating the broker responsible for a given mirror name. The coordinator partition is determined by hashing the mirror name across __mirror_state topic partitions.

Code Block
languagejava
public enum CoordinatorType { 
    // existing types unchanged
    CLUSTER_MIRROR((byte) 3);
}

Configuration

This section describes new configurations introduced by the Cluster Mirroring feature.

...

Set via broker config. Stored in server.properties or dynamic broker config.

Key

Description

Default

mirror.state.topic.num.partitions

Number of partitions for __mirror_state internal topic.

50

mirror.state.topic.replication.factor

Replication factor for __mirror_state internal topic. 

3

mirror.num.replica.fetchers

Number of fetcher threads per mirrored source broker,

1

mirror.metadata.refresh.interval.ms

The interval in milliseconds at which the coordinator refreshes metadata from source clusters. This controls how frequently the coordinator polls source clusters to detect new topics and metadata changes.

30000

mirror.replication.throttled.rate

A long representing the upper bound (bytes/sec) on replication traffic for mirrored follower node enumerated in the property “mirror.replication.throttled.replicas” (for each topic). This property can be only set dynamically. It is suggested that the limit be kept above 1MB/s for accurate behaviour.

MAX_LONG

request.timeout.ms

Maximum amount of time in milliseconds the client will wait for the response of a request.

30000

socket.*

Socket connection  configurations.


replica.*

Fetcher threads configurations.


Mirror

Set via CreateMirror CreateClusterMirror or IncrementalAlterConfigs. Stored in cluster metadata records.

...

Name

Type

Group

Tags

Description

JMX Bean

MaxLag

MirrorFetcherManager

kafka.server.mirror

clientId=MirrorReplica

Max lag in messages between destination leader and source leader replicas.

kafka.server.mirror:type=MirrorFetcherManager,name=MaxLag,clientId=MirrorReplica

MinFetchRate

MirrorFetcherManager

kafka.server.mirror

clientId=MirrorReplica

The min fetch rate between destination leader and source leader replicas.

kafka.server.mirror:type=MirrorFetcherManager,name=MirrorReplica

ConsumerLag

FetcherLagMetrics

kafka.server

clientId=MirrorFetcherThread-{sourceBroker.id}-{fetcherId}-{mirrorName},topic=([-.\w]+),partition=([0-9]+)

Lag in messages per remote leader replica.

kafka.serverr:type=FetcherLagMetrics,name=ConsumerLag,clientId=MirrorFetcherThread-{sourceBroker.id}-{fetcherId}-{mirrorName},topic=([-.\w]+),partition=([0-9]+)

DeadThreadCount

MirrorFetcherManager

kafka.server.mirror

clientId=MirrorReplica

Number of dead mirror fetcher threads.

kafka.server,mirror:type=MirrorFetcherManager,name=DeadThreadCount,clientId=MirrorReplica

FailedPartitionsCount

MirrorFetcherManager

kafka.server.mirror

clientId=MirrorReplica

Total count for failed partitions for any reason like auth, authorization, failed network with source.

kafka.serve.mirror:type=MirrorFetcherManager,name=FailedPartitionsCount,clientId=MirrorReplica

BytesPerSec

FetcherStats

kafka.server

clientId=MirrorFetcherThread-{sourceBroker.id}-{fetcherId}-{mirrorName},brokerHost={host},brokerPort={port}

Extend kafka.server.FetcherStats to report mirror fetcher threads.

kafka.server:type=FetcherStats,name=BytesPerSec,clientId=MirrorFetcherThread-{sourceBroker.id}-{fetcherId}-{mirrorName},brokerHost={host},brokerPort={port},mirror-name={mirrorName}

RequestsPerSec

FetcherStats

kafka.server

clientId=MirrorFetcherThread-{sourceBroker.id}-{fetcherId}-{mirrorName},brokerHost={host},brokerPort={port}

Extend kafka.server.FetcherStats to report mirror fetcher threads.

kafka.server:type=FetcherStats,name=RequestsPerSec,cclientId=MirrorFetcherThread-{sourceBroker.id}-{fetcherId}-{mirrorName}, brokerHost={host},brokerPort={port},mirror-name={mirrorName}



LocalTimeMs,

MessageConversionsTimeMs,

RemoteTimeMs,

RequestBytes,

RequestQueueTimeMs,

ResponseQueueTimeMs,

ResponseSendTimeMs,

TemporaryMemoryBytes,

TotalTimeMs

RequestMetrics

kafka.network

request=[mirror_requests]

Extend kafka.network:type=RequestMetrics to list cluster mirror requests.

kafka.network:type=RequestMetrics,name=*, request=*

ErrorsPerSec

RequestMetrics

kafka.network

request=[mirror_requests],error=*

Extend kafka.network:type=RequestMetrics to list cluster mirror requests.

kafka.network:type=RequestMetrics,name=ErrorsPerSec, request=*, error=*

RequestsPerSec

RequestMetrics

kafka.network

request=[mirror_requests],version=*

Extend kafka.network:type=RequestMetrics to list cluster mirror requests.

kafka.network:type=RequestMetrics,name=RequestsPerSec, request=*, version=*

connection-close-rate,

connection-close-total,

connection-count,

connection-creation-rate,

connection-creation-total,

failed-authentication-rate,

failed-authentication-total,

failed-reauthentication-rate,

failed-reauthentication-total,

incoming-byte-rate,

incoming-byte-total,

network-io-rate,

network-io-total,

outgoing-byte-rate,

outgoing-byte-total,

reauthentication-latency-avg,

reauthentication-latency-max,

request-rate,

request-size-avg,

request-size-max,

request-total,

response-rate,

response-total,

select-rate,

select-total,

successful-authentication-no-

reauth-total,

successful-authentication-rate,

successful-authentication-total,

successful-reauthentication-rate,

successful-reauthentication-total

mirror-broker-{DestinationBroker.id}-fetcher-{fetcherId}-mirror-{mirrorName}-metrics

kafka.server

broker-id={sourceBroker.id},fetcher-id={fetcherId}

Fetcher requests in the cluster mirror metrics.

kafka.server:type=mirror-broker-{sourceBroker.id}-fetcher-{fetcherId}-mirror-{mirrorName}-metrics,broker-id={sourceBroker.id},fetcher-id={fetcherId}

MetadataRefreshError

MirrorMetadataManager

kafka.server.mirror


Number of topic metadata refresh sync errors.

kafka.server.mirror:type=MirrorMetadataManager,name=aclSyncError

TopicConfigMetadataSyncError

MirrorMetadataManager

kafka.server.mirror


Number of topic configuration sync errors.


ConsumerGroupOffsetSyncError

MirrorMetadataManager

kafka.server.mirror


Number of CGs sync errors.


ShareGroupOffsetSyncError

MirrorMetadataManager

kafka.server.mirror


Number of SGs sync errors.


AclSyncError

MirrorMetadataManager

kafka.server.mirror


Number of ACLs sync errors.

kafka.server.mirror:type=MirrorMetadataManager,name=aclSyncError

ByteRate

MirrorReplication

kafka.server


Bandwidth quota metrics. Indicates the throttled data mirror replication rate of the broker in bytes/sec.

kafka.server:type=MirrorReplication

FailedPartitionState

MirrorMetadataManager

kafka.server.mirror


Number of partitions in failed state.

kafka.server.mirror:type=MirrorMetadataManager,name=FailedPartitionState

StoppedPartitionState

MirrorMetadataManager

kafka.server.mirror


Number of partitions in a stopped state.

kafka.server.mirror:type=MirrorMetadataManager,name=StoppedPartitionState

StoppingPartitionState

MirrorMetadataManager

kafka.server.mirror


Number of partitions in stopping state.

kafka.server.mirror:type=MirrorMetadataManager,name=StoppingPartitionState

MirroringPartitionState

MirrorMetadataManager

kafka.server.mirror


Number of partitions in mirroring state.

kafka.server.mirror:type=MirrorMetadataManager,name=MirroringPartitionState

LogTruncationPartitionState

MirrorMetadataManager

kafka.server.mirror


Number of partitions in log truncation state.

kafka.server.mirror:type=MirrorMetadataManager,name=LogTruncationPartitionState

EpochFencingPartitionStatePreparingPartitionState

MirrorMetadataManager

kafka.server.mirror


Number of partitions in preparing epoch fencing state.

kafka.server.mirror:type=MirrorMetadataManager,name=PreparingPartitionStateEpochFencingPartitionState

Errors

List of protocol-level errors returned by the new RPCs:

CodeNameMessageUsed By
3UNKNOWN_TOPIC_OR_PARTITIONThe topic does not exist on the target clusterStopMirrorTopics, PauseMirrorTopics, ResumeMirrorTopics
15COORDINATOR_NOT_AVAILABLEThe mirror coordinator is not activeWriteMirrorStates, ReadMirrorStates
31CLUSTER_AUTHORIZATION_FAILEDThe client is not authorized to perform the mirror operation

WriteMirrorStates, ReadMirrorStates

35UNSUPPORTED_VERSIONCluster mirroring is disabled (mirror.version=0)

CreateMirrorCreateClusterMirror, StartMirrorTopics, StopMirrorTopics, PauseMirrorTopics, ResumeMirrorTopics, ListMirrorsListClusterMirrors, DescribeMirrorsDescribeClusterMirrors, DeleteMirrorDeleteClusterMirror

TBDREAD_ONLY_TOPICThe topic is read-only because it is a mirror topic on the target clusterProduce
TBDINVALID_CLUSTER_MIRROR_NAMEThe cluster mirror name does not meet the naming rulesCreateMirrorCreateClusterMirror
TBDCLUSTER_MIRROR_ALREADY_EXISTSThe cluster mirror already existsCreateMirrorCreateClusterMirror
TBDUNKNOWN_CLUSTER_MIRRORThe topic is not assigned to any cluster mirrorStopMirrorTopics, PauseMirrorTopics, ResumeMirrorTopics
TBDTOPIC_ALREADY_IN_CLUSTER_MIRRORThe topic is already assigned to a cluster mirrorStartMirrorTopics
TBDTOPIC_NOT_IN_CLUSTER_MIRRORThe topic does not belong to the specified cluster mirrorStopMirrorTopics, PauseMirrorTopics, ResumeMirrorTopics
TBDMIRROR_TOPIC_ALREADY_PAUSEDThe mirror topic is already pausedPauseMirrorTopics
TBDMIRROR_TOPIC_NOT_PAUSEDThe mirror topic is not pausedResumeMirrorTopics
TBDMIRROR_TOPIC_BEING_STOPPEDThe mirror topic is being stoppedResumeMirrorTopics
TBDCLUSTER_MIRROR_NOT_EMPTYThe cluster mirror still has active or non-removed topicsDeleteMirrorDeleteClusterMirror
TBDCLUSTER_MIRROR_AUTHORIZATION_FAILEDMirror Cluster mirror authorization failed

CreateMirrorCreateClusterMirror, StartMirrorTopics, StopMirrorTopics, PauseMirrorTopics, ResumeMirrorTopics, DeleteMirrorDeleteClusterMirror

Compatibility, Deprecation, and Migration Plan

...

Note that some features require support from the source cluster.

Feature

Source Cluster Requirement

Destination Cluster Requirement

Notes

Core mirroring and failover

2.1

4.x

Kafka 4 is compatible with old clients versions up to 2.1 included.

Failback (reverse mirroring)

4.x

4.x

Requires LME tracking on both sides, otherwise it will fallback and truncate to zero, effectively mirroring from scratch.

Share Groups

4.x

4.y

If the source doesn't support share groups, mirroring continues but share group offsets won't be synchronized.

Additional notes:

  1. Sources older than Kafka 2.7 (pre-KIP-595) do not support Fetch API v12+, which introduced the lastFetchedEpoch field for truncation-on-fetch. When mirroring from these sources, the fetcher omits lastFetchedEpoch from fetch requests once the negotiated API version is known. Truncation-on-fetch mode remains enabled to avoid falling back to OffsetsForLeaderEpoch, which does not work cross-cluster. This means divergence detection is effectively disabled for pre-2.7 sources, but data replication works correctly.
  2. Sources older than Kafka 2.8 (pre-KIP-516) do not support topic IDs and return ZERO_UUID in metadata responses. In this case, the destination controller assigns a new topic ID, so topic identity is not preserved across clusters. Topic matching falls back to name-based lookup. Partition scaling and topic creation work correctly, but the source and destination will have different topic IDs for the same topic.
  3. Share groups were introduced in Kafka 4.0 (KIP-932). When mirroring from older sources, share group offset sync is automatically skipped because the source Admin client does not support the share group listing API. A warning is logged but does not affect other sync operations.

Migration From MirrorMaker 2

...

  • CLI Workflow: Create mirror with kafka-cluster-mirrors.sh, start mirroring, verify log convergence
  • Basic Replication: Create mirror via API, start mirroring, verify log convergence
  • Metadata Sync: Modify topic config in source, verify automatic sync to destination
  • Partition Expansion: Add partitions to source topic, verify destination expands
  • Consumer Groups: Commit offsets in source, verify replication to destination
  • ACL Replication: Create ACL in source, verify creation in destination
  • Leader Changes: Trigger leader election in source, verify fetcher reconnects
  • Broker Failures: Stop destination broker, verify replication continues after recovery

...

  • Log Convergence: Check log convergence after a series of leader elections in source and destination clusters.
  • Failover Test: Simulate source cluster failure, measure consumer recovery time.
  • Security Validation: Test all authentication mechanisms (SASL PLAIN, SCRAM, Kerberos, mTLS) via kafka-mirrors.sh config files.
  • Migration Test: Test migration from older Kafka versions.
  • Scalability Test: Replicate 1000 topics with 100,000 partitions across clusters.
  • Long-Running Stability: Run continuous replication for 7 days, verify no memory leaks or performance degradation.
  • Performance Benchmark: Measure replication throughput and latency across WAN.

...