Authors: Luke Chen, Federico Valeri, Omnia Ibrahim, PoAn Yang, Kuan-Po Tseng, Jiunn-Yang Huang

Status

Current state: Under Discussion

Discussion thread: here

Voting thread: here

JIRA: KAFKA-20186 - Getting issue details... STATUS

Please keep the discussion on the mailing list rather than commenting on the wiki (wiki discussions get unwieldy fast).

Motivation

Cluster Mirroring follows and simplifies the design goals of KIP-986: Cross-Cluster Replication by Greg Harris, whose high level proposal provided the foundational vision for this feature.

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 shortcomings and operational challenges:

• Operational Burden: MM2 runs as standalone Connect workers external to Kafka brokers, requiring separate deployment, monitoring, and lifecycle management. Operators must provision additional hosts, manage Connect-specific configurations, and coordinate MM2 upgrades independently from Kafka broker upgrades.
• Compression Inefficiency: If source cluster records are compressed, MM2 will decompress and compress them again when producing to the destination cluster. These redundant operations decrease the mirroring throughput and increase latency.
• Lossy Offset Translation: The offset translation process in MM2 is inherently lossy. When translating an offset from the source cluster to the target cluster, MM2 cannot guarantee returning the exact same record. It is not possible to maintain a complete in-memory mapping of source to destination offsets for all mirrored records. When an exact translation is unavailable, MM2 guarantees that the record at the translated offset is always earlier than the actual record, ensuring consuming applications never skip data at the cost of potential reprocessing. This conservative approach can lead to significant duplicate processing during failover scenarios, particularly for high-throughput topics where offset translation granularity is coarse.
• External Offset Complexity: Advanced streaming platforms like Apache Flink and Apache Spark, Kafka Connect source connectors, and transactional applications often store consumed offsets externally rather than in Kafka's __consumer_offsets topic. When a failover happens, these applications face additional complexity because source and destination offsets don't match. Applications must query MM2's offset-sync internal topic to translate offsets, adding operational complexity and potential failure points. This offset translation dependency complicates DR procedures and increases the risk of incorrect offset mapping leading to data loss or duplicate processing.
• Unclean Leader Election: MM2 does not handle unclean leader elections on the source cluster gracefully. When an unclean leader election occurs, data loss on the source may not be reflected on the destination, leading to divergent state between clusters.

Cluster Mirroring addresses these issues by integrating cross-cluster replication directly into Kafka brokers, providing a simpler and more robust solution for cross-cluster replication. Producers send messages 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). Each topic partition's data is then asynchronously replicated to the destination cluster with no impact on producer latency or throughput. This design provides the following properties:

• Integrated Architecture: The replication logic runs within broker processes, eliminating external dependencies and reducing the operational footprint.
• Simplified Configuration: Creating a cluster mirror requires a single command-line invocation or Admin API call with bootstrap servers and security credentials.
• Metadata Synchronization: Topic configurations, consumer group offsets, and ACLs are periodically synchronized from source to destination cluster without additional configuration.
• Unified Monitoring: Metrics are exposed through standard Kafka broker JMX metrics alongside existing replication metrics. Operators use familiar tools and dashboards to monitor cross-cluster replication.
• Faster Failover: The failover operation is simplified because metadata synchronization is continuous and automatic. Consumer applications can resume processing immediately after switching clusters without any offset translation.
• Incremental Failback: The destination leader fetches from its local log end offset to catch up with the source leader, so only the delta is replicated when failing back (reverse mirroring).
• Version Compatibility: For migration or DR use cases where failback is not required, this proposal supports source brokers up to version 2.1 included, leveraging the client/broker forward compatibility introduced in version 4.0.
• ULE Support: With additional configuration, it is possible to detect and reconcile unclean leader elections ensuring consistency across clusters.

Cluster Mirroring addresses two operational scenarios that are difficult to solve with existing Kafka primitives: keeping a standby cluster in sync for disaster recovery, and moving data between clusters during major version upgrades:

• Disaster Recovery: Asynchronous replication provides the right balance for DR use cases where availability and performance of the primary cluster must not be compromised. 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. Organizations must handle a non-zero RPO determined by the replication lag, which typically ranges from seconds to minutes depending on network bandwidth, throughput, and geographic distance. Applications requiring zero data loss can wait for the follow-up KIP that will extend this design to support synchronous cross-cluster replication, or handle the lag using application-level caching. Stretched clusters are not suitable for DR scenarios because they provide no protection against software failures, configuration errors, and accidental topic deletions.
• Cluster Migration: Upgrading a Kafka cluster across multiple major versions requires careful review of every intermediate release note to verify configuration compatibility, deprecated features, and breaking changes. Migrating from ZooKeeper to KRaft adds further complexity, as the cluster must first reach a bridge release that supports both modes before performing an in-place metadata migration. Executing these multi-step upgrades in production without strong automation is risky and operationally demanding. Cluster Mirroring offers a simpler alternative: stand up a new cluster running the target version with KRaft, mirror the data from the old cluster preserving offsets and topic UUIDs, and perform a controlled failover once replication has caught up. This eliminates the need for intermediate upgrades and avoids in-place metadata migration entirely.

Proposed Changes

A cluster mirror is a named, unidirectional replication channel from a remote source cluster to the local destination cluster. It is defined by a mirror name, the source cluster's bootstrap servers, and security credentials. These properties are stored as a CLUSTER_MIRROR config resource in the metadata log, similar to how broker or topic configs are stored. Once created, individual topics can be started, stopped, paused, or resumed for replication within the mirror. Each of these operations writes a config record to the metadata log that sets the mirror name and desired state (active, stopped, or paused) on the topic. Brokers receive this through the standard metadata update mechanism, and the mirror name and desired state become part of the TopicImage in the metadata cache. A dedicated component reacts to these changes in its onMetadataUpdate callback, initiating the appropriate state transitions. Cluster mirrors are managed through the Admin API and the kafka-cluster-mirrors.sh CLI tool.

Architecture

The architecture consists of three main components on each broker, each with a distinct role. MirrorMetadataManager reacts to KRaft metadata updates and periodically syncs metadata from source clusters (topic state, configs, group offsets, ACLs). ClusterMirrorCoordinator owns the partition state machine, persisting state transitions to __mirror_state and driving the replication lifecycle (truncation, epoch fencing, failover). MirrorFetcherManager creates and manages MirrorFetcherThread instances that replicate data from source brokers, copying compressed batches byte for byte without decompression. The following diagram illustrates how these components are wired together; some internal APIs are excluded for clarity.

ClusterMirrorCoordinator

The ClusterMirrorCoordinator (CMC) manages mirror partition state trasitions using a partitioned coordinator pattern similar to the group and transaction coordinators. This state is persisted in __mirror_state, an internal compacted topic.

Two hashing schemes distribute work across this topic's partitions. Partition state records use a composite key (mirror name, topic id, partition number), so a mirror with hundreds of partitions spreads its state management evenly across the cluster rather than concentrating it on a single broker. Periodic source sync (topic configs, group offsets, ACLs, pattern discovery) hashes by mirror name alone, meaning one broker coordinates all source sync for a given mirror regardless of how many partitions it has.

Each broker leads a subset of __mirror_state partitions and coordinates the mirror partitions that hash to them. When a broker gains leadership of a __mirror_state partition, it replays the log to rebuild its in-memory cache. When it loses leadership, the cache is cleared. Writes targeting a remote coordinator are forwarded.

Mirror Partition States

State Transition Validation

Each state-change RPC includes a StateValidationOffset field used for an optimistic locking guard that prevents the operation (start, stop, pause, resume, delete) from going through if a mirror's topic state changed between the moment the broker did its validation (preconditions) and the moment the controller processes the operation.  Mirror partition state transitions involve two separate nodes: a broker and the controller. The broker validates preconditions against the mirror coordinator, then sends a request to the controller, which writes the actual metadata records. There is a window between the broker's check and the controller's write where another operation can sneak in and change the state, making the original operation invalid. The StateValidationOffset field contains the cluster metadata offset before any validation. It represents the metadata snapshot against which the broker made its decision. If the controller detects that the MirrorTopicStateChangeRecord offset is strictly greater than StateValidationOffset, it rejects the operation with InvalidClusterMirrorStatesException.

