Page History

...

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.

...

We use a composite key (mirror name, topic id, and partition number) to distribute coordination work across the __cluster_mirror_state topic's partitions, which is the internal compacted topic used to store mirror metadata. Each mirror partition independently hashes to a coordinator, spreading the load across all brokers in the cluster. This means that a mirror with hundreds of partitions will have its state management distributed evenly across the cluster rather than concentrated on a single broker.

...

• State Management: Mirror partition states and control records are stored in __cluster_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 __cluster_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 _ _cluster_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:

...

• Last Mirror Epoch (LME): The greatest leader epoch of a given partition that a destination cluster recognizes from the source cluster and stores in the __cluster_mirror_state internal topic. It represents the synchronization point between source and destination. After the log is truncated to LME, the destination cluster does not contain any record with a leader epoch beyond the LME. Only the source cluster owns leader epochs exceeding the LME. This ensures the source leader epoch remains the source of truth, even when epoch histories diverge across clusters during asynchronous mirroring.
• Leader Epoch Bump (LEB): The leader epoch in the destination cluster remains unchanged during replication. That means it is possible that the fetched batch carries a leader epoch that does not match with the local leader epoch. To ensure the leader epoch remains monotonically increasing, it is incremented when a partition becomes writable after failover. New records produced on the cluster will then carry an epoch higher than anything in the existing log.

...

...

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

...

1. The user sends 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 __cluster_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 LOG_TRUNCATION. During this 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 __cluster_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.

...

1. User sends StopMirrorTopics request with mirror name, topics, and optional patterns.
2. If patterns are provided, the controller removes matching entries from mirror.topics.include and adds them to mirror.topics.exclude on the MIRROR resource in the metadata log. Any currently mirroring topic that matches the updated exclude is also stopped.
3. For each topic, the controller writes a ConfigRecord updating mirror.name=<mirrorName>.stopped on the TOPIC resource.
4. Brokers receive the metadata update. The MirrorMetadataManager detects the .stopped suffix and transitions partitions to STOPPING.
5. During STOPPING, the following operations execute sequentially:
   
   1. Fetcher threads are removed for the affected partitions stopping replication.
   2. The current leader epoch is collected and persisted as LME in _ _cluster_mirror_state.
   3. The partition's leader epoch is bumped to draw a boundary between mirrored and locally produced records.

   4. ABORT markers are appended for all ongoing transactions using the new leader epoch.

   5. A MIRROR_PID_RESET control record is written to expire all producer state entries.

6. Partitions transition to STOPPED and becomes writable.

...

1. User sends PauseMirrorTopics request with topics and mirror name.
2. The controller validates each topic belongs to the specified mirror and is currently in MIRRORING state. It updates the mirror name to mirror.name=<mirrorName>.paused, generating a ConfigRecord.
3. When the MirrorMetadataManager in the partition leader node gets notified, it detects the .paused suffix and transitions the state to PAUSING.
4. During PAUSING, the MirrorFetcherManager removes the fetcher threads for the affected partitions. No more data is replicated.
5. The state transitions from PAUSING to PAUSED. The partition remains read only. Metadata synchronization (configs, groups, ACLs) is also paused for these topics.
6. The partition state change is persisted to the __cluster_mirror_state topic.

Resume Mirror Topics

...

1. The user sends a 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 __cluster_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.

...

$ bin/kafka-dump-log.sh --mirror-state-decoder --files /tmp/server*/data/__cluster_mirror_state-1/00000000000000000000.log
Dumping /home/fvaleri/Documents/kafka/build/test/server4/data/__cluster_mirror_state-0/00000000000000000000.log
Log starting offset: 0
baseOffset: 0 lastOffset: 0 count: 1 baseSequence: -1 lastSequence: -1 producerId: -1 producerEpoch: -1 partitionLeaderEpoch: 0 isTransactional: false isControl: false deleteHorizonMs: OptionalLong.empty position: 0 CreateTime: 1771239071191 size: 98 magic: 2 compresscodec: none crc: 3314855113 isvalid: true
| offset: 0 CreateTime: 1771239071191 keySize: 13 valueSize: 17 sequence: -1 headerKeys: [] key: {"type":"2","data":{"mirrorName":"my-mirror"}} payload: {"version":"0","data":{"topicName":"my-topic","partition":0,"state":0}}
baseOffset: 1 lastOffset: 1 count: 1 baseSequence: -1 lastSequence: -1 producerId: -1 producerEpoch: -1 partitionLeaderEpoch: 0 isTransactional: false isControl: false deleteHorizonMs: OptionalLong.empty position: 98 CreateTime: 1771239071219 size: 98 magic: 2 compresscodec: none crc: 3968746657 isvalid: true
| offset: 1 CreateTime: 1771239071219 keySize: 13 valueSize: 17 sequence: -1 headerKeys: [] key: {"type":"2","data":{"mirrorName":"my-mirror"}} payload: {"version":"0","data":{"topicName":"my-topic","partition":0,"state":1}}

...

$ bin/kafka-console-consumer.sh --bootstrap-server :9094 --topic __cluster_mirror_state --from-beginning \
  --formatter org.apache.kafka.tools.consumer.MirrorStateMessageFormatter
{"key":{"type":2,"data":{"mirrorName":"my-mirror"}},"value":{"version":0,"data":{"topicName":"my-topic","partition":0,"state":0}}}
{"key":{"type":2,"data":{"mirrorName":"my-mirror"}},"value":{"version":0,"data":{"topicName":"my-topic","partition":0,"state":2}}}

...

$ echo "bootstrap.servers=localhost:9092" >/tmp/mirror.properties
$ bin/kafka-cluster-mirrormirrors.sh --bootstrap-server :9094 --create --mirror my-mirror --mirror-config /tmp/mirror.properties
Created mirror my-mirror

...

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

...

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

...

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

...

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

...

Internal API that reads the current mirror partition states from the internal __cluster_mirror_state topic on the destination cluster.

...

Internal API that persists mirror partition state transitions to the internal __cluster_mirror_state topic on the destination cluster.

...

This section describes records written to the _ _cluster_mirror_state internal topic by the MirrorCoordinator to track mirror state and synchronization points across brokers.

...

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

...

{
  "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

Config

...

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

...

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
language java
public enum TypeConfigType {     // existing types unchanged      CLUSTER_MIRROR((byte) 64, "mirrorMIRRORS("cluster-mirrors"); }

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

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

...

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 _ _cluster_mirror_state topic partitions.

...

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

_cluster_cluster

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.

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

...