Versions Compared

Old Version 243

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.

...

  • 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
DescribeClusterMirrorsMCReadClusterClusterMirrorLog truncation
ApiVersionsMMM

Feature negotiation
ListOffsetsMFTDescribeTopicOffset bounds discovery
OffsetsForLeaderEpochMFTDescribeTopicLeader epoch validation for truncation

...

Code Block
languagebash
# 9091 (source) -----> 9094 (destination)
# in case of disaster, the operator can failover by running the following command
bin/kafka-cluster-mirrormirrors.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)

...

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

...

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.shRepresents 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 enum TypeConfigType {
    // existing types unchanged 
    MIRROR((byte) 64, "mirrorCLUSTER_MIRRORS("cluster-mirrors");
}

Entity

 A 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.

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

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

Schema Field

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 CreateClusterMirror or IncrementalAlterConfigs. Stored in cluster metadata records.

...

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

...