Page History

...

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

We use a composite key (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 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 __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 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 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.

...

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 epoch from the source cluster and truncates 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.
EPOCH_FENCING: The destination leader epoch needs to be bumped. A BumpLeaderEpochs request is sent to the controller. On success, the partition transitions to MIRRORING.
PAUSING: Triggered by the pause operation. The system removes fetchers for the affected partitions.
PAUSED: Fetchers have been removed. The partition stays read-only with no active fetchers and no metadata sync (configs, consumer groups, ACLs)synchronization. On resume, transitions directly to MIRRORING.
STOPPING: The mirror fetcher is removed, the leader epoch is bumped, ABORT markers are appended for all ongoing transactions, last mirror epochs are persisted, and a MIRROR_PID_RESET barrier is writtenrecord appended.
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 restart topic mirroring. More sophisticated recovery strategies can be added later with a follow-up KIP.

...

The MirrorMetadataManager (MMM) implements periodic metadata synchronization between source and destination clusters. It maintains persistent network connections to all source clusters. During periodic metadata refresh, the broker validates that the source cluster ID has not changed. If a mismatch is detected, metadata sync synchronization for that mirror is halted and an error is logged. This prevents silent data corruption in case of misconfiguration or unintended source cluster replacement.

Metadata synchronization operates at the mirror level rather than the partition level, so it uses a separate coordinator assignment based on the mirror name alone. Only the broker assigned as the metadata coordinator for a given mirror performs synchronization, and it applies changes only to the mirror partitions it manages. This avoids both redundant synchronization across brokers and unnecessary updates to partitions managed by other coordinators.

Responsibilities:

Connection Management: The manager maintains a connection pool with one blocking sender per source cluster. These connections are created lazily when the first topic for a mirror is added. Each sender uses the security credentials and network settings from the mirror configuration, allowing different mirrors to use different authentication mechanisms.
Topic Metadata Synchronization: Every refresh cycle, the manager fetches topic metadata from source clusters using standard MetadataRequest calls. For each topic in the mirror configuration:
1. Topic Creation: If a topic exists in the source but not the destination, the manager sends a CreateTopics request to the controller with identical partition count and configurations.
2. Partition Expansion: If the source topic has more partitions than the destination, the manager sends a CreatePartitions request to scale up the destination topic to match.
3. Configuration Sync: Topic configurations are compared between source and destination. Any differences trigger an IncrementalAlterConfigs request to align destination configs with the source.
  1. Topic Auto-Discovery: Periodically discovers new topics on the source cluster that match mirror.topics.include and do not match mirror.topics.exclude, sending a

StartMirrorTopicsRequest

1. 1. StartMirrorTopics request to the controller for atomic topic creation. Exclude patterns are also enforced on already-mirroring topics.
2. Topic Deletion: When a topic is deleted on the source cluster, the mirror partitions on the destination cluster moves to STOPPED state. This prevents accidental deletions to affect the destination cluster. In case it was intentional, the operator would need to manually remove the topic from the mirror.
Group Offset Synchronization: The manager synchronizes classic and share consumer group offsets to enable seamless failover (no offset translation):
1. Consumer groups: Lists consumer groups on the source cluster, then fetches their committed offsets. For each group, offsets are filtered to include only partitions belonging to actively mirrored topics. Groups that are active on the destination cluster (i.e., in any state other than EMPTY or DEAD) are skipped to avoid regressing offsets after failover. The filtered offsets are committed to the destination.
2. Share groups: Lists share groups on the source cluster, then fetches their Share‐Partition Start Offsets (SPSO). Offsets are filtered and active‐group checks are applied in the same way as for consumer groups. The filtered offsets are committed to the destination.
ACL Synchronization: Access control lists are mirrored from source to destination to maintain consistent security policies:
1. Fetches all ACLs from the source using DescribeAcls request.
2. Compares with the destination cluster's current ACLs from the metadata image.
3. Creates missing ACLs using CreateAcls request.
4. Deletes ACLs that exist in destination but not in source using DeleteAcls request.

Cluster Mirroring allows users to modify configurations in the destination cluster, though these changes are periodically overridden by the topic configuration synchronization cycle. This design choice was made because while dynamic configuration changes could be blocked, static configuration changes via properties files cannot be prevented, making override inevitable. However, this approach presents challenges in environments with external governing systems like the Strimzi operator, where the continuous reconciliation process conflicts with the refresh cycle, potentially causing performance impacts. More critically, temporary configuration mismatches such as reduced retention periods or altered partition counts could lead to data loss or missing partitions until the next synchronization cycle detects and corrects the discrepancy, highlighting the need for careful operational awareness when mixing mirroring with external cluster management solutions.

Metadata synchronization operates at the mirror level rather than the partition level, so it uses a separate coordinator assignment based on the mirror name alone. Only the broker assigned as the metadata coordinator for a given mirror performs synchronization, and it applies changes only to the mirror partitions it manages. This avoids both redundant synchronization across brokers and unnecessary updates to partitions managed by other coordinators.

MirrorFetcherThread

The MirrorFetcherManager (MFM) extends AbstractFetcherManager to handle fetcher thread lifecycle for mirror partitions. It uses a three-dimensional key (fetcher ID, source broker endpoint, mirror name) to organize threads, ensuring that:

...

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

Mirror Leader Epoch

...

In KAFKA-18723, we identified a race condition where a late-arriving fetch response could contain corrupted or inconsistent records. The fix ensures that only record batches whose partition leader epoch is less than or equal to the leader epoch in the Fetch request are appended. The destination cluster stores batches using the leader epoch from the source cluster. For the mirroring leader in the destination cluster, this works naturally: the leader epoch in the Fetch request is set to the latest source cluster leader epoch, so the existing validation applies without issue. For followers in the destination cluster, however, the situation is different. During mirroring, the local leader epoch diverges from the batch leader epoch. The fetched batch may carry a leader epoch of X while the local leader epoch is Y, where X > Y or X < Y. In either case, the fix no longer applies correctly. To address this, we introduce the MirrorLeaderEpoch field in the Fetch request and response.
The CurrentLeaderEpoch in Fetch response serves 2 purposes:
The leader will verify it to make sure the fetch request is up-to-date
The fetch response receiver will use it to validate the records in the fetch response.
For followers in the destination cluster, the CurrentLeaderEpoch can only serve for the first purpose. For the second purpose, because of the leader epoch inconsistency in the batches and the local metadata, the validation will not work. Therefore, the MirrorLeaderEpoch in the Fetch request will be set to the latest leader epoch in the leader's log, and the CurrentLeaderEpoch will still be set to the local current leader epoch. This way, when receiving the Fetch response, the follower's validation can work as expected.

Log Convergence

There are two main concepts to keep in mind when dealing with log convergence across Kafka clusters:

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

A two-phase truncation protocol is applied by the mirror fetchers:

LME truncation: Synchronizes leader epoch history between clusters. Truncates log at the start offset of the first non-mirrored epoch and waits until all ISRs complete truncation. After this phase: epoch histories match, but log may still diverge within the last epoch.
Replication truncation: Triggered by the existing leader epoch comparison during fetch. Handles offset-level divergence within remaining epochs. Truncates to the exact offset where the source's epoch ends. After this phase: logs fully converged, normal replication proceeds.

Common scenarios:

A) Cluster B mirrors from source cluster A for the first time. A has no LME knowledge for this partition, so it returns epoch -1. B truncates everything and replicates from scratch.

