Page History

...

• 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 mirror epochs from the source cluster and truncates logs to align the local log with the source.
• MIRRORING: All ISR set members have completed truncation. A mirror fetcher thread is started to continuously replicate records from the source cluster.
• PAUSING: Triggered by the pause operation. The system removes fetchers for the affected partitions. Valid from: MIRRORING only.
• PAUSED: Fetchers have been removed. The partition stays read-only with no active fetchers and no metadata sync (configs, consumer groups, ACLs). On resume, transitions directly to MIRRORING.
• STOPPING: Performs the following actions (see MirrorFetcherThread paragraph for more details)When a mirror partition enters the STOPPING state, the following operations execute sequentially.
  1. Remove mirror fetcher threads : Stops to stop cross-cluster replication for the affected partitions.
  2. In parallel:Bump leader epoch (async): Sends a BumpLeaderEpochs by sending a request to the controller with the latest local log epoch as minLeaderEpoch, ensuring the destination leader epoch exceeds . The destination's leader epoch must exceed the source cluster's last known epoch., fencing any stale producers.
  3. Append ABORT markers for all ongoing transactions.
  4. Update last mirror epochs in the Truncate to LSO, then update LME: Truncates each partition's log to LSO, discarding any uncommitted tail, then persists the latest leader epoch from each partition's log into the __mirror_state coordinator topic, recording LME for the latest leader epoch from each partition for future failback.
  5. Write PID reset barrier: Once both parallel branches complete, appends a MIRROR_PID_RESET control record to each barrier per partition, fencing stale producer IDs which clears all producer state entries from the source clusterProducerStateManager.
• STOPPED: The topic becomes writable on the destination cluster. The mirror fetcher is removed and the read-only flag is cleared.
• FAILED: An error occurred. Valid from: any state. The operator that wants to restart a failed mirror partition can remove the topic from the mirror and add it back again. The STOPPING handler already removes fetchers and truncates to the LSO, both of which are safe operations on a failed partition (fetchers are likely already gone, and truncation is a best-effort cleanup). More  More sophisticated recovery strategies can be added later with a follow-up KIP.

...

The MirrorFetcherThread (MFT) is a specialized implementation of AbstractFetcherThread that handles cross-cluster data replication with consumer Fetch requests and different epoch semantics than standard intra-cluster replication, but keeping the same log consistency validations. The destination cluster's leader replica is not registered as a follower in the source cluster. Using a follower Fetch request would cause the source broker to attempt updating follower replica status for a replica it doesn't know about. A consumer Fetch request avoids this issue, as it carries no such side effects on the source broker's replica state. In other words, destination partition leaders operate in a dual-role. They act as followers when fetching committed data from the source cluster leader up to the last stable offset (LSO), while simultaneously serving as leaders for their local replicas in the destination cluster. 

A mirror topic is created with the same topic ID as in the source cluster. This serves two purposes: it satisfies fetch request validation on the source broker, and it enables identity verification during failback where the destination cluster can confirm it is working with the exact same topic by comparing topic IDs. To maintain data consistency, destination partitions are marked as read-only and reject produce requests from clients with with ReadOnlyTopicException.

Mirror Leader Epoch

...

The approach is to proactively expire the stale producer state on failover. The key insight is that, during mirroring, the destination partition is read-only: no local producers exist, so all ProducerStateManager (PSM) entries originate from mirrored data. When mirroring stops, all PSM entries are stale and can be safely expired. Records from the source are stored as-is on the destination, with no PID modification, which otherwise would require a checksum recalculation. A MIRROR_PID_RESET control record (type 7) is written to each destination partition's log during the STOPPING state transition, after the fetcher has been removed and truncation to LSO is completed, but just before the partition becomes writable.

When a failover happens, the relevant mirror partition state transitions are:

...

The key follows the standard control record format (version=0, type=7). The value uses the following schema:

...

• Active-passive (A to B): B mirrors from A, stores records as-is. On failover, the MIRROR_PID_RESET record expires all PSM entries. Local producers get fresh PIDs from the coordinator with no collision risk.
• Failback (A to B, then B to A): After failover, B becomes writable. Later, A starts mirroring from B and truncates its log to the LSO. A then stores B's records as-is. B's MIRROR_PID_RESET record is included in the fetched data and appended to A's log, triggering PSM expiration on A. This is consistent with the general rule: when the MIRROR_PID_RESET record is encountered during append or during log recovery, all producer entries are removed from the PSM. A will write its own MIRROR_PID_RESET record when it eventually stops mirroring from B, producing a clean slate before A becomes writable again.
• Fan-out (A to B, A to C): B and C mirror independently from A, each with its own PSM per partition. On failover, each writes its own MIRROR_PID_RESET record independently.
• Fan-in (A to C, B to C, different topics): Each topic's partitions have independent PSMs. The MIRROR_PID_RESET record is written per partition during the STOPPING transition of each mirror.
• Chain (A to B to C): B mirrors from A, stores records as-is. C mirrors from B, stores records as-is. On failover at any point in the chain, the MIRROR_PID_RESET record expires all PSM entries on the stopping node. Longer chains work inductively by the same principle.