Incremental Failback

During failback (B->A), the requesting cluster sends its own cluster ID (A) in the DescribeClusterMirrors request. The receiving cluster matches it against the source cluster ID stored in its mirror configs at creation time, identifies the original mirror (A->B), and returns the truncation epoch for the requested partitions, that was stored in the coordinator's metadata log when the mirror was stopped. For each partition, the requesting cluster truncates its log to the last offset of that epoch. If the API is not supported or no truncation epoch is found, the broker truncates to zero and mirrors from scratch. Before transitioning from LOG_TRUNCATION to MIRRORING state, the coordinator waits until all ISR members have completed truncation. If fewer than min ISR are available, the check is retried. This prevents the mirror leader from appending new data while followers still hold divergent log segments. Note that log truncation on failback may cause data loss if there are records that were not mirrored to the old destination cluster

MirrorMetadataManager

The MirrorMetadataManager (MMM) runs on every broker and it does serve those two purposes:

1. Periodically sync metadata from the source cluster. Source topic state sync (leader caches, deletion detection, missed partition recovery) runs on every broker. Topic creation, partition scaling, and coordinator-level sync (topic configs, consumer group offsets, ACLs, topic discovery by pattern) run only on the broker that leads the __mirror_state partition determined by hashing the mirror name.

2. Trigger mirror partition state transitions. When a KRaft metadata update arrives, each broker collects only the partitions where it gained or holds leadership and triggers state transitions for those partitions. State writes are routed to the coordinator that leads the relevant __mirror_state partition for the composite key (mirror name, topic id, partition number).

It maintains one Admin client per source cluster, created lazily on first use. Each uses the security credentials and network settings from its mirror configuration, allowing different mirrors to use different authentication mechanisms. A separate Admin client connects to the destination cluster for offset commits and ACL updates. During periodic source cluster metadata refresh (60 seconds by default), the broker validates that the source cluster ID has not changed. If a mismatch is detected, metadata synchronization for that mirror is halted and moved to terminal FAILED state. This prevents silent data corruption in case of misconfiguration or unintended source cluster replacement.

This component holds the mirror partition cache, which is populated in two ways: by replaying __mirror_state partitions this broker leads, and by fetching state from remote coordinators via ReadMirrorStates RPCs. When this broker gains __mirror_state leadership, the ClusterMirrorCoordinator replays the log to rebuild the cache. When it loses leadership, the cache is cleared entirely.

Leadership Changes

When a broker becomes the new leader of a mirrored data partition, the KRaft metadata publisher invokes onMetadataUpdate on the local MirrorMetadataManager. It detects the leadership change from the metadata delta and needs to determine the current mirror partition state before it can act. If this broker also happens to coordinate the __mirror_state partition responsible for that record key, it reads the state directly from its local in-memory cache. Otherwise, it sends a ReadMirrorStates RPC to the broker that coordinates the relevant __mirror_state partition and waits for the response asynchronously. Once the state is retrieved, the broker applies the appropriate action for that state. For example, if the state is MIRRORING, it creates a mirror fetcher thread to start fetching from the source cluster. If the state is PAUSED, it does nothing. If the state is LOG_TRUNCATION, it initiates the truncation process. This way, leadership moves for data partitions are transparent to the mirroring state machine: the new leader picks up exactly where the previous leader left off.

Configuration Override

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.

MirrorFetcherThread

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

• Partitions from different mirrors use separate threads for authentication isolation.
• Partitions from the same mirror are distributed across multiple threads for load balancing.
• Leader changes in the source partition trigger thread reassignment or recreation to the new source broker.

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. Timestamp validation is skipped for mirrored records to preserve source timestamps.

Leader Epoch Invariant

