Authors: Luke Chen, Federico Valeri, Omnia Ibrahim, PoAn Yang, Kuan-Po Tseng, Jiunn-Yang Huang
Current state: Under Discussion
Discussion thread: here
JIRA:
Please keep the discussion on the mailing list rather than commenting on the wiki (wiki discussions get unwieldy fast).
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 (MM2) provides cross-cluster replication capabilities, it presents significant operational challenges.
Cluster Mirroring addresses these operational challenges by integrating cross-cluster replication directly into Kafka brokers, providing a simpler and more robust solution for cross-cluster replication. Producers write to the source cluster and receive acknowledgments based on the source cluster's replication requirements (e.g. acks=all ensures replication to all in-sync replicas within the source cluster). Data is then asynchronously replicated to destination clusters with no impact on producer latency or throughput. While Cluster Mirroring is optimized for geo-replication, DR and migration use cases where a single source cluster replicates to one or more destination clusters, it also provides a foundation for more complex topologies.
Cross-datacenter network latency makes synchronous replication impractical for some deployments. Requiring synchronous acknowledgment from a geographically distant cluster would introduce significant latency (typically 50-200ms for inter-region replication), making it unsuitable for latency-sensitive applications. Asynchronous replication should provide the right balance for 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. In the event of a catastrophic failure of the source cluster, recently produced records that have not yet been replicated to the destination cluster will be lost. The amount of data loss depends on replication lag at the time of failure. Organizations must handle a non-zero RPO determined by the replication lag between source and destination clusters. Typical replication lag ranges from seconds to minutes depending on network bandwidth, throughput, and geographic distance.
Stretched clusters are not suitable for DR scenarios because they provide no protection against software failures, configuration errors, and accidental topic deletions. 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.
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-mirrors.sh CLI tool, with its state persisted in a coordinator that manages cross-cluster replication.
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.
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.
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 a mirror with hundreds of partitions will have its state management distributed evenly across the cluster rather than concentrated on a single broker.
Responsibilities:
State descriptions:
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 for that mirror is halted and an error is logged. This prevents silent data corruption in case of misconfiguration or unintended source cluster replacement.
Responsibilities:
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.
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:
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.
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.
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:
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.
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:
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.
There are two main concepts to keep in mind when dealing with log convergence across Kafka clusters:
A two-phase truncation protocol is applied by the mirror fetchers:
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.
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.
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.
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.
3) Before cluster B detects the leadership change, failover to B. And when A failback, the log truncation of LME will truncate records beyond epoch 1.
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 unclean election, the replication protocol cannot detect the divergence.
1) Cluster B is mirroring from A, and there are ULEs triggered 3 times in cluster A. This is the first time, and the new leader appends 3 records with epoch 2.
2) After the second ULE, the new leader is the ex-leader before step 1. Now, failover to B, B bumps leader epoch to 2 and appends record in offset 3. And A starts to mirror B.
3) After the third ULE, the leader is the ex-leader in step 1. Now the log diverge cannot be resolved by the replication protocol.
A new dynamic topic level configuration mirror.support.unclean.leader.election (boolean, default false) is introduced to support ULE. When enabled, LME log truncation waits for all replicas (not just ISR members) to join the ISR and complete the truncation. This ensures that every replica has been truncated past the LME, so even subsequent unclean leader elections cannot introduce undetectable divergence. If some replicas cannot catch up due to slow network or disk issues, mirroring remains pending or enters the FAILED state, requiring manual intervention (resolving the issue or reassigning partition replicas to healthy brokers). This is a dynamic configuration, so it can be enabled at any point during the mirroring lifecycle. If it is enabled before mirroring starts, all records are guaranteed to be consistent between the two clusters, even if an unclean leader election occurs. If it is enabled when mirroring is already running, only records produced after the next LME log truncation are guaranteed to be consistent. If some replicas cannot catch up with the leader during LME log truncation due to slow network, disk issues, or other failures, the mirror partition moves to the FAILED state. In this case, users have to manually resolve the underlying issue or reassign the partition replicas to healthy brokers, and then restart mirroring.
Cluster Mirroring preserves the compression format of record batches from the source cluster without recompression. When mirroring data, compressed record batches are copied directly from the source to the destination cluster, maintaining the original compression type (gzip, snappy, lz4, zstd, or none) and the exact byte-level representation of the data. This approach avoids unnecessary CPU overhead from decompression and recompression during replication, ensures bit-for-bit data integrity, and prevents potential issues with different compression implementations producing different outputs for the same data.
Cluster Mirroring fully supports log compacted topics, preserving both compacted records and offset gaps from the source cluster. When a topic uses cleanup.policy=compact, Kafka removes obsolete records with duplicate keys, creating gaps in the offset sequence. For example, if a source partition contains offsets 0-100 and compaction removes records at offsets 30-40 and 60-70, the remaining records will have gaps: offsets 0-29, 41-59, and 71-100 are missing. 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 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.
When a destination cluster lags behind the source on a compacted topic, tombstone records may not yet be replicated at the time of failover. For example, if the source has a tombstone at offset 100 that deletes a key originally written at offset 3, but the destination has only replicated up to offset 50, that tombstone is never applied on the destination. After failover, offset 3 remains as a stale entry that will never be cleaned up. The problem is compounded on failback: truncating the former source to match the destination also discards the tombstone, so the orphaned key persists in both clusters permanently. This is an inherent limitation of asynchronous replication. Compaction correctness depends on the full sequence of tombstones being present, and any that fall beyond the replication watermark at failover time are lost. The same problem exists with MirrorMaker 2. The follow-up KIP for synchronous mirroring may address this by ensuring zero lag at switchover, guaranteeing all tombstones are replicated before failover occurs.
Cluster Mirroring handles topic retention policies by periodically synchronizing the topic configurations from the source cluster, ensuring that the topic retention policies are consistent. When the source cluster applies retention policies, older log segments are deleted and the log start offset advances. For example, if a topic originally contained offsets 0-100 and retention deletes offsets 0-99, the source cluster's log start offset becomes 100. When the mirror leader fetches from the source, it discovers the new log start offset and updates its local log start offset to match, creating the same offset gap. 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.
Cluster Mirroring synchronizes group offsets from the source cluster to the destination cluster, enabling consumers to resume consumption from their last position after failover. Both traditional consumer groups and share consumer groups (Kafka Queue functionality) are supported, with offset synchronization running in two separate phases to prevent cross-type conflicts. For consumer groups, committed offsets are periodically fetched from the source and applied to the destination. For share groups, which use a different offset management model based on Share-Partition Start Offset (SPSO) and Share-Partition End Offset (SPEO), the current SPSO is retrieved from the source and applied to the destination, which also initializes the group state in both the group coordinator and share coordinator. This means a share group can be initialized in the destination cluster even if it doesn't exist yet, eliminating the need for pre-creation or complex state management.
Each phase lists only groups of its own type on both source and destination, preventing a consumer group on the source from overwriting a share group with the same name on the destination (or vice versa). Kafka enforces that consumer group and share group names must be unique within a single cluster. When a name conflict does occur across types, the offset commit operation will fail for the affected group, which is logged and skipped without affecting other groups. Users can resolve these conflicts by deleting the conflicting group in the destination cluster or excluding it from offset synchronization. These conflicts affect only offset synchronization and do not impact data mirroring itself. Fetched offsets are filtered to only include partitions belonging to actively mirrored topics, preventing offset ping-pong in bidirectional setups and excluding internal topics. Groups that are active on the destination (non-EMPTY, non-DEAD) are skipped. Offset sync prepares the destination for failover; once consumers are active there, they own their offsets. When they stop and the group becomes EMPTY, sync resumes automatically.
During offset synchronization, the committed offset in the destination cluster may temporarily exceed the current log end offset (LEO) of the mirror topic. For example, if a consumer commits offset 100 in the source cluster but the destination cluster has only mirrored up to offset 80, the offset 100 is still committed to the destination. This is acceptable because the mirror leader continues fetching data and the LEO will eventually catch up. However, if a failover occurs before the mirrored data catches up, consumers attempting to resume from offset 100 will receive an OffsetOutOfRangeException. To handle this gracefully, consumers should configure auto.offset.reset=latest when consuming from mirror topics, ensuring that if a committed offset is beyond the current LEO after failover, the consumer automatically resets to the latest available offset.
Cluster Mirroring supports comprehensive security controls through both authorization and authentication mechanisms. This ensures that only authorized principals can establish and manage cluster mirrors. When configuring a mirror, operators specify ACLs that should be synchronized from the source cluster, and these ACLs are periodically replicated to the destination cluster to maintain consistent access control policies across both environments.
When connecting to the source cluster, Cluster Mirroring requires only the bootstrap server address and appropriate credentials, no other sensitive cluster information is exposed or required. The destination cluster's mirror configuration supports all standard Kafka authentication mechanisms including TLS/SSL for encrypted transport and SASL for client authentication. Each mirror can be configured with its own security settings, allowing different mirrors to connect to source clusters with varying security requirements. This enables secure cross-cluster replication even when source and destination clusters use different authentication protocols or when connecting across security boundaries such as on-premises to cloud environments. All credentials are stored as mirror configuration records in the destination cluster metadata log, and used exclusively for establishing authenticated connections to the source cluster.
Source cluster permissions (mirror principal):
RPC | Component | ACL Operation | ACL Resource | Purpose |
| Fetch | MFT | Read | Topic | Data replication |
| Metadata | MMM | Describe | Topic | Topic discovery and leader tracking |
| DescribeConfigs | MMM | DescribeConfigs | Topic | Topic configuration sync |
| ListGroups | MMM | Describe | Group | Consumer group offset sync |
| OffsetFetch | MMM | Describe | Group | Consumer group offset sync |
| DescribeAcls | MMM | Describe | Cluster | ACL synchronization |
| DescribeMirrors | MC | Read | Cluster | Log truncation when preparing |
| ApiVersions | MMM | Feature negotiation | ||
| ListOffsets | MFT | Describe | Topic | Offset bounds discovery |
| OffsetsForLeaderEpoch | MFT | Describe | Topic | Leader epoch validation for truncation |
Destination cluster permissions:
RPC | Component | ACL Operation | ACL Resource | Purpose |
| CreateMirror | Controller | Create | ClusterMirror | New cluster mirror creation |
| StartMirrorTopics | Controller | Alter | ClusterMirror | Mirror topics creation |
| StartMirrorTopics | Controller | AlterConfigs | Topic | Mirror topics creation |
| StopMirrorTopics | Controller | Alter | ClusterMirror | Mirror topics removal (failover) |
| StopMirrorTopics | Controller | AlterConfigs | Topic | Mirror topics removal (failover) |
| PauseMirrorTopics | Controller | Alter | ClusterMirror | Mirror topics pause |
| PauseMirrorTopics | Controller | AlterConfigs | Topic | Mirror topics pause |
| ResumeMirrorTopics | Controller | Alter | ClusterMirror | Mirror topics resume |
| ResumeMirrorTopics | Controller | AlterConfigs | Topic | Mirror topics resume |
| DeleteMirror | Controller | Alter | ClusterMirror | Delete a cluster mirror |
| ListMirrors | Broker | Describe | ClusterMirror | Mirror topic listing |
| DescribeMirrors | Broker | Describe | ClusterMirror | Mirror topic describe (state, lag) |
| DescribeConfigs | Broker | DescribeConfigs | ClusterMirror | Mirror configuration describe |
| WriteMirrorStates | MC | ClusterAction | Cluster | Mirror partition state write |
| ReadMirrorStates | MC | ClusterAction | Cluster | Mirror partition state read |
| BumpLeaderEpochs | MC | ClusterAction | Cluster | Leader epoch bump when stopping |
| FindCoordinator | Broker | ClusterAction | Cluster | Mirror coordinator location |
| CreateTopics | MMM | Create | Topic | Topic creation |
| CreatePartitions | MMM | Alter | Topic | Partitions scaling |
| IncrementalAlterConfigs | MMM | AlterConfigs | ClusterMirror | Mirror configuration update |
| OffsetCommit | MMM | Read | Topic | Source CG offsets commit |
| OffsetCommit | MMM | Read | Group | Source CG offsets commit |
| CreateAcls | MMM | Alter | Cluster | Source ACLs creation |
| DeleteAcls | MMM | Alter | Cluster | Source ACLs removal |
An operator can grant ClusterMirror:*:CREATE,ALTER,DESCRIBE for full mirror management, or scope it to specific mirrors like ClusterMirror:prod-dr:DESCRIBE for read-only monitoring of a single mirror, without granting any broker-level privileges.
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 is written to each destination partition's log during the STOPPING state transition, just before the partition becomes writable.
When the control batch is encountered during append or during log recovery, all producer entries are removed from the PSM. This ensures both leaders and followers handle the control batch barrier consistently. Given that the partition is read-only during mirroring, all PSM entries originate from mirrored data. Expiring all entries is safe: no local producer state exists to preserve. Control batches are filtered out by the consumer fetcher via isControlBatch checks, just like transaction markers (commit/abort). The log dump tool is enhanced to deserialize MIRROR_PID_RESET records.
The control record approach works correctly with all practical mirroring topologies:
Cluster mirroring does not support exactly-once semantics across clusters. Transactional support means that after failover, the destination cluster will not have hanging transactions that block READ_COMMITTED 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, ongoing transactions are decided by appending explicit ABORT markers, preserving all previously committed data.
In this example, source cluster log at the time of failure:
Offset | Type | PID | Content |
0 | DATA | 4001 | key=A, value=1 |
1 | DATA | 4001 | key=B, value=2 |
2 | DATA | 4002 | key=X, value=9 |
3 | COMMIT | 4001 | |
4 | DATA | 4003 | key=Y, value=5 |
5 | DATA | none | key=Z, value=10 |
Destination cluster log at failover (replication reached offset 2):
Offset | Type | PID | Content |
0 | DATA | 4001 | key=A, value=1 |
1 | DATA | 4001 | key=B, value=2 |
2 | DATA | 4002 | key=X, value=9 |
After the STOPPING transition appends abort markers:
Offset | Type | PID | Content |
0 | DATA | 4001 | key=A, value=1 |
1 | DATA | 4001 | key=B, value=2 |
2 | DATA | 4002 | key=X, value=9 |
3 | ABORT | 4001 | |
4 | ABORT | 4002 |
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 above.
Cluster Mirroring adopts a dual-sided throttling mechanism that extends Kafka's existing bandwidth control capabilities to work across cluster boundaries.
In a follow-up KIP we will add source-side throttling that allows source cluster leaders to limit bandwidth served to all mirror fetchers, similar to how leader.replication.throttled.rate controls intra-cluster replication. This provides independent control over mirror catch-up traffic without impacting local replication or consumer workloads. Combined with destination-side throttling, operators gain complete bidirectional bandwidth control for mirror traffic.
Mirror topics in the destination cluster currently only replicate data from local storage on the source broker. Integrating with tiered storage would allow mirroring to handle data that has been offloaded to remote storage (e.g., S3, HDFS), enabling full replication of topics with long retention periods without requiring all data to reside in local broker storage. A detailed design of the metadata synchronization protocol, API schema, and state management will be provided in a follow-up KIP.
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.
Currently, mirroring is asynchronous. The source cluster acknowledges the producer without waiting for the destination to replicate the data. Sync mirroring would guarantee that records are replicated to the destination cluster before the source acknowledges the produce request, providing stronger durability guarantees at the cost of higher latency. This would be useful for workloads where zero data loss across clusters is a strict requirement.
Future extensions to synchronous mirroring could enable preservation of transactional semantics across clusters. Streaming platforms using exactly-once mode (Apache Kafka Streams, Apache Flink, Apache Spark) rely on the source cluster's transactional protocol and coordination. During failover or migration scenarios, transactional metadata for pending transactions does not transfer to the destination cluster, potentially breaking exactly-once guarantees. Supporting transactional cross-cluster replication would require coordinating transactional metadata and ensuring transaction state consistency across clusters, something MM2's Connect-based architecture cannot support.
At the time of writing, the Diskless Topics design is still under discussion (KIP-1500 and other sub-KIPs), so there will be future KIPs to support this feature. Diskless topics store data exclusively in tiered storage, with no local log segments on brokers. Supporting mirroring for diskless topics requires adapting the fetch and replication mechanisms to work without local storage, which introduces changes to how mirror offsets are tracked and how truncation is handled during failover.
Failover is initiated by calling the StopMirrorTopics API, which appends a .stopped suffix to the mirror.name internal config. This transitions the mirror topics from read-only to writable state after the stopping process completes gracefully. When producers reconnect to the destination cluster after failover, they obtain new producer IDs which are separate from previously mirrored IDs, so they begin writing with fresh sequence numbers starting from 0. Consumers can reconnect to the destination cluster using the same group ID, resuming from the last synchronized offsets, minimizing data re-processing or gaps. The transition is transparent from the consumer's perspective and offset management continues normally through the destination's group coordinator.
# 9091 (source) -----> 9094 (destination) # in case of disaster, the operator can failover by running the following command bin/kafka-mirror.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) |
Failback enables mirroring to be reversed after a failover, allowing the original source cluster to become the destination and vice versa. This is critical for scenarios where you want to fail back to the original cluster after recovering from an outage or planned maintenance. When 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 LastMirrorEpochs request to the new source cluster asking for the LME, and then truncates its local log to the last offset of the returned epoch. 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 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 everse mirroring may cause the data loss if there are records that didn't get mirrored to the old destination cluster.
# when the source cluster is back, the operator can failback by creating a mirror with the same name echo "bootstrap.servers=localhost:9094" > /tmp/my-mirror.properties bin/kafka-mirrors.sh --bootstrap-server :9091 --create --mirror my-mirror --mirror-config /tmp/my-mirror.properties bin/kafka-mirrors.sh --bootstrap-server :"9091 --start --topic .* --mirror my-mirror # 9091 (destination) <----- 9094 (source) |
ABORT markers are appended for all ongoing transactions using the new leader epoch.
A MIRROR_PID_RESET control record is written to expire all producer state entries.
The user sends ListMirrorsRequest to any broker (no parameters required).
The broker handler queries two sources:
The ReplicaManager which provides source offset, destination offset, and lag for each partition.
The request is forwarded to each broker that only reports partitions for which it has lag information or is the partition leader. This avoids duplicate reporting across brokers.
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 kafka-mirrors.sh provides administrative operations for managing cluster mirrors:
$ bin/kafka-mirrors.sh --help
Create cluster mirrors and manage mirrored 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. |
Create a new cluster mirror in the destination cluster (forbidden suffixes: .stopped, .paused):
$ echo "bootstrap.servers=localhost:9092" >/tmp/mirror.properties $ bin/kafka-mirror.sh --bootstrap-server :9094 --create --mirror my-mirror --mirror-config /tmp/mirror.properties Created mirror my-mirror |
Start mirroring a topic or set of topics (the topic flag accepts regex expression):
$ bin/kafka-mirrors.sh --bootstrap-server :9094 --start \ --topics 'orders-.*' --exclude 'orders-internal' --mirror my-mirror Started 2 mirror topic(s) in mirror my-mirror: [orders-us, orders-eu] |
Stop mirroring a topic or set of topics (failover; topics become writable):
$ bin/kafka-mirrors.sh --bootstrap-server :9094 --stop --topics 'orders-us' --mirror my-mirror Stopped mirroring for 1 topic(s) in mirror my-mirror: [orders-us] |
Delete a mirror including its configuration (the mirror must be empty or include only stopped partitions):
$ bin/kafka-mirrors.sh --bootstrap-server :9094 --delete --mirror my-mirror Deleted mirror my-mirror |
Pause mirroring for a specific topic or set of topics (topics remain read-only):
$ bin/kafka-mirrors.sh --bootstrap-server :9094 --pause --topic my-topic --mirror my-mirror Paused mirroring for 1 topic(s) in mirror my-mirror: [my-topic] |
Resume mirroring for a specific topic or set of topics:
$ bin/kafka-mirrors.sh --bootstrap-server :9094 --resume --topic my-topic --mirror my-mirror Resumed mirroring for 1 topic(s) in mirror my-mirror: [my-topic] |
List configured mirrors with additional information:
$ bin/kafka-mirrors.sh --bootstrap-server :9094 --list MIRROR TOPICS CLUSTER-ID BOOTSTRAP-SERVER my-mirror 2 lBq12jYZRp-9wF3M9MPopg localhost:9091,localhost:9092 new-mirror 1 lBq12jYZRp-9wF3M9MPopg localhost:9091,localhost:9092 |
Describe configured mirrors to check their lag compared to their source topics (use --mirror flag to filter other partitions):
$ bin/kafka-mirrors.sh --bootstrap-server :9094 --describe MIRROR TOPIC PARTITION SOURCE-OFFSET DESTINATION-OFFSET LAG STATE my-mirror bar 0 - - - STOPPED my-mirror foo 0 69 66 3 MIRRORING my-mirror foo 1 94 84 10 MIRRORING my-mirror foo 2 94 90 4 MIRRORING new-mirror baz 0 - - - PAUSED new-mirror baz 1 - - - PAUSED |
Alter mirror configuration (any valid configuration triggers a reconnection):
$ bin/kafka-configs.sh --bootstrap-server :9094 --entity-type mirrors --entity-name my-mirror \
--alter --add-config 'bootstrap.servers=localhost:9092'
Completed updating config for mirror my-mirror. |
Throttling on the destination cluster:
$ bin/kafka-configs.sh --bootstrap-server :9094 --entity-type brokers --entity-name 4 \ --alter --add-config mirror.replication.throttled.rate=100000000 Completed updating config for broker 4. $ bin/kafka-configs.sh --bootstrap-server :9094 --entity-type topics --entity-name my-topic \ --alter --add-config mirror.replication.throttled.replicas=[0:4] Completed updating config for topic my-topic. |
Throttling on the source cluster:
$ bin/kafka-configs.sh --bootstrap-server :9091 --alter --add-config 'consumer_byte_rate=1024' \ --entity-type clients --entity-name broker-4-fetcher-0-mirror-my-mirror Completed updating config for client broker-4-fetcher-0-mirror-my-mirror. |
Grant mirror admin full access to a specific mirror:
$ bin/kafka-acls.sh --bootstrap-server :9094 --add \
--cluster-mirror my-mirror \
--operation Create --operation Alter --operation Describe --operation Delete \
--operation AlterConfigs --operation DescribeConfigs \
--allow-principal User:mirror-admin
Adding ACLs for resource `ResourcePattern(resourceType=CLUSTER_MIRROR, name=my-mirror, patternType=LITERAL)`:
(principal=User:mirror-admin, host=*, operation=CREATE, permissionType=ALLOW)
(principal=User:mirror-admin, host=*, operation=ALTER, permissionType=ALLOW)
(principal=User:mirror-admin, host=*, operation=DESCRIBE, permissionType=ALLOW)
(principal=User:mirror-admin, host=*, operation=DELETE, permissionType=ALLOW)
(principal=User:mirror-admin, host=*, operation=ALTER_CONFIGS, permissionType=ALLOW)
(principal=User:mirror-admin, host=*, operation=DESCRIBE_CONFIGS, permissionType=ALLOW) |
Grant read-only monitoring access to all mirrors:
$ bin/kafka-acls.sh --bootstrap-server :9094 --add \
--cluster-mirror '*' \
--operation Describe --operation DescribeConfigs \
--allow-principal User:monitor
Adding ACLs for resource `ResourcePattern(resourceType=CLUSTER_MIRROR, name=*, patternType=LITERAL)`:
(principal=User:monitor, host=*, operation=DESCRIBE, permissionType=ALLOW)
(principal=User:monitor, host=*, operation=DESCRIBE_CONFIGS, permissionType=ALLOW) |
List ACLs for a specific mirror:
$ bin/kafka-acls.sh --bootstrap-server :9094 --list --cluster-mirror my-mirror
Current ACLs for resource `ResourcePattern(resourceType=CLUSTER_MIRROR, name=my-mirror, patternType=LITERAL)`:
(principal=User:mirror-admin, host=*, operation=CREATE, permissionType=ALLOW)
(principal=User:mirror-admin, host=*, operation=ALTER, permissionType=ALLOW)
(principal=User:mirror-admin, host=*, operation=DESCRIBE, permissionType=ALLOW)
(principal=User:mirror-admin, host=*, operation=DELETE, permissionType=ALLOW)
(principal=User:mirror-admin, host=*, operation=ALTER_CONFIGS, permissionType=ALLOW)
(principal=User:mirror-admin, host=*, operation=DESCRIBE_CONFIGS, permissionType=ALLOW) |
New methods are added to the Admin interface for programmatic cluster mirror management, along with their supporting classes:
/**
* Create a new cluster mirror.
*
* @param mirrorName The name of the cluster mirror
* @param configs Configuration for the cluster mirror, including bootstrap servers and security settings
* @param options Options for the create mirror operation
* @return The CreateMirrorResult
*/
CreateMirrorResult createMirror(String mirrorName, Map<String, String> configs, CreateMirrorOptions options);/**
* Create a new cluster mirror.
*
* @param mirrorName The name of the cluster mirror
* @param configs Configuration for the cluster mirror, including bootstrap servers and security settings
* @param options Options for the create mirror operation
* @return The CreateMirrorResult
*/
CreateMirrorResult createMirror(String mirrorName, Map<String, String> configs, CreateMirrorOptions options);
/**
* Start mirroring for the specified topics.
*
* When topics are started in a mirror, they become read-only on the destination cluster and start
* replicating data from the source cluster. This operation marks the specified topics with the
* mirror name, preventing local writes and enabling the MirrorFetcherThread to begin replication.
*
* @param mirrorName The cluster mirror name
* @param topics Set of topic names to start mirroring
* @param options Options for the start mirror topics operation
* @return The StartMirrorTopicsResult containing futures for each topic
*/
StartMirrorTopicsResult startMirrorTopics(String mirrorName, Set<String> topics, StartMirrorTopicsOptions options);
/**
* Options for {@link Admin#startMirrorTopics(String, Set, StartMirrorTopicsOptions)}.
*/
public class StartMirrorTopicsOptions extends AbstractOptions<StartMirrorTopicsOptions> {
private List<String> includePatterns = List.of();
private List<String> excludePatterns = List.of();
private Map<String, StartMirrorTopicsRequestData.TopicData> topicMetadata = Map.of();
public StartMirrorTopicsOptions includePatterns(List<String> patterns) {
this.includePatterns = patterns;
return this;
}
public StartMirrorTopicsOptions excludePatterns(List<String> patterns) {
this.excludePatterns = patterns;
return this;
}
public StartMirrorTopicsOptions topicMetadata(Map<String, StartMirrorTopicsRequestData.TopicData> metadata) {
this.topicMetadata = metadata;
return this;
}
public List<String> includePatterns() {
return includePatterns;
}
public List<String> excludePatterns() {
return excludePatterns;
}
public Map<String, StartMirrorTopicsRequestData.TopicData> topicMetadata() {
return topicMetadata;
}
}
/**
* Stop mirroring for the specified topics.
*
* This operation is typically used during failover scenarios when the destination cluster needs to
* be promoted from passive (read-only mirror) to active (accepting writes). Stopping mirror topics
* clears the mirrorName field from partition metadata, which allows producers to write
* to these partitions.
*
* @param mirrorName The cluster mirror name
* @param topics Set of topic names to stop mirroring
* @param options Options for the stop mirror topics operation
* @return The StopMirrorTopicsResult containing futures for each topic
*/
StopMirrorTopicsResult stopMirrorTopics(String mirrorName, Set<String> topics, StopMirrorTopicsOptions options);
/**
* Options for {@link Admin#stopMirrorTopics(String, Set, StopMirrorTopicsOptions)}.
*/
public class StopMirrorTopicsOptions extends AbstractOptions<StopMirrorTopicsOptions> {
private List<String> patterns = List.of();
public StopMirrorTopicsOptions patterns(List<String> patterns) {
this.patterns = patterns;
return this;
}
public List<String> patterns() {
return patterns;
}
}
/**
* Pause mirroring for the specified topics.
*
* Paused topics remain read-only on the destination cluster but stop fetching new data from the
* source cluster. The mirror fetcher threads are removed for these partitions, preserving the
* current replicated state. Mirroring can be resumed later with {@link #resumeMirrorTopics}.
*
* @param mirrorName The cluster mirror name
* @param topics Set of topic names to pause mirroring
* @param options Options for the pause mirror topics operation
* @return The PauseMirrorTopicsResult containing futures for each topic
*/
PauseMirrorTopicsResult pauseMirrorTopics(String mirrorName, Set<String> topics, PauseMirrorTopicsOptions options);
/**
* Resume mirroring for previously paused topics.
*
* Resumed topics restart fetching data from the source cluster, picking up from where they
* left off. New mirror fetcher threads are created and the partitions transition back to the
* MIRRORING state.
*
* @param mirrorName The cluster mirror name
* @param topics Set of topic names to resume mirroring
* @param options Options for the resume mirror topics operation
* @return The ResumeMirrorTopicsResult containing futures for each topic
*/
ResumeMirrorTopicsResult resumeMirrorTopics(String mirrorName, Set<String> topics, ResumeMirrorTopicsOptions options);
/**
* Delete a cluster mirror including its configuration.
*
* The mirror must be empty (no topics) or all its topics must have been removed (in STOPPED
* state). After deletion, all mirror metadata are tombstoned and failback is no longer possible.
*
* @param mirrorName The cluster mirror name
* @param options Options for the delete mirror operation
* @return The DeleteMirrorResult
*/
DeleteMirrorResult deleteMirror(String mirrorName, DeleteMirrorOptions options);
/**
* List the cluster mirrors available in the cluster.
*
* @param options The options to use when listing the mirrors.
* @return The ListMirrorsResult.
*/
ListMirrorsResult listMirrors(ListMirrorsOptions options);
/**
* Describe cluster mirrors.
*
* This operation retrieves detailed information about cluster mirrors including:
* - Topics being mirrored
* - Partition-level lag information (source offset vs destination offset)
* - Mirroring state for each partition (INITIALIZING, PREPARING, MIRRORING, etc.)
*
* @param mirrorNames The names of the mirrors to describe
* @param options The options to use when describing mirrors
* @return The DescribeMirrorsResult
*/
DescribeMirrorsResult describeMirrors(Collection<String> mirrorNames, DescribeMirrorsOptions options); |
This section describes all protocol level changes.
The CreateTopic API request is extended to add information required for mirror topic creation.
{ "name": "MirrorInfo", "type": "MirrorInfo", "versions": "8+", "nullableVersions": "8+", "ignorable": true,
"about": "Mirror information for creating a mirror topic from a source cluster.", "fields": [
{ "name": "TopicId", "type": "uuid", "versions": "8+",
"about": "The topic ID from the source cluster." }
]} |
The topic ID field ensures mirror topics retain the same topic ID as the source cluster topic. This allows fetch requests to pass validation on the source broker, and enables the system to verify that a topic being mirrored to a same-named topic in the destination cluster is indeed the same logical topic, not a name collision.
In normal topic creation, the MirrorInfo field will be null. When receiving the CreateTopic request, the controller will check the new field. If it is not set, the topic ID will be generated with random UUID as usual. Otherwise, the controller will do the following validation:
The Fetch API is extended to add the MirrorLeaderEpoch field used by the destination cluster internal replication.
// new request field in FetchPartition type
{ "name": "MirrorLeaderEpoch", "type": "int32", "versions": "19+", "default": "-1", "taggedVersions": "19+", "tag": 2, "ignorable": true,
"about": "The latest known mirror leader epoch." }
// new response field in PartitionData type
{ "name": "MirrorLeaderEpoch", "type": "int32", "versions": "19+", "default": "-1", "taggedVersions": "19+", "tag": 3, "ignorable": true,
"about": "The latest known mirror leader epoch." },
|
Allows users to create a mirror and supply its configuration. The broker validates that the mirror name is not already in use, contains only permitted characters, and does not end with .stopped or .paused suffix. Once validated, the request is forwarded to the controller, which persists the configuration in the metadata log.
{
"apiKey": TBD,
"type": "request",
"listeners": ["broker", "controller"],
"name": "CreateMirrorRequest",
// 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": "Config", "type": "[]MirrorConfig", "versions": "0+",
"about": "The cluster mirror configurations.", "fields": [
{ "name": "Name", "type": "string", "versions": "0+", "mapKey": true,
"about": "The configuration key name." },
{ "name": "Value", "type": "string", "versions": "0+", "nullableVersions": "0+",
"about": "The value to set for the configuration key."}
]}
]
}
{
"apiKey": TBD,
"type": "response",
"name": "CreateMirrorResponse",
// 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+",
"about": "The error message, or null if there was no error." }
]
} |
Start mirroring for the specified topics. The broker validates that all target topic partitions are in either UNKNOWN or STOPPED state; otherwise, the request is rejected. Once validated, the request is forwarded to the controller, which sets the mirror.name topic config to the specified mirror name.
{
"apiKey": TBD,
"type": "request",
"listeners": ["broker", "controller"],
"name": "StartMirrorTopicsRequest",
// 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": "NumPartitions", "type": "int32", "versions": "0+",
"about": "The number of partitions for the topic. Must match the source topic." }
]},
{ "name": "IncludePatterns", "type": "[]string", "versions": "0+",
"about": "Regex patterns to add to mirror.topics.include." },
{ "name": "ExcludePatterns", "type": "[]string", "versions": "0+",
"about": "Regex patterns to add to mirror.topics.exclude." }
]
}
{
"apiKey": TBD,
"type": "response",
"name": "StartMirrorTopicsResponse",
// 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." }
]}
]
} |
Stop mirroring for the specified topics. 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 .stopped suffix to the mirror.name topic config to mark the topics as no longer mirrored.
{
"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+",
"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." }
]}
]
} |
Pauses data replication and metadata sync for the specified mirror topics, keeping them read-only on the destination cluster.
{
"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+",
"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." }
]}
]
} |
Resumes data replication and metadata sync for previously paused mirror topics from where they left off.
{
"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+",
"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." }
]}
]
} |
Permanently deletes a cluster mirror, including its configuration. The mirror must be empty (no topics) or all its partitions must be in STOPPED state. After deletion, all metadata are tombstoned, making failback impossible. This is an irreversible operation.
{
"apiKey": TBD,
"type": "request",
"listeners": ["broker", "controller"],
"name": "DeleteMirrorRequest",
// Version 0 is the initial version.
"validVersions": "0",
"flexibleVersions": "0+",
"fields": [
{ "name": "MirrorName", "type": "string", "versions": "0+", "entityType": "mirrorName",
"about": "The cluster mirror name to delete."}
]
}
{
"apiKey": TBD,
"type": "response",
"name": "DeleteMirrorResponse",
// 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+",
"about": "The error message, or null if there was no error." }
]
} |
Returns the current mirror names and their associated topic counts in the cluster. It also includes source cluster ID and bootstrap server.
{
"apiKey": TBD,
"type": "request",
"listeners": ["broker"],
"name": "ListMirrorsRequest",
// Version 0 is the initial version.
"validVersions": "0",
"flexibleVersions": "0+",
"fields": []
}
{
"apiKey": TBD,
"type": "response",
"name": "ListMirrorsResponse",
// 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": "Mirrors", "type": "[]ListedMirror", "versions": "0+",
"about": "Each mirror in the response.", "fields": [
{ "name": "MirrorName", "type": "string", "versions": "0+", "entityType": "mirrorName",
"about": "The cluster mirror name." },
{ "name": "SourceBootstrap", "type": "string", "versions": "0+",
"about": "The source cluster bootstrap servers." },
{ "name": "SourceClusterId", "type": "string", "versions": "0+", "default": "",
"about": "The source cluster ID, or empty if not yet resolved." },
{ "name": "TopicCount", "type": "int32", "versions": "0+", "default": "0",
"about": "The number of topics configured for this mirror. 0 indicates an empty mirror with no topics." }
]}
]
} |
Returns the current mirroring status, state, and configuration for the specified mirror topics on the destination cluster. Allows destination cluster partition leaders to query the LME from the source cluster.
{
"apiKey": TBD,
"type": "request",
"listeners": ["broker"],
"name": "DescribeMirrorsRequest",
// Version 0 is the initial version.
"validVersions": "0",
"flexibleVersions": "0+",
"fields": [
{ "name": "MirrorNames", "type": "[]string", "versions": "0+", "entityType": "mirrorName",
"about": "The names of the mirrors to describe. Null or empty array means all mirrors." },
{ "name": "IncludeAuthorizedOperations", "type": "bool", "versions": "0+", "default": "false",
"about": "Whether to include authorized operations." }
]
}
{
"apiKey": TBD,
"type": "response",
"name": "DescribeMirrorsResponse",
// 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": "Mirrors", "type": "[]DescribedMirror", "versions": "0+",
"about": "Each described mirror.", "fields": [
{ "name": "ErrorCode", "type": "int16", "versions": "0+",
"about": "The error code, or 0 if there was no error." },
{ "name": "MirrorName", "type": "string", "versions": "0+", "entityType": "mirrorName",
"about": "The cluster mirror name." },
{ "name": "AuthorizedOperations", "type": "int32", "versions": "0+", "default": "-2147483648",
"about": "32-bit bitfield to represent authorized operations for this mirror." },
{ "name": "Topics", "type": "[]TopicPartitions", "versions": "0+",
"about": "Each topic in the mirror.", "fields": [
{ "name": "TopicName", "type": "string", "versions": "0+",
"about": "The topic name." },
{ "name": "Partitions", "type": "[]PartitionDetail", "versions": "0+",
"about": "Each partition detail.", "fields": [
{ "name": "PartitionIndex", "type": "int32", "versions": "0+",
"about": "The partition index." },
{ "name": "SourceOffset", "type": "int64", "versions": "0+", "default": "-1",
"about": "The high watermark offset from the source cluster leader, or -1 if not yet available." },
{ "name": "DestinationOffset", "type": "int64", "versions": "0+", "default": "-1",
"about": "The log end offset on the destination cluster, or -1 if not yet available." },
{ "name": "Lag", "type": "int64", "versions": "0+", "default": "-1",
"about": "The lag (source offset - destination offset), or -1 if not yet available." },
{ "name": "State", "type": "string", "versions": "0+",
"about": "The partition state." },
{ "name": "", "type": "int32", "versions": "0+", "default": "-1",
"about": "The last mirror leader epoch, or -1 if not available." }
]}
]}
]}
]
} |
Internal API that sets a minimum leader epoch on the specified partitions. The controller increments each partition's leader epoch to at least the requested value.
{
"apiKey": TBD,
"type": "request",
"listeners": ["broker", "controller"],
"name": "BumpLeaderEpochsRequest",
// Version 0 is the initial version.
"validVersions": "0",
"flexibleVersions": "0+",
"fields": [
{ "name": "Topics", "type": "[]TopicState", "versions": "0+", "about": "The topic and partitions state.",
"fields": [
{"name": "TopicId", "type": "uuid", "versions": "0+", "about": "The unique topic ID."},
{ "name": "Partitions", "type": "[]LeaderEpochState", "versions": "0+", "about": "The partition leader epochs.",
"fields": [
{"name": "partitionIndex", "type": "int32", "versions": "0+", "about": "The partition index."},
{"name": "minLeaderEpoch", "type": "int32", "versions": "0+", "default": -1, "about": "The minimum leader epoch that the destination cluster should bump to."}
]}
]}
]
}
{
"apiKey": TBD,
"type": "response",
"name": "BumpLeaderEpochsResponse",
// 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": "Topics", "type": "[]TopicPartitions", "versions": "0+",
"about": "Each topic in the mirror.", "fields": [
{ "name": "TopicName", "type": "string", "versions": "0+",
"about": "The topic name." },
{ "name": "Partitions", "type": "[]PartitionDetail", "versions": "0+",
"about": "Each partition state.", "fields": [
{ "name": "PartitionIndex", "type": "int32", "versions": "0+",
"about": "The partition index." },
{ "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 __mirror_state topic on the destination cluster.
{
"apiKey": TBD,
"type": "request",
"listeners": ["broker", "controller"],
"name": "ReadMirrorStatesRequest",
// 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": "Name", "type": "string", "versions": "0", "entityType": "topicName",
"about": "The topic name." },
{ "name": "Partitions", "type": "[]PartitionData", "versions": "0",
"about": "The data for the partitions.", "fields": [
{ "name": "PartitionIndex", "type": "int32", "versions": "0",
"about": "The partition index." }
]}
]}
]
}
{
"apiKey": TBD,
"type": "response",
"name": "ReadMirrorStatesResponse",
// 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": "Topics", "type": "[]TopicResult", "versions": "0",
"about": "The read results for the topics.", "fields": [
{ "name": "Name", "type": "string", "versions": "0", "entityType": "topicName",
"about": "The topic name." },
{ "name": "Partitions", "type": "[]PartitionResult", "versions": "0",
"about": "The results for the partitions.", "fields": [
{ "name": "PartitionIndex", "type": "int32", "versions": "0",
"about": "The partition index." },
{ "name": "LastMirrorEpoch", "type": "int32", "versions": "0", default": "-1",
"about": "The last mirror leader epoch, or -1 if not available." },
{ "name": "State", "type": "int8", "versions": "0+",
"about": "The mirror partition state." },
{ "name": "ErrorCode", "type": "int16", "versions": "0",
"about": "The error code, or 0 if there was no error." }
]}
]}
]
} |
Internal API that persists mirror partition state transitions to the internal __mirror_state topic on the destination cluster.
{
"apiKey": TBD,
"type": "request",
"listeners": ["broker", "controller"],
"name": "WriteMirrorStatesRequest",
// Version 0 is the initial version.
"validVersions": "0",
"flexibleVersions": "0+",
"fields": [
{ "name": "MirrorName", "type": "string", "versions": "0+", "entityType": "mirrorName",
"about": "The mirror name." },
{ "name": "Topics", "type": "[]TopicData", "versions": "0",
"about": "The data for the topics.", "fields": [
{ "name": "Name", "type": "string", "versions": "0", "entityType": "topicName",
"about": "The topic name." },
{ "name": "Partitions", "type": "[]PartitionData", "versions": "0",
"about": "The data for the partitions.", "fields": [
{ "name": "PartitionIndex", "type": "int32", "versions": "0",
"about": "The partition index." },
{ "name": "", "type": "int32", "versions": "0", "default": "-1",
"about": "The last mirror leader epoch, or -1 if not available." },
{ "name": "State", "type": "int8", "versions": "0+",
"about": "The mirror partition state." }
]}
]},
{ "name": "StoppedTopics", "type": "[]string", "versions": "0+", "about": "The topic names to be stopped." }
]
}
{
"apiKey": TBD,
"type": "response",
"name": "WriteMirrorStatesResponse",
// 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": "Topics", "type": "[]TopicResult", "versions": "0",
"about": "The write results for the topics.", "fields": [
{ "name": "Name", "type": "string", "versions": "0", "entityType": "topicName",
"about": "The topic name." },
{ "name": "Partitions", "type": "[]PartitionResult", "versions": "0",
"about": "The results for the partitions.", "fields": [
{ "name": "PartitionIndex", "type": "int32", "versions": "0",
"about": "The partition index." },
{ "name": "ErrorCode", "type": "int16", "versions": "0",
"about": "The error code, or 0 if there was no error." }
]}
]}
]
} |
This section describes records written to the KRaft metadata log by the active controller as part of Cluster Mirroring operations.
Written by the controller when processing a BumpLeaderEpochs request. The record carries a minLeaderEpoch field that sets a floor for the partition's leader epoch.
{
"apiKey": 5,
"type": "metadata",
"name": "PartitionChangeRecord",
"validVersions": "0-3",
"flexibleVersions": "0+",
"fields": [
// ... existing fields ...
{"name": "MinLeaderEpoch", "type": "int32", "versions": "3+", "default": -1,
"about": "The minimum leader epoch requested."}
]
} |
A control record (type MIRROR_PID_RESET) written to each partition's data log during the STOPPING transition.
{
"type": "data",
"name": "MirrorPidResetRecord",
"validVersions": "0",
"flexibleVersions": "0+",
"fields": [
{ "name": "Version", "type": "int16", "versions": "0",
"about": "The version of the mirror PID reset record."},
{ "name": "SourceClusterId", "type": "string", "versions": "0",
"about": "The source cluster UUID for verification."}
]
} |
This section describes records written to the __mirror_state internal topic by the MirrorCoordinator to track mirror state and synchronization points across brokers.
Written during the STOPPING transition to record the last mirrored leader epoch for each partition before the destination becomes writable.
{
"apiKey": 1,
"type": "coordinator-key",
"name": "LastMirrorEpochsKey",
"validVersions": "0",
"flexibleVersions": "none",
"fields": [
{ "name": "MirrorName", "type": "string", "versions": "0",
"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." }
]}
]}
]
} |
Written on every state transition. Tracks the current state of each mirrored partition.
{
"apiKey": 2,
"type": "coordinator-key",
"name": "MirrorPartitionStateKey",
"validVersions": "0",
"flexibleVersions": "none",
"fields": [
{ "name": "MirrorName", "type": "string", "versions": "0",
"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." }
]
} |
This section lists the new values added to existing Kafka type enumerations to support Cluster Mirroring
Represents a mirror as a configurable resource in cluster metadata. Mirror-level properties such as source cluster bootstrap servers, security credentials, mirror.topics.include, and mirror.topics.exclude are stored under this type, keyed by mirror name.
public enum Type {
// existing types unchanged
MIRROR((byte) 64, "mirror");
} |
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 EntityType {
// existing types unchanged
@JsonProperty("mirrorName")
MIRROR_NAME(FieldType.StringFieldType.INSTANCE);
} |
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.
public enum ResourceType {
// existing types unchanged
CLUSTER_MIRROR((byte) 8); |
A 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.
public enum CoordinatorType {
// existing types unchanged
MIRROR((byte) 3);
} |
This section describes new configurations introduced by Cluster Mirroring.
Set via broker config. Stored in server.properties or dynamic broker config.
Key | Description | Default |
|---|---|---|
mirror.topic.num.partitions | Number of partitions for __mirror_state internal topic. | 50 |
mirror.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. |
Set via CreateMirror or IncrementalAlterConfigs. Stored in cluster metadata records.
Key | Description | Default |
|---|---|---|
bootstrap.servers | A list of host/port pairs to use for establishing the initial connection to the source cluster. | |
mirror.topic.properties.exclude | A comma-separated list of topic config property names to exclude from synchronization. Properties in this list will not be replicated from the source cluster. The mirror.name property is always excluded regardless of this setting. | follower.replication.throttled.replicas, leader.replication.throttled.replicas, message.timestamp.difference.max.ms, log.message.timestamp.before.max.ms, log.message.timestamp.after.max.ms, message.timestamp.type, unclean.leader.election.enable, min.insync.replicas, mirror.name |
mirror.topics.include | A comma-separated list of regex patterns for topic names to include in mirroring. Topics on the source cluster whose names match at least one of the patterns will be automatically discovered and mirrored.
| |
mirror.topics.exclude | A comma-separated list of regex patterns for topic names to exclude from mirroring. Topics matching the exclude pattern are not mirrored even if they match mirror.topics.include. Internal topics are always excluded. Exclude always wins over include. By default internal topics (starting with __ ) are excluded. | |
mirror.groups.include | A comma-separated list of regex patterns for group IDs to include in offset synchronization. Only groups whose IDs match at least one of the patterns will have their offsets replicated from the source cluster. | |
| mirror.groups.exclude | A comma-separated list of regex patterns for group IDs to exclude from offset synchronization. Groups matching the exclude pattern are not replicated even if they match mirror.groups.include. | |
mirror.acl.include | A comma-separated list of ACL include rules. Each rule uses semicolon-separated fields: resourceType;resourceName;operation;permissionType;principal. Use '*' as wildcard for any field. The resourceName field supports regex patterns. Trailing wildcard fields can be omitted. See AclRule javadoc for examples. Examples: TOPIC;orders.* (all ACLs for topics matching orders.*) *;*;*;*;User:alice (all ACLs for principal User:alice) *;*;*;*;User:app-.* (all ACLs for principals matching User:app-.*) TOPIC;*;READ;ALLOW (all topic READ/ALLOW ACLs) GROUP;consumer-.*;READ;ALLOW;User:bob (READ/ALLOW ACLs on groups matching consumer-.* for User:bob) TOPIC;orders.*,*;*;*;*;User:alice (sync all topic ACLs for orders.* topics and all ACLs for User:alice) | * |
security.protocol | Protocol for source cluster communication (PLAINTEXT, SSL, SASL_PLAINTEXT, SASL_SSL). | |
sasl.* | SASL configuration properties. | |
ssl.* | SSL configuration properties. |
Set by topic creation or alter. Stored in topic config.
Key | Description | Default |
|---|---|---|
mirror.name | Identifies the mirror that manages this topic. Topics with this configuration set are read-only and can only be modified through mirror management APIs. This property is filtered out from DescribeConfigs responses to avoid exposing internal state to users. | |
mirror.replication.throttled.replicas | A list of replicas for which log replication should be throttled on the mirror follower node. The list should describe a set of replicas in the form [PartitionId]:[BrokerId],[PartitionId]:[BrokerId]:... or alternatively the wildcard '*' can be used to throttle all replicas for this topic." | |
mirror.support.unclean.leader.election | When enabled, LME log truncation waits for all replicas (not just ISR members) to join the ISR and complete the truncation. | false |
A core set of metrics will be provided with the initial implementation.
Name | Type | Group | Tags | Description | JMX Bean |
|---|---|---|---|---|---|
MaxLag | MirrorFetcherManager | kafka.server.mirror | clientId=MirrorReplica | Max lag in messages between destination leader and source leader replicas. | kafka.server.mirror:type=MirrorFetcherManager,name=MaxLag,clientId=MirrorReplica |
MinFetchRate | MirrorFetcherManager | kafka.server.mirror | clientId=MirrorReplica | The min fetch rate between destination leader and source leader replicas. | kafka.server.mirror:type=MirrorFetcherManager,name=MirrorReplica |
ConsumerLag | FetcherLagMetrics | kafka.server | clientId=MirrorFetcherThread-{sourceBroker.id}-{fetcherId}-{mirrorName},topic=([-.\w]+),partition=([0-9]+) | Lag in messages per remote leader replica. | kafka.serverr:type=FetcherLagMetrics,name=ConsumerLag,clientId=MirrorFetcherThread-{sourceBroker.id}-{fetcherId}-{mirrorName},topic=([-.\w]+),partition=([0-9]+) |
DeadThreadCount | MirrorFetcherManager | kafka.server.mirror | clientId=MirrorReplica | Number of dead mirror fetcher threads. | kafka.server,mirror:type=MirrorFetcherManager,name=DeadThreadCount,clientId=MirrorReplica |
FailedPartitionsCount | MirrorFetcherManager | kafka.server.mirror | clientId=MirrorReplica | Total count for failed partitions for any reason like auth, authorization, failed network with source. | kafka.serve.mirror:type=MirrorFetcherManager,name=FailedPartitionsCount,clientId=MirrorReplica |
BytesPerSec | FetcherStats | kafka.server | clientId=MirrorFetcherThread-{sourceBroker.id}-{fetcherId}-{mirrorName},brokerHost={host},brokerPort={port} | Extend kafka.server.FetcherStats to report mirror fetcher threads. | kafka.server:type=FetcherStats,name=BytesPerSec,clientId=MirrorFetcherThread-{sourceBroker.id}-{fetcherId}-{mirrorName},brokerHost={host},brokerPort={port},mirror-name={mirrorName} |
RequestsPerSec | FetcherStats | kafka.server | clientId=MirrorFetcherThread-{sourceBroker.id}-{fetcherId}-{mirrorName},brokerHost={host},brokerPort={port} | Extend kafka.server.FetcherStats to report mirror fetcher threads. | kafka.server:type=FetcherStats,name=RequestsPerSec,cclientId=MirrorFetcherThread-{sourceBroker.id}-{fetcherId}-{mirrorName}, brokerHost={host},brokerPort={port},mirror-name={mirrorName} |
LocalTimeMs, MessageConversionsTimeMs, RemoteTimeMs, RequestBytes, RequestQueueTimeMs, ResponseQueueTimeMs, ResponseSendTimeMs, TemporaryMemoryBytes, TotalTimeMs | RequestMetrics | kafka.network | request=[mirror_requests] | Extend kafka.network:type=RequestMetrics to list cluster mirror requests. | kafka.network:type=RequestMetrics,name=*, request=* |
ErrorsPerSec | RequestMetrics | kafka.network | request=[mirror_requests],error=* | Extend kafka.network:type=RequestMetrics to list cluster mirror requests. | kafka.network:type=RequestMetrics,name=ErrorsPerSec, request=*, error=* |
RequestsPerSec | RequestMetrics | kafka.network | request=[mirror_requests],version=* | Extend kafka.network:type=RequestMetrics to list cluster mirror requests. | kafka.network:type=RequestMetrics,name=RequestsPerSec, request=*, version=* |
connection-close-rate, connection-close-total, connection-count, connection-creation-rate, connection-creation-total, failed-authentication-rate, failed-authentication-total, failed-reauthentication-rate, failed-reauthentication-total, incoming-byte-rate, incoming-byte-total, network-io-rate, network-io-total, outgoing-byte-rate, outgoing-byte-total, reauthentication-latency-avg, reauthentication-latency-max, request-rate, request-size-avg, request-size-max, request-total, response-rate, response-total, select-rate, select-total, successful-authentication-no- reauth-total, successful-authentication-rate, successful-authentication-total, successful-reauthentication-rate, successful-reauthentication-total | mirror-broker-{DestinationBroker.id}-fetcher-{fetcherId}-mirror-{mirrorName}-metrics | kafka.server | broker-id={sourceBroker.id},fetcher-id={fetcherId} | Fetcher requests in the cluster mirror metrics. | kafka.server:type=mirror-broker-{sourceBroker.id}-fetcher-{fetcherId}-mirror-{mirrorName}-metrics,broker-id={sourceBroker.id},fetcher-id={fetcherId} |
MetadataRefreshError | MirrorMetadataManager | kafka.server.mirror | Number of topic metadata refresh sync errors. | kafka.server.mirror:type=MirrorMetadataManager,name=aclSyncError | |
TopicConfigMetadataSyncError | MirrorMetadataManager | kafka.server.mirror | Number of topic configuration sync errors. | ||
ConsumerGroupOffsetSyncError | MirrorMetadataManager | kafka.server.mirror | Number of CGs sync errors. | ||
ShareGroupOffsetSyncError | MirrorMetadataManager | kafka.server.mirror | Number of SGs sync errors. | ||
AclSyncError | MirrorMetadataManager | kafka.server.mirror | Number of ACLs sync errors. | kafka.server.mirror:type=MirrorMetadataManager,name=aclSyncError | |
ByteRate | MirrorReplication | kafka.server | Bandwidth quota metrics. Indicates the throttled data mirror replication rate of the broker in bytes/sec. | kafka.server:type=MirrorReplication | |
FailedPartitionState | MirrorMetadataManager | kafka.server.mirror | Number of partitions in failed state. | kafka.server.mirror:type=MirrorMetadataManager,name=FailedPartitionState | |
StoppedPartitionState | MirrorMetadataManager | kafka.server.mirror | Number of partitions in a stopped state. | kafka.server.mirror:type=MirrorMetadataManager,name=StoppedPartitionState | |
StoppingPartitionState | MirrorMetadataManager | kafka.server.mirror | Number of partitions in stopping state. | kafka.server.mirror:type=MirrorMetadataManager,name=StoppingPartitionState | |
MirroringPartitionState | MirrorMetadataManager | kafka.server.mirror | Number of partitions in mirroring state. | kafka.server.mirror:type=MirrorMetadataManager,name=MirroringPartitionState | |
PreparingPartitionState | MirrorMetadataManager | kafka.server.mirror | Number of partitions in preparing state. | kafka.server.mirror:type=MirrorMetadataManager,name=PreparingPartitionState |
List of protocol-level errors returned by the new RPCs:
| Code | Name | Message | Used By |
|---|---|---|---|
| 3 | UNKNOWN_TOPIC_OR_PARTITION | The topic does not exist on the target cluster | StopMirrorTopics, PauseMirrorTopics, ResumeMirrorTopics |
| 15 | COORDINATOR_NOT_AVAILABLE | The mirror coordinator is not active | WriteMirrorStates, ReadMirrorStates |
| 31 | CLUSTER_AUTHORIZATION_FAILED | The client is not authorized to perform the mirror operation | WriteMirrorStates, ReadMirrorStates |
| 35 | UNSUPPORTED_VERSION | Cluster mirroring is disabled (mirror.version=0) | CreateMirror, StartMirrorTopics, StopMirrorTopics, PauseMirrorTopics, ResumeMirrorTopics, ListMirrors, DescribeMirrors, DeleteMirror |
| TBD | READ_ONLY_TOPIC | The topic is read-only because it is a mirror topic on the target cluster | Produce |
| TBD | INVALID_MIRROR_NAME | The mirror name does not meet the naming rules | CreateMirror |
| TBD | MIRROR_ALREADY_EXISTS | The mirror already exists | CreateMirror |
| TBD | UNKNOWN_MIRROR | The topic is not assigned to any mirror | StopMirrorTopics, PauseMirrorTopics, ResumeMirrorTopics |
| TBD | TOPIC_ALREADY_IN_MIRROR | The topic is already assigned to a mirror | StartMirrorTopics |
| TBD | TOPIC_NOT_IN_MIRROR | The topic does not belong to the specified mirror | StopMirrorTopics, PauseMirrorTopics, ResumeMirrorTopics |
| TBD | MIRROR_TOPIC_ALREADY_PAUSED | The mirror topic is already paused | PauseMirrorTopics |
| TBD | MIRROR_TOPIC_NOT_PAUSED | The mirror topic is not paused | ResumeMirrorTopics |
| TBD | MIRROR_TOPIC_BEING_STOPPED | The mirror topic is being stopped | ResumeMirrorTopics |
| TBD | MIRROR_NOT_EMPTY | The mirror still has active or non-removed topics | DeleteMirror |
| TBD | MIRROR_AUTHORIZATION_FAILED | Mirror authorization failed | CreateMirror, StartMirrorTopics, StopMirrorTopics, PauseMirrorTopics, ResumeMirrorTopics, DeleteMirror |
Cluster Mirroring will be introduced through a phased rollout across multiple Kafka releases to ensure stability and gather community feedback.
Cluster Mirroring is introduced as an early access feature, disabled by default to prevent accidental production usage. To enable it, all cluster nodes (controllers and brokers) must explicitly enable unstable API versions (unstable.api.versions.enable=true) and unstable feature versions (unstable.feature.versions.enable=true) in all configuration files. After starting the cluster with a minimum metadata version, operators can dynamically enable the mirror version feature to activate Cluster Mirroring (bin/kafka-features.sh --bootstrap-server :9092 upgrade --feature mirror.version=1). This stage is intended for testing and evaluation in non-production environments only, as the new APIs and metadata record formats may change in subsequent releases without backward compatibility guarantees.
In a future release, Cluster Mirroring will transition to preview status with frozen protocol and metadata schemas. The feature will still require explicit enablement via dynamic feature upgrades but will no longer require the unstable API and feature configuration. The feature remains disabled by default to ensure operators consciously opt-in, but the upgrade path from early access clusters will be officially supported with compatibility guarantees. This stage is suitable for pre-production testing and pilot deployments where API stability is required but production-grade maturity is not yet needed.
When Cluster Mirroring reaches general availability, the feature will be enabled by default when clusters reach the corresponding production metadata version. All new APIs will become stable production APIs with all unstable markers removed from their definition. No special configuration flags or explicit feature enablement will be required beyond setting an appropriate metadata version, and the feature will be fully supported for mission-critical production workloads under Kafka's standard compatibility guarantees. Clusters using Cluster Mirroring in preview can upgrade seamlessly to GA releases without migration steps. Downgrade is also supported, but it would require manual cleanup of the internal topic.
Cluster Mirror is not compatible with MirrorMaker 2. This is a critical consideration for users planning to migrate from MirrorMaker 2 to Cluster Mirroring.
MM2 and Cluster Mirroring use different internal topic structures and naming conventions for storing metadata and offsets. The two systems track and store consumer offsets differently, making it impossible to seamlessly transition between them.
Follow this process to switch from MirrorMaker 2 to Cluster Mirroring:
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. |
MirrorFetcherThread uses the same fetch protocol optimizations as ReplicaFetcherThread:
Cluster Mirroring introduces additional replication threads and network I/O on brokers configured as read-only leaders for mirror partitions. The performance impact on existing intra-cluster replication is minimized through resource isolation:
Unit tests will cover individual component behavior:
Integration tests will validate end-to-end functionality across multiple brokers:
System tests will validate behavior under realistic production conditions:
This KIP introduces native cluster mirroring to address the limitations of MirrorMaker 2 described in the motivation section. The following tables provide a detailed comparison across deployment, features, and performance characteristics.
Use Case Guidance:
Deployment Comparison
| MirrorMaker 2 | Cluster Mirroring | |
|---|---|---|
| Architecture | External Connect workers (separate JVM) | Integrated into Kafka brokers using native replication protocol |
| Operational Complexity | Manage Connect cluster lifecycle independently | Unified with broker operations |
| Monitoring | Separate Connect metrics and dashboards | Standard Kafka JMX metrics |
Feature Support Comparison
| MirrorMaker 2 | Cluster Mirroring | |
|---|---|---|
Offset Translation | Lossy, requires remapping, causes reprocessing overhead | None needed, offsets preserved exactly |
Metadata Sync | Requires separate connector configuration (MirrorSourceConnector, MirrorCheckpointConnector) | Automatic (topics, configs, consumer groups, ACLs) |
Transactional Topics | Markers copied as regular records, incomplete transactions possible during replication | Markers mirrored. Inflight transactions are automatically aborted. |
Topic Write Protection | Not supported (mirror topics always writable) | Read-only enforcement during mirroring, writable only after explicit failover |
Tiered Storage | Fetches from broker (which reads from remote storage) | Not initially supported (future work) |
Active-Active | Supported via topic prefixing and cycle detection | Not supported (read-only enforcement prevents cycles) |
Share Groups | Not supported | Supported |
Failback | Full re-mirror from offset 0 | Delta sync using DescribeMirror |
Topic Name Preservation | No, destination topics prefixed with source cluster alias (e.g., source.topic-name) | Yes, same topic name as source |
Topic ID Preservation | No, destination gets new topic ID | Yes: same topic ID as source |
Performance & Isolation Comparison
| MirrorMaker 2 | Cluster Mirroring | |
|---|---|---|
Compression Overhead | Decompress + recompress records | Preserve source compression (zero overhead) |
Failover Time | Offset translation + consumer group sync + bootstrap reconfiguration | Consumer group sync + bootstrap reconfiguration only |
Bandwidth Control (Source) | None (unless client quotas manually configured) | Client quotas (current), dedicated mirror throttling (future) |
Bandwidth Control (Destination) | None (unless client quotas manually) | Replica-level throttling (mirror.replication.throttled.rate) |
Resource Isolation (Source) | Shares network bandwidth and disk I/O with all consumers | Shares network bandwidth and disk I/O with all consumers (same - current); isolated on replica-level (future) |
Resource Isolation (Destination | Separate JVM heap (memory isolated from brokers) but still consume resources from the broker as any client | Shares network, disk I/O; dedicated thread pool within broker JVM |
Malformed Batch Handling | Crashes Connect task, all partitions in task restart | Fails individual partition (FAILED state), others continue |
Blast Radius(Failure) | All partitions in task affected | Single partition affected |
Catch-Up Surge Protection | Connect worker heap may OOM, affects all tasks | Throttling on destination replica + source quota prevent memory spikes |
Unclean Leader Election | Not supported | Supported with extra configuration and latency |