Exactly-Once Semantics

Cluster Mirroring ensures transactional consistency when stopping by truncating to the last stable offset. Note that this doesn't mean it supports mirroring does not support exactly-once semantics (EOS) across clusters, which would require synchronous communication.

During the mirror stopping transition, the MirrorCoordinator performs a log truncation operation that resets each mirror partition to its LSO. This offset represents the point in the log where all transactions have been decided (committed or aborted), essentially the highest offset where data is known to be consistent from a transactional perspective. Any records beyond this point may belong to incomplete transactions and should not persist after mirroring stops. Note that the actual lag may be greater than what's reported by the metrics. 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 would be no __transaction_state metadata in the destination cluster.

Kafka consumers with isolation.level=read_committed determine transaction visibility using only the LSO, which is computed from COMMIT/ABORT control markers in the log. Consumers never interact with the transaction coordinator or validate producer IDs. This separation between log-level markers (replicated) and coordinator state (not replicated) is why transactional consumers work correctly on mirror topics without mirroring coordinator state. The LSO truncation during failover ensures all remaining transactions have mirrored markers, maintaining this guarantee.

Consider this source cluster log:

. Transactional support means that after failover, the destination cluster will not have hanging transactions that block READ_COMMITTES consumers, but it does not guarantee that committed records from the source are atomically synced to the destination.

The mirror fetcher thread uses READ_UNCOMMITTED isolation, so records from uncommitted transactions are replicated to the destination before the source decides them. This reduces replication lag compared to READ_COMMITTED, but means uncommitted data is visible to READ_UNCOMMITTED consumers on the destination before failover. When stop mirroring is triggered, all in-flight transactions are decided by appending explicit ABORT markers, preserving all previously committed data.

In this example, source cluster log at the time of failure:

Destination cluster log at failover (replication reached offset 2):

After the STOPPING transition appends abort markers: 

Transaction 4001 was committed at the source but aborted at the destination because the COMMIT marker (offset 3) had not yet been replicated. Transaction 4002 was correctly aborted at both clusters. Applications that require strict transactional guarantees across clusters should implement deduplication or reconciliation logic after failover.

Additionally, the kafka-transactions tool can only abort transactions originated from the local cluster. It cannot abort transactions replicated via mirroring because the __transaction_state topic is not mirrored. Hanging transactions from mirrored data are resolved exclusively by the STOPPING transition flow described aboveIf replication reaches offset 4 and the source cluster fails, the destination cluster contains data records for transaction 4002 (offset 2) without the abort marker (offset 4). This creates a hanging transaction that can never be committed or aborted on the destination cluster. Note that this approach causes data loss for any in-flight transactions or non-mirrored completed transactions when we experiencing a lag during the failover and may result in already-processed records being lost if consumers on the destination cluster read uncommitted data.

Bandwidth Control

Cluster Mirroring adopts a dual-sided throttling mechanism that extends Kafka's existing bandwidth control capabilities to work across cluster boundaries.

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

...

1. User sends StopMirrorTopics request with topics and mirror name.
2. The controller validates each topic belongs to the specified mirror and is in MIRRORING state. It then updates the topic config by appending the .removed suffix, e.g. mirror.name=my-mirror.removed, generating a ConfigRecord.
3. When the MirrorMetadataManager gets notified, it detects the .removed suffix on mirror.name. It queries the current mirror partition state from the coordinator, and transitions the mirror partition to STOPPING.
4. During STOPPING, the following operations execute sequentially:
   
   1. The MirrorFetcherManager removes all fetcher threads for the affected partitions, stopping replication.
   2. Bump the The leader epoch is bumped for the partitions to ensure monotonically increasing epochs for new records.The log is truncated to LSO for transactional consistency

   3. ABORT markers are appended for all ongoing transactions. For each partition, ProducerStateManager provides the set of in-flight transaction entries and an EndTransactionMarker(ABORT) is appended for each one. This resolves hanging transactions without truncating committed data.

   4. The LME is recorded as sKeyLastMirrorEpochsKey/LastMirrorEpochsValue records into the __mirror_state topic for potential future failback.

   5. A MIRROR_PID_RESET control record is written to the partition log, which expires all ProducerStateManager entries so that new producers get fresh PIDs with no collision risk.

5. The state transitions from STOPPING to STOPPED. The read only flag is cleared and the topic becomes writable. New producers can start producing with fresh PIDs starting at sequence 0 and a higher leader epoch.

...

...