Cluster mirroring enforces the invariant that the destination leader epoch is always greater than or equal to the source leader epoch (DLE >= 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 the guarantee that committed offsets from the source cluster always carry an epoch lower than or equal to 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 chosen 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:

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

The bump is performed via the BumpLeaderEpochs API. The active controller computes the final epoch as max(requestedEpoch + 1, currentLeaderEpoch) and writes a PartitionChangeRecord to the KRaft metadata log with the explicit LeaderEpoch value. 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.

Mirror Leader Epoch

Log Convergence

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

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

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

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

Unclean Leader Election

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

A new dynamic topic configuration called mirror.support.unclean.leader.election (boolean, default false) controls whether LME truncation waits for all assigned replicas to converge, not just current ISR members. This is a destination-side config that protects against destination ULE during the LOG_TRUNCATION phase. When enabled, the leader will not complete truncation until every replica has joined the ISR and truncated past the LME offset. This prevents subsequent unclean leader elections from introducing undetectable log divergence between clusters. Like in the example above, during step (2), the failback will stay in LOG_TRUNCATION until all replicas caught up and truncated to LME offset. So that assures the situation in step (3) won't happen. If a replica cannot catch up (slow network, disk failure), mirroring remains pending until the issue is resolved or the partition is reassigned to healthy brokers. Because this is a dynamic configuration, it can be toggled at any point in the mirroring lifecycle. Enabling it before mirroring starts guarantees full consistency between clusters even if an ULE occurs. Enabling it while mirroring is already running only guarantees consistency for records produced after the next LME triggered truncation completes.

Features Integration

Batch Compression

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.

Topic Compaction

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 present, while offsets 30–40 and 60–70 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.

Topic Retention

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 (LSO) 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 LSO and updates its local LSO to match, creating the same offset gap. If a mirror follower attempts to fetch from an offset below the source cluster's LSO (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 LSO 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.

Group Offsets

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.

Offset sync prepares the destination for failover. Once consumers are active there, they own their offsets. Fetched offsets are filtered to only include partitions belonging to actively mirrored topics, preventing offset ping-pong in bidirectional setups. Group ids with active members on the destination cluster are retrieved with a ListGroups RPC and skipped with a warning log. When they stop and the group becomes DEAD (EMPTY for more than offsets.retention.minutes), sync resumes automatically.

During offset synchronization, the committed offset is clamped to the valid range of the destination partition using the formula max(dest log start offset, min(dest LEO, source committed offset)). This ensures the committed offset never exceeds the destination's log end offset (LEO) and never falls below its log start offset. 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 committed to the destination will be 80 (the current LEO) rather than 100. As the mirror leader continues fetching data, it will update the committed offset toward the source value as the LEO advances. This clamping prevents OffsetOutOfRangeException during failover, since the committed offset is always within the valid log segment range of the destination partition.

Security Control

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 the access control lists (ACLs) that should be synchronized from the source cluster, and these ACLs are periodically replicated to the destination cluster in order to maintain consistent access control policies across both environments. An operator can grant ClusterMirror:*:CREATE,ALTER,DESCRIBE for full mirror management, or scope it to specific mirrors like ClusterMirror:my-mirror:DESCRIBE for read-only monitoring of a single mirror, without granting any broker-level privileges.

When connecting to the source cluster, Cluster Mirroring requires only the bootstrap server address and appropriate credentials, no other sensitive cluster information is exposed. 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. You have the option to configure a dedicated listener with minimal permissions for all mirroring traffic in the destination cluster or fallback to the inter-broker listener.

Source cluster permissions (mirror principal):

Destination cluster permissions:

Idempotent Producer

During mirroring, the destination partition is read-only: no local producers exist, so all ProducerStateManager (PSM) entries originate from mirrored data. When mirroring stops (failover), 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:                                                                                                                                                                                                            

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

Exactly-Once Semantics

Cluster Mirroring does not support exactly-once semantics (EOS) across clusters. The only guarantee is that, after a 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:

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

After the STOPPING transition appends abort markers after leader epoch bump: 

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. Ongoing transactions from mirrored data are resolved exclusively by the STOPPING transition flow described above.

Bandwidth Control

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

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

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.

Active-Active Topology

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. Bidirectional mirroring is supported, but only when mirroring different topics between clusters, allowing records produced to either cluster to be consumed from both. To prevent infinite replication loops when the same topic is mirrored in both directions, Cluster Mirroring performs mirror loop detection before starting: it queries the source cluster for its active mirrors and checks whether any of them already mirror the same topics back to the local cluster. If an overlap is detected, the affected partitions are moved to a failed state rather than entering mirroring.

Tiered Storage

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. During the LOG_TRUNCATION state, the mirror truncates the local log to the LME. This operation does not support tiered storage on the destination cluster because LME may be moved to remote segments. When tiered storage is enabled locally for a mirror topic, its partitions transition to FAILED state. This limitation applies only to the destination cluster during log truncation phase, and will be removed once tiered storage truncation will be fully supported. A detailed design of the metadata synchronization protocol, API schema, and state management will be provided in a follow-up KIP.

Synchronous Mirroring

Currently, mirroring is asynchronous. The source cluster acknowledges the producer without waiting for the destination to replicate the data. Synchronous 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.

Streams Applications

Stateful Kafka Streams rely on internal topics (changelogs, repartition topics, offset tracking) that are tightly coupled through atomic transactions. A single EOS transaction spans input offset commits, state store mutations written to changelog topics, and intermediate records written to repartition topics. Cluster mirroring replicates topics asynchronously and independently, so it cannot preserve these transactional boundaries. This means internal topics on the destination cluster can end up at inconsistent points in time relative to each other, making state store recovery produce incorrect results. The synchronous mirroring extension would preserve these guarantees. For this reason, mirroring of Kafka Streams internal topics is not supported.

Diskless Topics

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.

2PC Protocol

When mirroring stops, the MIRROR_PID_RESET control record aborts in-flight transactions and clears the ProducerStateManager. After the partition becomes writable again, the Transaction Coordinator may retry its WriteTxnMarker request with a COMMIT for the same transaction. Because the producer state was cleared, the second marker is accepted without error, resulting in two end transaction markers for one transaction: an ABORT followed by a COMMIT. READ_COMMITTED consumers are unaffected because the ABORT is already recorded in the transaction index and the dangling COMMIT produces no CompletedTxn, leaving both the transaction index and LSO unchanged. The Fetch response builds its aborted transactions list correctly from the existing index entry. However, __transaction_state on the source cluster records the transaction as COMMITTED while the destination partition log has it as aborted, creating a silent metadata inconsistency. With KIP-939, the window for this race becomes unbounded because prepared transactions can sit indefinitely. This category of divergence between coordinator state and partition log is not unique to mirroring. KAFKA-20716 documents the same class of inconsistency after unclean leader elections. A general fix for transaction state divergence would close this scenario as well.

Command Workflows

All mirror operations require the mirror.version feature flag to be enabled and pass authorization checks. Operations that modify mirror state (start, stop, pause, resume, delete) are sent to an arbitrary broker. The broker validates partition states by checking its local coordinator cache and sending ReadMirrorStates RPCs to remote coordinators as needed. If validation passes, the broker stamps the current metadata offset on the request and forwards it to the controller. The controller performs an optimistic locking check using the stamped offset: if any mirror state changed after the broker's validation, the entire request is rejected. The only exception is CreateClusterMirror, which the broker forwards directly to the controller without broker-side state validation.

ReadMirrorStates and WriteMirrorStates are inter-broker RPCs used internally by MirrorMetadataManager. ReadMirrorStates is used during broker-side validation (as described above) and also when a partition leader needs to retrieve its current mirror state from a remote coordinator during state transitions. WriteMirrorStates is used when a partition leader needs to persist state changes (partition state, last mirror epoch) to a __mirror_state partition coordinated by another broker. Both RPCs are batched per coordinator node to reduce the number of network round trips.

Create Mirror

1. The user sends a CreateClusterMirror request to any broker with the mirror name and configuration properties (bootstrap servers, source cluster ID, security settings, etc.).
2. The broker forwards the request directly to the active controller (no broker-side validation).
3. The controller validates required configs (bootstrap servers and source cluster ID) and mirror name uniqueness. If validation passes, it persists the properties as ConfigRecord entries with resource type CLUSTER_MIRROR in the metadata log.
4. If this is the first mirror being created, the controller also auto-creates the __mirror_state internal topic.
5. All brokers receive the metadata update and MirrorMetadataManager registers the new mirror configuration.

Start Mirror Topics

1. The user sends a StartMirrorTopics request to any broker with the mirror name, topics, and optional include/exclude patterns.
2. The broker validates that all partitions of the requested topics are are in STOPPED or UNKNOWN state (via local coordinator cache or ReadMirrorStates RPC), then forwards to the controller.
3. The controller persists the include/exclude patterns as ConfigRecord entries on the CLUSTER_MIRROR resource. For each topic, it creates the topic on the destination if it does not already exist (preserving the source topic ID), and writes a MirrorTopicStateChangeRecord with DesiredState=MIRRORING. Individual topics can fail independently (partial success model).
4. Brokers receive the metadata update. On each broker that leads affected partitions, MirrorMetadataManager detects the new mirrorName and DesiredState from the TopicImage and queries the coordinator for the current partition state (local cache or remote RPC).
5. Partitions transition from UNKNOWN to LOG_TRUNCATION. During this state, the broker queries the source cluster for last mirror epochs (for failback scenarios), truncates the log accordingly, and waits for all ISR members (or all replicas if unclean leader election support is enabled).
6. Partitions transition to MIRRORING. A MirrorFetcherThread is created and begins fetching from the source cluster. Partition state is persisted to __mirror_state (via local append or WriteMirrorStates RPC).
7. On subsequent metadata refresh cycles, MirrorMetadataManager discovers new source topics matching the persisted include/exclude patterns and repeats steps 3 through 7 for each.

Stop Mirror Topics

1. The user sends a StopMirrorTopics request to any broker with the mirror name, topics, and optional patterns.
2. The broker validates that all partitions are in MIRRORING or PAUSED state (via local coordinator cache or ReadMirrorStates RPC), then forwards to the controller.
3. If patterns are provided, the controller removes matching entries from mirror.topics.include or adds them to mirror.topics.exclude on the CLUSTER_MIRROR resource. Any currently mirroring topic that now matches the updated exclude pattern is also stopped.
4. For each topic, the controller writes a MirrorTopicStateChangeRecord with DesiredState=STOPPED.
5. When MirrorMetadataManager on the partition leader broker receives the metadata update, it reads the desired state from the TopicImage and transitions the partition to STOPPING.
6. During STOPPING, the following operations execute sequentially:
   1. Fetcher threads are removed for the affected partitions, stopping replication.
   2. The current leader epoch is persisted as the last mirror epoch (LME) in __mirror_state (via local append or WriteMirrorStates RPC).
   3. The partition's leader epoch is bumped to draw a boundary between mirrored and locally produced records.
   4. ABORT markers are appended for all in-flight transactions using the new leader epoch.
   5. A MIRROR_PID_RESET control record is written to expire all producer state entries from the source cluster.
7. Partitions transition to STOPPED and become writable.

Pause Mirror Topics

1. The user sends a PauseMirrorTopics request to any broker with the mirror name and topics.
2. The broker validates that all partitions are in MIRRORING state (via local coordinator cache or ReadMirrorStates RPC), then forwards to the controller.
3. The controller writes a MirrorTopicStateChangeRecord with DesiredState=PAUSED for each topic.
4. When MirrorMetadataManager on the partition leader broker receives the metadata update, it transitions the partition to PAUSING.
5. During PAUSING, MirrorFetcherManager removes the fetcher threads for the affected partitions. No more data is replicated.
6. The partition transitions to PAUSED. The partition remains read-only. Metadata synchronization (configs, groups, ACLs) is also paused for these topics.
7. The state change is persisted to __mirror_state (via local append or WriteMirrorStates RPC).

Resume Mirror Topics

1. The user sends a ResumeMirrorTopics request to any broker with the mirror name and topics.
2. The broker validates that all partitions are in PAUSED state (via local coordinator cache or ReadMirrorStates RPC), then forwards to the controller.
3. The controller writes a MirrorTopicStateChangeRecord with DesiredState=MIRRORING, restoring the mirroring intent.
4. When MirrorMetadataManager on the partition leader broker receives the metadata update, it transitions the partition back to MIRRORING.
5. New MirrorFetcherThread instances are created and resume replication from the current log end offset. Metadata synchronization (configs, groups, ACLs) also resumes.

Delete Mirror

1. The user sends a DeleteClusterMirror request to any broker with the mirror name.
2. The broker validates that all partitions under the mirror's topics are in STOPPED state (via local coordinator cache or ReadMirrorStates RPC), then forwards to the controller.
3. The controller verifies that no topics are in MIRRORING or PAUSED state. For each stopped topic, it writes a MirrorTopicStateChangeRecord with a null mirror name and DesiredState=UNKNOWN, disassociating the topic from the mirror. These records are excluded from metadata snapshots (only topics with a non-null mirror name are written to snapshots).
4. The controller tombstones the mirror configuration by deleting all ConfigRecord entries for the CLUSTER_MIRROR resource. The topic disassociations and config tombstones are written as a single atomic batch.
5. When brokers receive the metadata update, MirrorMetadataManager detects the deleted mirror, tombstones the corresponding records in __mirror_state, closes source cluster connections, and cleans up cached state. The mirror name becomes available for reuse.

List Mirrors

1. The user sends a ListClusterMirrors request to any broker (no parameters required).
2. The admin client sends the request to a single broker because mirror configuration is available from the replicated metadata log.
3. The broker reads all configured mirrors from its local metadata cache and filters by authorization.
4. For each authorized mirror, the broker returns: mirror name, source cluster ID, source bootstrap servers, and topic count.
5. This is a read-only operation. No metadata records are written.

Describe Mirrors

1. The user sends a DescribeClusterMirrors request with optional mirror names (empty means all mirrors).
2. The admin client fans out the request to all brokers and merges the responses client-side.
3. Each broker queries two local sources: MirrorFetcherManager (which tracks source offset, destination offset, and lag, updated on every fetch response) and ClusterMirrorCoordinator (which provides the current partition state from its __mirror_state cache).
4. Each broker only reports partitions for which it is the partition leader, avoiding duplicates.
5. For each partition, the response includes: topic name, partition index, source offset, destination offset, lag, current state, retry attempt count, and error message (for FAILED partitions).

Public Interfaces

Command-Line

Examples for the new command line tool and updates to extisting tools.

New Tool

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

$ bin/kafka-cluster-mirrors.sh --help
Create cluster mirrors and manage mirror 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:

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

Start mirroring a topic or set of topics (the --topics flag accepts regex expression):

$ bin/kafka-cluster-mirrors.sh --bootstrap-server :9094 --start \                                                                                                                                                                                                                                  
  --topics 'orders-.*' --exclude 'orders-internal' --mirror a-to-b
Started 2 mirror topic(s) in mirror a-to-b: [orders-us, orders-eu]

Stop mirroring a topic or set of topics (failover; topics become writable):

$ bin/kafka-cluster-mirrors.sh --bootstrap-server :9094 --stop --topics 'orders-us' --mirror a-to-b
Stopped mirroring for 1 topic(s) in mirror a-to-b: [orders-us]

Delete a mirror including its configuration (the mirror must be empty or include only stopped partitions):

$ bin/kafka-cluster-mirrors.sh --bootstrap-server :9094 --delete --mirror a-to-b
Deleted mirror a-to-b

Pause mirroring for a specific topic or set of topics (topics remain read-only):

$ bin/kafka-cluster-mirrors.sh --bootstrap-server :9094 --pause --topics my-topic --mirror a-to-b
Paused mirroring for 1 topic(s) in mirror a-to-b: [my-topic]

Resume mirroring for a specific topic or set of topics:

$ bin/kafka-cluster-mirrors.sh --bootstrap-server :9094 --resume --topics my-topic --mirror a-to-b
Resumed mirroring for 1 topic(s) in mirror a-to-b: [my-topic]

List configured mirrors with additional information:

$ bin/kafka-cluster-mirrors.sh --bootstrap-server :9094 --list
MIRROR                         TOPICS     SOURCE-CLUSTER-ID          SOURCE-BOOTSTRAP-SERVER
a-to-b                         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 only show partitions from a specific mirror):

