...

Kafka deployments often require replicating data across geographically distributed clusters for disaster recovery (DR), regulatory compliance, data locality, cluster migrations or active-active architectures. While MirrorMaker 2 .0 (MM2) provides cross-cluster replication capabilities, it presents significant operational challenges.

...

While Cluster Mirroring is optimized for geo-replication, disaster recovery DR and migration use cases where a single source cluster replicates to one or more destination clusters, its coordinator-based architecture provides a foundation for more complex topologies.

Non-Goals

Synchronous

...

Replication

This proposal describes asynchronous replication between clusters. Support for synchronous replication is deferred to future work.

...

Asynchronous replication should provide the right balance for disaster recovery DR use cases where availability and performance of the primary cluster must not be compromised by cross-datacenter latency. Applications requiring zero data loss across cluster failures can wait for the follow-up KIP that will extend this design to support synchronous mirroring, or handle the lag using application-level caching.

Stretched clusters are not suitable for disaster recovery DR scenarios because they provide no protection against software failures or configuration incidents. Vendors that recommend stretched cluster deployments typically position them for high availability (HA) rather than DR, and notably, most do not offer stretched clusters as a managed service option, further underscoring the operational challenges and limited DR effectiveness of this architecture.

...

In normal Kafka operation, once a record is committed (, part of the high watermark), it is immutable and will never be changed or removed. When a new leader is elected, followers use the epoch information to determine which records are safe to keep and which must be truncated to align with the new leader's log. Replicas eventually converge to the same data through epoch-based reconciliation. Unclean leader elections break this guarantee by allowing non-ISR brokers to become leaders, potentially with fewer records than were previously committed.

...

The mirror name is stored as a topic-level configuration (called mirror.name) that propagates through Kafka's metadata log as configuration change records. When topics are added to a mirror via the addTopicsToMirror API, the controller generates configuration metadata records that are replicated to all brokers through the standard metadata update mechanism.

...

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

We use a composite key of (mirror name, topic id, and partition number) to distribute coordination work across the __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 a mirror with hundreds of partitions will have its state management distributed evenly rather than concentrated on a single broker.

...

• State Management: Mirror configuration and partition states are stored in the internal topic. The coordinator loads the state on startup and partition leadership changes.
• Partition Assignment: Cluster mirrors are assigned to coordinator partitions using consistent hashing based on the mirror name. This distributes coordinator load 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.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 topology changes, configuration updates, and 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 the internal topic. Remote brokers read and write partition state via new RPCs, enabling distributed coordination across the cluster. Both local and remote state updates trigger callbacks to execute appropriate actions for each state.

Figure 3: Mirror Partition Lifecycle.

...

• UNKNOWN: The partition has no cached state (broker just became leader, state not loaded yet). Not an explicit API-driven state, just the absence of state.
• PREPARING: The coordinator for this partition detects via onMetadataUpdate that it leads a mirror partition. It fetches last mirrored offsets from the source cluster and truncate logs to align the local log with the source. Valid from: null, UNKNOWN, STOPPED, FAILED.
• MIRRORING: All ISR members have completed truncation. A MirrorFetcherThread is started to continuously replicate records from the source cluster. Valid from: PREPARING only.
• STOPPING: Triggered by RemoveTopicsFromMirror API (user wants to fail over) or topic deletion on the source. The system records the last mirrored offset to the internal topic. Valid from: PREPARING, MIRRORING.
• STOPPED: Last mirrored offsets have been persisted. The topic becomes writable on the destination cluster (the mirror fetcher is removed and the read-only flag is cleared). Valid from: STOPPING only.
• FAILED: An error occurred. Can be entered from any state. Can transition back to to PREPARING to retry. Valid from: any state.
• (TODO: pause/unpause)

...

Starting a mirror (UNKNOWN -> PREPARING -> MIRRORING): The addTopicsToMirror command sets mirror.name config via the controller. The metadata update propagates to brokers. The broker leading the partition finds out the partition state via the coordinator, and this might trigger readMirrorState RPC to query from the remote coordinator and transitions to PREPARING if it’s it's in a valid transition state (e.g. UNKNOWN). After truncation completes, it moves to MIRRORING and starts the mirror fetcher to fetch data from the source cluster.