Image Added

B) Mirroring stops on B. B stores LME=1 and bumps to epoch 2. Meanwhile A also bumps to epoch 2 and gets a new record. Then A starts mirroring from B. A gets LME=1 from B, truncates the new record, and starts replicating from there.

Image Added

Unclean Leader Election

Cluster Mirroring relies on leader epoch alignment between source and destination to guarantee log consistency. This mechanism assumes that the source cluster's log is an authoritative, append-only sequence of records for each leader epoch. That assumption holds as long as leader elections on the source are clean, meaning each new leader was a fully caught-up ISR member and no committed records were lost during the transition. When unclean leader election (ULE) occurs on the source cluster, this assumption breaks. An out-of-sync replica becomes leader and the source log silently loses committed records from the previous epoch. The source cluster's log now has a gap or a divergent suffix that was never replicated to the destination.

In this example we see how LME truncation and the subsequent replication protocol resolve an ULE that happens before mirroring starts.

1) Cluster B is mirroring from cluster A, and the leader node in A has a failure.

Image Added

2) Unclean leader election is triggered in cluster A. The new leader only contains a record at offset 0. Then 3 more records are appended.

Image Added

3) Before cluster B detects the leadership change, failover to B. And when A failback, the LME log truncation will truncate records beyond epoch 1.

Image Added

The problem is that Cluster Mirroring cannot detect log divergence caused by ULE that happens after LME truncation. A non-ISR replica that missed truncation may still hold records with leader epochs beyond the LME. If that replica becomes leader through ULE, the replication protocol cannot detect the divergence.

In this example, Cluster B is mirroring from A, and there are ULEs triggered 3 times in cluster A.

1) This is the first time, and the new leader appends 3 records with epoch 2.

Image Added

2) After the second ULE, the

Cluster mirroring enforces the invariant that the destination leader epoch (DLE) is always greater than the source leader epoch (SLE). This invariant is required to prevent a liveness problem in destination consumers.

When a destination consumer initializes a partition, it retrieves the last committed offset along with its leader epoch from the group coordinator. If the committed leader epoch originates from the source cluster and is greater than the local leader epoch, the consumer's Metadata.updateLastSeenEpochIfNewer accepts it, causing all subsequent metadata updates from the destination cluster to be filtered out. The partition enters AWAIT_VALIDATION but requests cannot be sent because the partition has no known leader node. This creates an infinite loop of metadata refreshes that never resolves.

Enforcing DLE > SLE guarantees that committed offsets from the source cluster always carry an epoch lower than the destination's current epoch, so destination consumers can validate positions normally. Before appending mirrored data, the destination sets DLE = SLE + LEADER_EPOCH_BUMP_INCREMENT whenever SLE > DLE − LEADER_EPOCH_BUMP_THRESHOLD. The increment is 10 and the threshold is 3, which provides a window of 7 source leader elections before the next bump. These two constants are a tradeoff: larger values mean fewer epoch bumps but faster epoch consumption, smaller values mean more bumps but slower epoch growth.

There are three bump trigger points:

Reactive: During record ingestion, if the source batch epoch exceeds the local epoch (SLE > DLE), a MirrorLeaderEpochExceededException is thrown. The partition transitions to EPOCH_FENCING, which gates the transition back to MIRRORING until the bump is confirmed.
Proactive: If the source batch epoch is within the threshold (SLE > DLE − LEADER_EPOCH_BUMP_THRESHOLD), a background bump is scheduled while the current batch is still appended.
Periodic: As an optimization, the coordinator checks all mirrored partitions during source cluster metadata sync and bumps any partition where the threshold condition holds.

The bump is performed via the BumpLeaderEpochs API, which sets a minimum leader epoch on the controller. The active controller writes a PartitionChangeRecord to the KRaft metadata log with the requested minLeaderEpoch. When applied, the new leader epoch is set to max(minLeaderEpoch + 1, currentLeaderEpoch), ensuring it is at least one greater than the requested minimum. If the current epoch already exceeds the requested value, no record is produced. The updated epoch propagates to all brokers through the standard metadata image update mechanism.