$ bin/kafka-cluster-mirrors.sh --bootstrap-server :9094 --describe
MIRROR                         TOPIC                                    PARTITION  SOURCE-OFFSET   DESTINATION-OFFSET LAG      STATE       
a-to-b                         bar                                      0          -               -                  -        STOPPED   
a-to-b                         foo                                      0          69              66                 3        MIRRORING   
a-to-b                         foo                                      1          94              84                 10       MIRRORING   
a-to-b                         foo                                      2          94              90                 4        MIRRORING   
new-mirror                     baz                                      0          -               -                  -        PAUSED   
new-mirror                     baz                                      1          -               -                  -        PAUSED

Describe failed mirror partitions with retry count and error messages:

$ bin/kafka-cluster-mirrors.sh --bootstrap-server :9094 --describe --failed
MIRROR                         TOPIC                                    PARTITION  RETRY   ERROR
a-to-b                         my-topic                                 0          2/3     Truncation requires source metadata refresh
a-to-b                         my-topic                                 1          3/3     Fetch IO error: Connection to localhost:9092 (id: 2 rack: null isFenced: fals...

Failover

Failover is initiated by calling the StopMirrorTopics API, which transitions mirror topics from read only to writable after the stopping process completes. Producers reconnect with fresh producer IDs and sequence numbers. Consumers resume from the last synchronized offsets using the same group ID.

# 9091 (source) -----> 9094 (destination)
# in case of disaster, the operator can failover by running the following command
$ bin/kafka-cluster-mirrors.sh --bootstrap-server :9094 --stop --topic .* --mirror a-to-b
# 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

Failback is initiated by creating a new mirror on the old source cluster with the new source as target, then calling the StartMirrorTopics API to begin reverse mirroring. Only the delta is mirrored if the source cluster supportsCluster Mirroring.

# when the source cluster is back, the operator can failback by creating a new mirror
$ echo "bootstrap.servers=localhost:9094" > /tmp/b-to-a.properties
$ bin/kafka-cluster-mirrors.sh --bootstrap-server :9091 --create --mirror b-to-a --mirror-config /tmp/b-to-a.properties
$ bin/kafka-cluster-mirrors.sh --bootstrap-server :9091 --start --topic .* --mirror b-to-a
# 9091 (destination) <----- 9094 (source)

Configuration

Alter mirror configuration (any valid configuration triggers a reconnection):

$ bin/kafka-configs.sh --bootstrap-server :9094 --entity-type cluster-mirrors --entity-name a-to-b \
    --alter --add-config 'bootstrap.servers=localhost:9092'
Completed updating config for mirror a-to-b.

Grant mirror admin full access to a specific mirror:

$ bin/kafka-acls.sh --bootstrap-server :9094 --add \
  --cluster-mirror a-to-b \
  --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=a-to-b, 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 a-to-b
Current ACLs for resource `ResourcePattern(resourceType=CLUSTER_MIRROR, name=a-to-b, 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)   

Throttling

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-a-to-b
Completed updating config for client broker-4-fetcher-0-mirror-a-to-b.

Debugging

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

Alternatively, you can use the new message formatter to consume from the internal metadata topic:

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

Admin Client

New methods are added to the Admin interface for programmatic cluster mirror management, along with their supporting classes.

CreateClusterMirror

/**
 * 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 CreateClusterMirrorResult
 */
CreateClusterMirrorResult createClusterMirror(String mirrorName, Map<String, String> configs, CreateClusterMirrorOptions options);

/**
 * Options for {@link Admin#createClusterMirror(String, Map, CreateClusterMirrorOptions)}.
 */
public class CreateClusterMirrorOptions extends AbstractOptions<CreateClusterMirrorOptions> {
}

/**
 * The result of the {@link Admin#createClusterMirror(String, Map, CreateClusterMirrorOptions)} call.
 */
public class CreateClusterMirrorResult {
    private final KafkaFuture<Void> future;

    CreateClusterMirrorResult(final KafkaFuture<Void> future) {
        this.future = future;
    }

    /**
     * Return a future which succeeds if the operation is successful.
     */
    public KafkaFuture<Void> all() {
        return future;
    }
}

StartMirrorTopics

/**
 * 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> {
    
    public StartMirrorTopicsOptions includePatterns(List<String> patterns) {
    }

    public StartMirrorTopicsOptions excludePatterns(List<String> patterns) {
    }

    public List<String> includePatterns() {
    }

    public List<String> excludePatterns() {
    }
}

/**
 * The result of the {@link Admin#startMirrorTopics(String, Set, StartMirrorTopicsOptions)} call.
 */
public class StartMirrorTopicsResult {

    StartMirrorTopicsResult(final KafkaFuture<Void> future) {
    }

    /**
     * Return a future which succeeds if the operation is successful.
     */
    public KafkaFuture<Void> all() {
    }
}

StopMirrorTopics

/**
 * 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> {

    public StopMirrorTopicsOptions patterns(List<String> patterns) {
    }

    public List<String> patterns() {
    }
}

/**
 * The result of the {@link Admin#stopMirrorTopics(String, Set, StopMirrorTopicsOptions)} call.
 */
public class StopMirrorTopicsResult {

    StopMirrorTopicsResult(KafkaFuture<Void> future) {
    }

    /**
     * Return a future which succeeds if the operation is successful.
     */
    public KafkaFuture<Void> all() {
    }
}

PauseMirrorTopics

/**
 * 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);

/**
 * Options for {@link Admin#pauseMirrorTopics(String, Set, PauseMirrorTopicsOptions)}.
 */
public class PauseMirrorTopicsOptions extends AbstractOptions<PauseMirrorTopicsOptions> {
}

/**
 * The result of the {@link Admin#pauseMirrorTopics(String, Set, PauseMirrorTopicsOptions)} call.
 */
public class PauseMirrorTopicsResult {

    PauseMirrorTopicsResult(final KafkaFuture<Void> future) {
    }

    public KafkaFuture<Void> all() {
    }
}

ResumeMirrorTopics

/**
 * 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);

/**
 * Options for {@link Admin#resumeMirrorTopics(String, Set, ResumeMirrorTopicsOptions)}.
 */
public class ResumeMirrorTopicsOptions extends AbstractOptions<ResumeMirrorTopicsOptions> {
}

/**
 * The result of the {@link Admin#resumeMirrorTopics(String, Set, ResumeMirrorTopicsOptions)} call.
 */
public class ResumeMirrorTopicsResult {

    ResumeMirrorTopicsResult(final KafkaFuture<Void> future) {
    }

    public KafkaFuture<Void> all() {
    }
}

DeleteClusterMirror

/**
 * 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 DeleteClusterMirrorResult
 */
DeleteClusterMirrorResult deleteClusterMirror(String mirrorName, DeleteClusterMirrorOptions options);

/**
 * Options for {@link Admin#deleteClusterMirror(String, DeleteClusterMirrorOptions)} .
 */
public class DeleteClusterMirrorOptions extends AbstractOptions<DeleteClusterMirrorOptions> {
}

/**
 * The result of the {@link Admin#deleteClusterMirror(String, DeleteClusterMirrorOptions)}  call.
 */
public class DeleteClusterMirrorResult {
    private final KafkaFuture<Void> future;

    DeleteClusterMirrorResult(final KafkaFuture<Void> future) {
        this.future = future;
    }

    public KafkaFuture<Void> all() {
        return future;
    }
}

ListClusterMirrors

/**
 * List the cluster mirrors available in the cluster. The request is sent to a single
 * broker because mirror configuration is available from the replicated metadata log.
 *
 * @param options The options to use when listing the mirrors.
 * @return The ListClusterMirrorsResult.
 */
ListClusterMirrorsResult listClusterMirrors(ListClusterMirrorsOptions options);

/**
 * Options for {@link Admin#listClusterMirrors()}.
 */
public class ListClusterMirrorsOptions extends AbstractOptions<ListClusterMirrorsOptions> {

    /**
     * Set whether the response should include mirror topic names for each mirror.
     */
    public ListClusterMirrorsOptions shouldIncludeTopicNames(boolean shouldIncludeTopicNames) {
    }

    public boolean shouldIncludeTopicNames() {
    }
}

/**
 * The result of the {@link Admin#listClusterMirrors()} call.
 */
public class ListClusterMirrorsResult {
    private final KafkaFutureImpl<Collection<ClusterMirrorListing>> all;
    private final KafkaFutureImpl<Collection<ClusterMirrorListing>> valid;
    private final KafkaFutureImpl<Collection<Throwable>> errors;

    ListClusterMirrorsResult(KafkaFuture<Collection<Object>> future) {
    }

    /**
     * Returns a future that yields either an exception, or the full set of mirror listings.
     * <p>
     * In the event of a failure, the future yields nothing but the first exception which
     * occurred.
     */
    public KafkaFuture<Collection<ClusterMirrorListing>> all() {
    }

    /**
     * Returns a future which yields just the valid listings.
     * <p>
     * This future never fails with an error, no matter what happens.  Errors are completely
     * ignored.  If nothing can be fetched, an empty collection is yielded.
     * If there is an error, but some results can be returned, this future will yield
     * those partial results.  When using this future, it is a good idea to also check
     * the errors future so that errors can be displayed and handled.
     */
    public KafkaFuture<Collection<ClusterMirrorListing>> valid() {
    }

    /**
     * Returns a future which yields just the errors which occurred.
     * <p>
     * If this future yields a non-empty collection, it is very likely that elements are
     * missing from the valid() set.
     * <p>
     * This future itself never fails with an error.  In the event of an error, this future
     * will successfully yield a collection containing at least one exception.
     */
    public KafkaFuture<Collection<Throwable>> errors() {
    }
}

/**
 * A listing of a cluster mirror.
 */
public class ClusterMirrorListing {

    /**
     * Create an instance with the specified parameters.
     *
     * @param mirrorName Mirror name
     * @param sourceBootstrap Source cluster bootstrap servers
     * @param sourceClusterId Source cluster ID
     * @param topicCount Number of topics configured for this mirror
     * @param topics List of topic names configured for this mirror
     */
    public ClusterMirrorListing(String mirrorName, String sourceBootstrap, String sourceClusterId, int topicCount, List<String> topics) {
    }

    /**
     * The mirror name.
     *
     * @return Mirror name
     */
    public String mirrorName() {
    }

    /**
     * The source cluster bootstrap servers.
     *
     * @return Source bootstrap servers, or null if not available
     */
    public String sourceBootstrap() {
    }

    /**
     * The source cluster ID.
     *
     * @return Source cluster ID, or empty string if not yet resolved
     */
    public String sourceClusterId() {
    }

    /**
     * The number of topics configured for this mirror.
     *
     * @return Number of topics, or 0 if mirror has no topics configured
     */
    public int topicCount() {
    }

    /**
     * The topic names configured for this mirror.
     *
     * @return List of topic names, or empty list if not requested or not available
     */
    public List<String> topics() {
    }
}

DescribeClusterMirrors

/**
 * Describe cluster mirrors on the destination cluster. Returns per-partition state,
 * replication lag, and last mirror epoch. The request is sent to all brokers because
 * each broker only reports partitions for which it is the mirror leader. Optionally
 * performs last mirror epoch lookups for failback truncation when cluster ID and lookup
 * entries are provided in the options.
 *
 * @param mirrorNames The names of the mirrors to describe
 * @param options The options to use when describing mirrors
 * @return The DescribeClusterMirrorsResult
 */
DescribeClusterMirrorsResult describeClusterMirrors(Collection<String> mirrorNames, DescribeClusterMirrorsOptions options);

/**
 * Options for {@link Admin#describeClusterMirrors(java.util.Collection, DescribeClusterMirrorsOptions)}.
 */
public class DescribeClusterMirrorsOptions extends AbstractOptions<DescribeClusterMirrorsOptions> {

    /**
     * Set whether authorized operations should be included in the response.
     *
     * @param includeAuthorizedOperations whether to include authorized operations
     * @return this instance
     */
    public DescribeClusterMirrorsOptions includeAuthorizedOperations(boolean includeAuthorizedOperations) {
    }

    /**
     * Return true if authorized operations should be included in the response.
     */
    public boolean includeAuthorizedOperations() {
    }

    public DescribeClusterMirrorsOptions clusterId(String clusterId) {
    }

    public String clusterId() {
    }

    public DescribeClusterMirrorsOptions lastMirrorEpochLookups(List<DescribeClusterMirrorsRequestData.LastMirrorEpochLookup> lastMirrorEpochLookups) {
    }

    public List<DescribeClusterMirrorsRequestData.LastMirrorEpochLookup> lastMirrorEpochLookups() {
    }
}

/**
 * The result of the {@link Admin#describeClusterMirrors(java.util.Collection)} call.
 */
public class DescribeClusterMirrorsResult {

    DescribeClusterMirrorsResult(KafkaFuture<Map<String, ClusterMirrorDescription>> future) {
    }

    DescribeClusterMirrorsResult(KafkaFuture<Map<String, ClusterMirrorDescription>> future,
                                 KafkaFuture<Map<Uuid, Map<Integer, Integer>>> lookupEpochsFuture) {
    }

    /**
     * Return a future which succeeds only if all the mirror descriptions succeed.
     */
    public KafkaFuture<Map<String, ClusterMirrorDescription>> allDescriptions() {
    }

    /**
     * Return a future containing last mirror epoch lookup results.
     * Keyed by topicId, then partitionIndex to lastMirrorEpoch.
     */
    public KafkaFuture<Map<Uuid, Map<Integer, Integer>>> lookupEpochs() {
    }
}

/**
 * A detailed description of a cluster mirror.
 */
public class ClusterMirrorDescription {

    public ClusterMirrorDescription(String mirrorName,
                                    Map<String, Set<LeaderStateDescription>> topics,
                                    Set<AclOperation> authorizedOperations) {
    }

    public String mirrorName() {
    }

    public Map<String, Set<LeaderStateDescription>> topics() {
    }

    public Set<AclOperation> authorizedOperations() {
    }

    /** Represents the mirroring state of a leader partition. */
    public static class LeaderStateDescription {

        public LeaderStateDescription(TopicPartition topicPartition, long sourceOffset, long destinationOffset,
                                      long lag, String state, int retryAttempt, String errorMessage) {
        }

        public TopicPartition topicPartition() {
        }

        public long sourceOffset() {
        }

        public long destinationOffset() {
        }

        public long lag() {
        }

        public String state() {
        }

        public int retryAttempt() {
        }

        public String errorMessage() {
        }
    }
}

We also added the leaderEpoch filed to TopicPartitionInfo so the DescribeTopicPartitions response (which already carries it on the wire) surfaces it through the Admin API.

public class TopicPartitionInfo {
    ...
    private final Optional<Integer> leaderEpoch;

Protocol Changes

This section describes all protocol level changes.

CreateTopic

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:

1. This topic ID is not used by other topics in the destination cluster.
2. The replicas for the partition assignment are all active and not in fenced or controlled shutdown. This is to make sure when a topic gets deleted and re-created with the same topic ID, the stale offline log dir won’t be treated as the active log dir after it becomes online (KAFKA-16234).

Fetch

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." },

CreateClusterMirror

Allows users to create a mirror and supply its configuration. The broker validates that the mirror name is not already in use and contains only permitted characters. 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": "CreateClusterMirrorRequest",
  // 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": "[]ClusterMirrorConfig", "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": "CreateClusterMirrorResponse",
  // 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." }
  ]
}

StartMirrorTopics

Start mirroring for the specified topics. The broker validates that all target topic partitions are in either UNKNOWN or STOPPED or FAILED state; otherwise, the request is rejected.

{
  "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": "[]TopicMetadata", "versions": "0+", "about": "The data for the topics.",
      "fields": [
        { "name": "TopicId", "type": "uuid", "versions": "0+", "about": "The 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+", "taggedVersions": "0+", "tag": 0,
      "about": "Regex patterns to add to mirror.topics.include (mirror.topics.exclude takes precedence over this)." },
    { "name": "ExcludePatterns", "type": "[]string", "versions": "0+", "taggedVersions": "0+", "tag": 1,
      "about": "Regex patterns to add to mirror.topics.exclude (this takes precendence over mirror.topics.include)." },
    { "name": "StateValidationOffset", "type": "int64", "versions": "0+", "default": "-1",
      "about": "Metadata log offset at which partition states were validated. -1 to skip fencing."}
  ]
}

{
  "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": "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." },
      { "name": "ErrorMessage", "type": "string", "versions": "0+", "nullableVersions": "0+", "default": "null",
        "about": "The error message, or null if there was no error." }
    ]}
  ]
}

StopMirrorTopics

Stop mirroring for the specified topics. The broker validates that all target topic partitions are in either LOG_TRUNCATION or MIRRORING or FAILED state.

{
  "apiKey": TBD,
  "type": "request",
  "listeners": ["broker", "controller"],
  "name": "StopMirrorTopicsRequest",
  // Version 0 is the initial version.
  "validVersions": "0",
  "flexibleVersions": "0+",
  "fields": [
    { "name": "MirrorName", "type": "string", "versions": "0+", "entityType": "mirrorName",
      "about": "The cluster mirror name." },
    { "name": "Topics", "type": "[]TopicMetadata", "versions": "0+", "about": "The data for the topics.",
      "fields": [
        { "name": "TopicName", "type": "string", "versions": "0+", "mapKey": true, "entityType": "topicName",
          "about": "The topic name." }
      ]},
    { "name": "Patterns", "type": "[]string", "versions": "0+", "taggedVersions": "0+", "tag": 0,
      "about": "Topic patterns to remove from mirror.topics.include or add to mirror.topics.exclude." },
    { "name": "StateValidationOffset", "type": "int64", "versions": "0+", "default": "-1",
      "about": "Cluster metadata offset at which the broker validated preconditions. -1 to skip validation."}
  ]
}

{
  "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": "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." },
      { "name": "ErrorMessage", "type": "string", "versions": "0+", "nullableVersions": "0+", "default": "null",
        "about": "The error message, or null if there was no error." }
    ]}
  ]
}

PauseMirrorTopics

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+", "entityType": "mirrorName",
      "about": "The cluster mirror name." },
    { "name": "Topics", "type": "[]TopicMetadata", "versions": "0+", "about": "The data for the topics.",
      "fields": [
        { "name": "TopicName", "type": "string", "versions": "0+", "mapKey": true, "entityType": "topicName",
          "about": "The topic name." }
      ]},
    { "name": "StateValidationOffset", "type": "int64", "versions": "0+", "default": "-1",
      "about": "Cluster metadata offset at which the broker validated preconditions. -1 to skip validation."}
  ]
}

{
  "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": "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." },
      { "name": "ErrorMessage", "type": "string", "versions": "0+", "nullableVersions": "0+", "default": "null",
        "about": "The error message, or null if there was no error." }
    ]}
  ]
}

ResumeMirrorTopics

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+", "entityType": "mirrorName",
      "about": "The cluster mirror name." },
    { "name": "Topics", "type": "[]TopicMetadata", "versions": "0+", "about": "The data for the topics.",
      "fields": [
        { "name": "TopicName", "type": "string", "versions": "0+", "mapKey": true, "entityType": "topicName",
          "about": "The topic name." }
      ]},
    { "name": "StateValidationOffset", "type": "int64", "versions": "0+", "default": "-1",
      "about": "Cluster metadata offset at which the broker validated preconditions. -1 to skip validation."}
  ]
}

{
  "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": "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." },
      { "name": "ErrorMessage", "type": "string", "versions": "0+", "nullableVersions": "0+", "default": "null",
        "about": "The error message, or null if there was no error." }
    ]}
  ]
}

DeleteClusterMirror

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": "DeleteClusterMirrorRequest",
  // 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."},
    { "name": "StateValidationOffset", "type": "int64", "versions": "0+", "default": "-1",
      "about": "Cluster metadata offset at which the broker validated preconditions. -1 to skip validation."}
  ]
}

{
  "apiKey": TBD,
  "type": "response",
  "name": "DeleteClusterMirrorResponse",
  // 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." }
  ]
}

ListClusterMirrors

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": "ListClusterMirrorsRequest",
  // Version 0 is the initial version.
  "validVersions": "0",
  "flexibleVersions": "0+",
  "fields": [
    { "name": "IncludeTopicNames", "type": "bool", "versions": "0+", "default": "false",
      "about": "If true, the response will include mirror topic names for each mirror." }
  ]
}

{
  "apiKey": TBD,
  "type": "response",
  "name": "ListClusterMirrorsResponse",
  // 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." },
      { "name": "TopicNames", "type": "[]string", "versions": "0+",
        "about": "The topic names included in this mirror." }
    ]}
  ]
}

DescribeClusterMirrors

Returns the current mirroring status, state, and configuration for the specified mirror topics on the destination cluster. The response includes the partition state as a string value as defined in the mirror partition states table. Allows destination cluster partition leaders to query the LME from the source cluster.

{
  "apiKey": TBD,
  "type": "request",
  "listeners": ["broker"],
  "name": "DescribeClusterMirrorsRequest",
  // 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." },
    { "name": "ClusterId", "type": "string", "versions": "0+", "nullableVersions": "0+",
      "taggedVersions": "0+", "tag": 0, "ignorable": true, "default": "null",
      "about": "The requesting cluster's own ID for last mirror epoch lookup." },
    { "name": "LastMirrorEpochLookups", "type": "[]LastMirrorEpochLookups", "versions": "0+",
      "taggedVersions": "0+", "tag": 1, "ignorable": true,
      "about": "Last mirror epoch lookups for incremental mirroring when failing back.", "fields": [
      { "name": "TopicId", "type": "uuid", "versions": "0+",
        "about": "The topic id." },
      { "name": "Partitions", "type": "[]int32", "versions": "0+",
        "about": "The partition indexes." }
    ]}
  ]
}

{
  "apiKey": TBD,
  "type": "response",
  "name": "DescribeClusterMirrorsResponse",
  // 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": "StateValue", "type": "string", "versions": "0+",
            "about": "The partition state as a string value." },
          { "name": "RetryAttempt", "type": "int16", "versions": "0+", "default": "0",
            "about": "The number of automatic retry attempts while in FAILED state, or -1 for non-retryable failure." },
          { "name": "ErrorMessage", "type": "string", "versions": "0+", "nullableVersions": "0+", "default": "null",
            "about": "The error message when in FAILED state, or null if not applicable." },
          { "name": "LastMirrorEpoch", "type": "int32", "versions": "0+", "default": "-1",
            "about": "The last mirror leader epoch, or -1 if not available."
          }
        ]}
      ]}
    ]},
    { "name": "LookupResults", "type": "[]LookupResult", "versions": "0+",
      "taggedVersions": "0+", "tag": 0, "ignorable": true,
      "about": "Last mirror epoch lookup results incremental mirroring when failing back.", "fields": [
      { "name": "TopicId", "type": "uuid", "versions": "0+",
        "about": "The topic id." },
      { "name": "Partitions", "type": "[]PartitionResult", "versions": "0+",
        "about": "Per-partition last mirror epoch results.", "fields": [
        { "name": "PartitionIndex", "type": "int32", "versions": "0+",
          "about": "The partition index." },
        { "name": "LastMirrorEpoch", "type": "int32", "versions": "0+", "default": "-1",
          "about": "The max last mirror epoch across all matching mirrors, or -1 if not found." }
      ]}
    ]}
  ]
}

ReadMirrorStates

Internal API that reads the current mirror partition states from the internal __mirror_state topic on the destination cluster. The "PreviousState" and "RetryAttempt" fields are for the FAILED state retry to transition back to the "PreviousState" from "FAILED" state once the exponential backoff expires. Response includes the current partition state stored in __mirror_state that can have one of the values defined in the mirror partition states table.  

{
  "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": "[]TopicMetadata", "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 mirrored 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." },
        { "name": "PreviousState", "type": "int8", "versions": "0+", "default": 16,
          "about": "The mirror partition state before the last transition; UNKNOWN if not recorded." },
        { "name": "RetryAttempt", "type": "int16", "versions": "0+", "default": 0,
          "about": "The number of automatic retry attempts while in FAILED state, or -1 for non-retryable failure." },
        { "name": "ErrorMessage", "type": "string", "versions": "0+", "nullableVersions": "0+", "default": "null",
          "about": "The error message, or null if there was no error." }
      ]}
    ]}
  ]
}

WriteMirrorStates

Internal API that persists mirror partition state transitions to the internal __mirror_state topic on the destination cluster. Request includes the new mirror partition state to replace the current state stored in __mirror_state, and can have one of the values defined in the mirror partition states table. 

{
  "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 cluster mirror name." },
    { "name": "Topics", "type": "[]TopicMetadata", "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": "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": "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." },
        { "name": "ErrorMessage", "type": "string", "versions": "0+", "nullableVersions": "0+", "default": "null",
          "about": "The error message, or null if there was no error." }
      ]}
    ]}
  ]
}

BumpLeaderEpochs

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": "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": "[]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." },
        { "name": "ErrorMessage", "type": "string", "versions": "0+", "nullableVersions": "0+", "default": "null",
          "about": "The error message, or null if there was no error." }
      ]}
    ]}
  ]
}

Cluster Metadata Records

This section describes records written to the KRaft metadata log by the active controller as part of Cluster Mirroring operations.

PartitionChangeRecord

Written by the controller when processing a BumpLeaderEpochs request. The record carries a LeaderEpoch field with the computed final epoch for the partition.

{
  "apiKey": 5,                                                                                                                                                                                                                                                                           
  "type": "metadata",                                                                                                                                                                                                                                                                    
  "name": "PartitionChangeRecord",
  "validVersions": "0-3",                                                                                                                                                                                                                                                                
  "flexibleVersions": "0+",
  "fields": [
    // ... existing fields ...
    { "name": "LeaderEpoch", "type": "int32", "versions": "3+", "default": -1,
      "about": "-1 if the leader epoch did not change; the new leader epoch otherwise."},
    // ... existing fields ...
  ]
}

MirrorTopicStateChangeRecord

Written by the controller when processing a start, stop and pause requests for a specific mirror. Includes the desired state for each mirror partition with the following valid values: MIRRORING, PAUSED, STOPPED (see mirror partition states table).

{
  "apiKey": 29,
  "type": "metadata",
  "name": "MirrorTopicStateChangeRecord",
  "validVersions": "0",
  "flexibleVersions": "0+",
  "fields": [
    { "name": "TopicId", "type": "uuid", "versions": "0+",
      "about": "The unique topic ID." },
    { "name": "MirrorName", "type": "string", "versions": "0+", "nullableVersions": "0+", "entityType": "mirrorName",
      "about": "The cluster mirror name. Null when disassociating a topic from a mirror." },
    { "name": "DesiredState", "type": "int8", "versions": "0+",
      "about": "The desired state for all mirror partitions." }
  ]
}

Cluster Control Records

This section describes control records written to data log as part of Cluster Mirroring operations.

MirrorPidResetRecord

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

Mirror Metadata Records

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

LastMirrorEpochs

Written during the STOPPING transition to record the last mirror leader epoch for each partition before the destination becomes writable. The key identifies the cluster mirror name, topic id, and partition. The value holds a single epoch.

{
  "apiKey": 1,
  "type": "coordinator-key",
  "name": "LastMirrorEpochsKey",
  "validVersions": "0",
  "flexibleVersions": "none",
  "fields": [
    { "name": "MirrorName", "type": "string", "versions": "0+", "entityType": "mirrorName",
      "about": "The cluster mirror name."},
    { "name": "TopicId", "type": "uuid", "versions": "0+",
      "about": "The topic id."},
    { "name": "Partition", "type": "int32", "versions": "0+",
      "about": "The partition index."}
  ]
}

{
  "apiKey": 1,
  "type": "coordinator-value",
  "name": "LastMirrorEpochsValue",
  "validVersions": "0",
  "flexibleVersions": "0+",
  "fields": [
    { "name": "LastMirrorEpoch", "type": "int32", "versions": "0+",
      "about": "The last mirror leader epoch for this partition." }
  ]
}

MirrorPartitionState

Written on every mirror partition state transition. The key identifies the cluster mirror name, topic id, and partition. The value holds the current state (one of the values in the mirror partition states table), the previous state for automatic retry handling, and optional error metadata.

{
  "apiKey": 2,
  "type": "coordinator-key",
  "name": "MirrorPartitionStateKey",
  "validVersions": "0",
  "flexibleVersions": "none",
  "fields": [
    { "name": "MirrorName", "type": "string", "versions": "0+", "entityType": "mirrorName",
      "about": "The cluster mirror name."},
    { "name": "TopicId", "type": "uuid", "versions": "0+",
      "about": "The topic id."},
    { "name": "Partition", "type": "int32", "versions": "0+",
      "about": "The partition index."}
  ]
}

{
  "apiKey": 2,
  "type": "coordinator-value",
  "name": "MirrorPartitionStateValue",
  "validVersions": "0",
  "flexibleVersions": "0+",
  "fields": [
    { "name": "State", "type": "int8", "versions": "0+",
      "about": "The mirror partition state." },
    { "name": "PreviousState", "type": "int8",  "versions": "0+", "default": 16,
      "about": "The mirror partition state before this transition; UNKNOWN if not recorded." },
    { "name": "RetryAttempt", "type": "int16", "versions": "0+", "default": "0",
      "about": "The number of automatic retry attempts while in FAILED state, or -1 for non-retryable failure." },
    { "name": "ErrorMessage", "type": "string", "versions": "0+", "nullableVersions": "0+", "default": "null",
      "about": "The error message when in FAILED state, or null if not applicable." }
  ]
}

Type Enumerations

This section lists the new values added to existing Kafka type enumerations to support Cluster Mirroring

Config

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

public enum ConfigType {
    // existing types unchanged 
    CLUSTER_MIRRORS("cluster-mirrors");
}

Config Resource

The resource type that represents the cluster mirror configuration in the metadata log. This is used in Admin API requests (DescribeConfigs, IncrementalAlterConfigs, etc.).

public final class ConfigResource
    // ...
    public enum Type {
        // existing types unchanged 
        CLUSTER_MIRROR((byte) 64);
    }

Schema Field

The schema-level annotation for MirrorName string fields in protocol messages. The message 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);
}

ACL Resource

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

Configuration

This section describes new configurations introduced by the Cluster Mirroring feature.

Broker

Stored in server.properties or dynamic broker config.

Mirror

Set when creating/altering a specific mirror. Stored in cluster metadata.

Topic

Set by topic creation or alter. Stored in topic config.

Metrics

A core set of metrics will be provided with the initial implementation.

Errors

List of protocol-level errors returned by the new RPCs:

Compatibility, Deprecation, and Migration Plan

Cluster Mirroring requires a metadata version bump to support the new metadata records (MirrorTopicStateChangeRecord) and the LeaderEpoch field in PartitionChangeRecord (version 3).  A new feature gate called mirror.version controls whether Cluster Mirroring is enabled. When mirror.version is 0 (the default), all mirroring APIs are rejected with UNSUPPORTED_VERSION. Setting mirror.version to 1 enables the feature. The mirror.version feature gate requires the new metadata version as its minimum. Cluster Mirroring will be introduced through a phased rollout across multiple Kafka releases to ensure stability and gather community feedback.

Release Phases

Early Access

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.

Preview

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.

General Availability

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.

Compatibility Matrix

Note that some features require support from the source cluster.

Additional notes:

1. Mirror fetcher threads always use the explicit OffsetsForLeaderEpochRequest path for log truncation rather than truncation-on-fetch (diverging epoch in fetch responses). This ensures correct divergence detection regardless of the source cluster version, since older source brokers may not include diverging epoch info in fetch responses.

2. Sources older than Kafka 2.8 (pre-KIP-516) do not support topic IDs and return ZERO_UUID in metadata responses. In this case, the destination controller assigns a new topic ID, so topic identity is not preserved across clusters. Topic matching falls back to name-based lookup. Partition scaling and topic creation work correctly, but the source and destination will have different topic IDs for the same topic.
3. Share groups were introduced in Kafka 4.0 (KIP-932). When mirroring from older sources, share group offset sync is automatically skipped because the source Admin client does not support the share group listing API. A warning is logged but does not affect other sync operations.

Scalability Considerations

MirrorFetcherThread uses the same fetch protocol optimizations as ReplicaFetcherThread (fetch sessions, pipelining, compression, zero-copy). The performance impact on existing intra-cluster replication is minimized through resource isolation:

• Separate Thread Pools: Cross-cluster fetcher threads run in a dedicated thread pool, which is independent from the intra-cluster fetcher thread pool. This separation ensures that cross-cluster replication latency does not impact local replica synchronization.
• Network I/O Overhead: Read-only leaders perform additional network I/O to fetch from source clusters. This overhead is proportional to the number of mirror partitions and the replication throughput. Brokers with many mirror partitions may experience increased CPU usage for network processing and data serialization.
• Memory Footprint: Each mirror fetcher thread maintains its own fetch session state, partition state map, and response buffers. With default configuration, memory overhead is comparable to standard replica fetchers. The metadata manager maintains connection pools and metadata caches, adding minimal memory overhead.
• Bandwidth Consumption: Cross-cluster traffic between source and destination clusters consumes WAN bandwidth. For large-scale deployments, operators should provision adequate inter-datacenter connectivity or configure throttling.
• State Management: Mirror partition state management is evenly distributed to available brokers to avoid any hot spot, especially during rolling update or restart events.

At high partition counts (e.g., 100k partitions per broker), the in-memory cache that stores per-partition state, epoch, and failure metadata remains modest in heap footprint (tens of MB). The more significant pressure points are:

1. The periodic metadata refresh (default every 60s) issues batched RPCs to the source cluster (describeTopics, describeConfigs, listConsumerGroupOffsets), but at this scale the response payloads become large, adding serialization and processing overhead on both sides each cycle.
2. During large leadership reassignments (e.g., rolling restarts), MirrorMetadataManager must process deltas with tens of thousands of partition changes in a single invocation, building intermediate collections proportional to the delta size, and send ReadMirrorStatesRequest RPCs to coordinators, similar to the share partition leader approach.

These are known scalability characteristics in Kafka. For (1), users can increase the refresh interval to match their deployment size. For (2), the RPC payload grows proportionally with the number of partitions changing leadership at once. Both cases could benefit from a follow up KIP to introduce chunked requests or per-RPC size limits, allowing large operations to be split into smaller batches transparently.

Migration From MirrorMaker 2

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:

1. Stop MM2 replication.
2. Delete mirror topics on destination cluster, including MM2 internal topics.
3. Start fresh with Cluster Mirroring.

Test Plan

Unit Tests

Unit tests will cover individual component behavior:

• MirrorCoordinator: State loading, partition assignment, metadata persistence.
• MirrorMetadataManager: Topic creation, config sync, offset commit, ACL sync.
• MirrorFetcherThread: Epoch tracking, fetch processing, leader change handling.
• MirrorCommand: Command-line parsing, Admin API invocation, error handling.
• Protocol Serialization: Extended and new APIs serialization and deserialization.

Integration Tests

Integration tests will validate end-to-end functionality across multiple brokers:

• CLI Workflow: Create mirror with kafka-cluster-mirrors.sh, start mirroring, verify log convergence
• Basic Replication: Create mirror via API, start mirroring, verify log convergence
• Metadata Sync: Modify topic config in source, verify automatic sync to destination
• Partition Expansion: Add partitions to source topic, verify destination expands
• Consumer Groups: Commit offsets in source, verify replication to destination
• ACL Replication: Create ACL in source, verify creation in destination
• Leader Changes: Trigger leader election in source, verify fetcher reconnects
• Broker Failures: Stop destination broker, verify replication continues after recovery

System Tests

System tests will validate behavior under realistic production conditions:

• Log Convergence: Check log convergence after a series of leader elections in source and destination clusters.
• Failover Test: Simulate source cluster failure, measure consumer recovery time.
• Security Validation: Test all authentication mechanisms (SASL PLAIN, SCRAM, Kerberos, mTLS).
• Migration Test: Test migration from older Kafka versions.
• Scalability Test: Replicate 1000 topics with 100,000 partitions across clusters.
• Long-Running Stability: Run continuous replication for 7 days, verify no memory leaks or performance degradation.
• Performance Benchmark: Measure replication throughput and latency across WAN.

Rejected Alternatives

Use MirrorMaker 2

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:

• MirrorMaker 2: Best for active-active topologies, multi-region writes, complex routing scenarios
• Cluster Mirroring: Best for disaster recovery, failover, migration, simplified operations with exact offset preservation

Deployment Comparison

Feature Support Comparison

Performance & Isolation Comparison