Failover (MIRRORING -> STOPPING -> STOPPED): The removeTopicsFromMirror command appends the ".removed” suffix in to the mirror.name config. The partition leader  detects the stop request, transitions to STOPPING, persists the last offset, then moves to STOPPED. The topic is now writable after the STOPPED state.

...

A mirror leader partition begins fetching with an unknown source leader epoch. When it sends Fetch requests to the source cluster, the source leader may respond with a FencedLeaderEpochException. When such an error occurs, the mirror fetcher extracts the current source leader epoch from the error response, or from a separate metadata request if the source cluster doesn’t support fetch API v12, and updates its internal fetch state to track the source cluster's actual leader epoch.   The last fetched epoch is always set to empty to disable log divergence checks due to unclean leader election (see non-goals section).

...

When users remove a topic from the mirror, the partition will be removed from the fetch fetcher thread, and any late fetch responses will be skipped because the partition is not registered anymore in the fetcher thread.

...

Failover is initiated by calling the RemoveTopicsFromMirror API, which appends a ".removed" suffix into to the mirror.name internal config. This transitions the mirror topics from read-only to writable state after the stopping process completes gracefully.

...

For each partition, we track the high watermark (HW) by storing it in the cluster metadata as Last Mirrored Offset last mirrored offset (LMO) when removing a topic from a mirror (failover phase). The LMO represents the last record successfully mirrored from the original source cluster to the destination cluster before failover.

When reverse mirroring failback is initiated on the old source cluster, it needs to determine where to truncate its log before starting to fetch from the new source cluster. If the new API is supported, the broker sends a LastMirrorredOffsets request to the new source cluster asking for the latest mirrored offsetLMO, and then truncates its local log to the returned offset. If the new API is not supported, the broker truncates to zero and starts mirroring from scratch.

Before transitioning a mirror partition from PREPARING to MIRRORING, the MirrorCoordinator must ensure that all in-sync replicas (ISR) 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 the reverse mirroring may cause the data loss for the records that didn’t get mirrored to the old destination cluster earlier.

...

The mirror leader replicates these compacted log segments exactly as they exist in the source cluster, maintaining the same offset assignments and gaps. After failover, when the mirrored mirror topic becomes writable, log compaction continues normally in the destination cluster according to the topic's compaction policy, and any new records produced locally will fill in after the highest mirrored offset.

...

If a mirror follower attempts to fetch from an offset below the source cluster's log start offset (e.g. , fetching offset 50 when log start offset is 100), the source broker returns an OffsetOutOfRangeException. The mirror leader handles this by truncating its local log to the source's current log start offset and resuming fetching from that point. This ensures the destination cluster mirrors the current retention state of the source cluster without attempting to replicate already-deleted data.

...

The idempotent producers rely on producer IDs to detect duplicate writes and ensure idempotent production. To avoid conflicts with the destination cluster's producer ID space, we rewrite source producer IDs to occupy the unused negative space by applying the formula: 

destinationProducerId = -(sourceProducerId + 2)

The rationale of this formula is to keep the existing semantic of NO_PRODUCER_ID (-1) but still have a way to avoid the conflict. The CRC checksum is automatically recalculated after the producer ID changes to maintain batch integrity. Producer epochs from the source cluster are preserved exactly as they appear in the source batches. This ensures the last stable offset is correctly reflected because the producer state is updated after each append.

...

Cluster Mirroring ensures transactional consistency when stopping by truncating to the LSOthe LSO. Note that this doesn’t mean it supports exactly-once semantics (EOS) across clusters, which would require synchronous communication.

...

This approach prevents a critical consistency issue: the destination cluster could retain partial transaction data that would never be completed since mirroring has stopped. This would leave the topic in an inconsistent state where read_committed consumers may be blocked due to incomplete transaction data. Additionally, the transaction coordinator would not be able to rollback these hanging transactions because there is would be no __transaction_state metadata in the destination cluster.

...

1. Destination Cluster Throttling: To avoid conflicts with intra-cluster replication controls, mirror-specific throttling configurations operate independently from standard replication throttling. The system provides two configuration levels: a broker-level rate limit (mirror.replication.throttled.rate) that sets the overall bandwidth ceiling for mirror replication traffic, and a topic-level replica list (mirror.replication.throttled.replicas) that specifies which partition-broker combinations should be throttled using the standard partition-index and :broker-id notation. Operators can dynamically adjust throttling rates at runtime without restarting brokers, first setting a cluster-wide default rate, then fine-tuning specific topic partitions as mirroring progresses. This allows gradual bandwidth allocation as mirror relationships are established.
2. Source Cluster Throttling: The source cluster side requires a different approach because mirror fetch requests operate as consumer traffic rather than replication traffic. This design is intentional since the mirroring must fetch only up to the LSO to maintain transactional consistency, which is a consumer-level guarantee not available through the replication protocol. Consequently, standard leader replication throttling mechanisms cannot apply to mirror traffic. Instead, the source cluster leverages Kafka's client quota system. Each mirror fetcher thread presents itself with a deterministic client identifier that encodes the broker ID, fetcher thread number, and mirror name. Operators can apply per-client byte rate quotas to these identifiers, effectively throttling the outbound mirror traffic from the source cluster. This approach integrates seamlessly with Kafka's existing quota enforcement infrastructure.

...

Active-active topology is not initially supported in Cluster Mirroring, though it could potentially be achieved through topic prefixing and removing the reliance on topic ID for mirroring. This is a candidate for a future improvement KIP. 

Instead, bidirectional mirroring is supported, but only when mirroring different topics between clusters, allowing records produced to either cluster to be consumed from both. Unlike MirrorMaker 2, Cluster Mirroring does not need special cycle detection or prevention logic because the read-only enforcement inherently blocks the conditions that would create infinite replication loops.

Public Interfaces

Command-Line

for mirroring. This is a candidate for a future improvement KIP. 

Instead, bidirectional mirroring is supported, but only when mirroring different topics between clusters, allowing records produced to either cluster to be consumed from both. Unlike MirrorMaker 2, Cluster Mirroring does not need special cycle detection or prevention logic because the read-only enforcement inherently blocks the conditions that would create infinite replication loops.

Public Interfaces

Command-Line

A new dump flag allows to decode cluster mirroring metadata for debugging purpose:

$ bin/kafka-dump-log.sh --mirror-state-decoder --files /tmp/server*/data/__mirror_state-1/00000000000000000000.log
Dumping /home/fvaleri/Documents/kafka/build/test/server4/data/__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}}

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

...

The CreateMirror API allows users to create a mirror and supply its configuration. When the broker receives the request, it validates that the mirror name is not already in use, contains only permitted characters, and does not end with the ".removed" suffix. Once validated, the request is forwarded to the controller, which persists the configuration in the metadata log.

...

The RemoveTopicsFromMirror API allows users to detach topics from their associated mirror. The broker validates that all target topic partitions are in either PREPARING or MIRRORING state. Once validated, the request is forwarded to the controller, which appends the ".removed" suffix to the mirror.name topic config to mark the topics as no longer mirrored.

...

The LastMirroredOffset API allows destination cluster partition leaders in PREPARING state to query the last mirrored offset LMO from the source cluster. If the source cluster has no record of this offset in its internal topic, it returns 0, meaning the log must be truncated to the beginning and mirroring starts from scratch. This is particularly important during failback. The last mirrored offset identifies where mirrored data ends and un-mirrored data begins. Records beyond this offset must be truncated before mirroring new data from the new source cluster; otherwise, the two clusters would contain inconsistent data.

...

Cluster Mirror is not compatible with MirrorMaker 2 (MM2). This is a critical consideration for users planning to migrate from MirrorMaker 2 to Cluster Mirroring.

...