Page History

...

A new dynamic topic level configuration mirror.support.unclean.leader.election (boolean, default false) is introduced to support ULE. When enabled, LME log truncation waits for all replicas (not just ISR members) to join the ISR and complete the truncation. This ensures that every replica has been truncated past the LME, so even subsequent unclean leader elections cannot introduce undetectable divergence. If some replicas cannot catch up due to slow network or disk issues, mirroring remains pending or enters the FAILED state, requiring manual intervention (resolving the issue or reassigning partition replicas to healthy brokers). This is a dynamic configuration, so it can be enabled at any point during the mirroring lifecycle. If it is enabled before mirroring starts, all records are guaranteed to be consistent between the two clusters, even if an unclean leader election occurs. If it is enabled when mirroring is already running, only records produced after the next LME log truncation are guaranteed to be consistent. If some replicas cannot catch up with the leader during LME log truncation due to slow network, disk issues, or other failures, the mirror partition moves to the FAILED state. In this case, users have to manually resolve the underlying issue or reassign the partition replicas to healthy brokers, and then restart mirroring.

Main Operations

Failover

Failover is initiated by calling the RemoveTopicsFromMirror API, which appends a ".removed" suffix to the mirror.name internal config. This transitions the mirror topics from read-only to writable state after the stopping process completes gracefully. When producers reconnect to the destination cluster after failover, they obtain new producer IDs which are separate from previously mirrored IDs, so they begin writing with fresh sequence numbers starting from 0. Consumers can reconnect to the destination cluster using the same group ID, resuming from the last synchronized offsets, minimizing data re-processing or gaps. The transition is transparent from the consumer's perspective and offset management continues normally through the destination's group coordinator.

# 9091 (source) -----> 9094 (destination)
# in case of disaster, the operator can failover by running the following command
bin/kafka-mirror.sh --bootstrap-server :9094 --remove --topic .* --mirror my-mirror
# 9091 (source) --x--> 9094 (destination)
# now all mirror topics are detached from the source cluster and accept writes (the two clusters are allowed to diverge)

Failback

Failback enables mirroring to be reversed after a failover, allowing the original source cluster to become the destination and vice versa. This is critical for scenarios where you want to fail back to the original cluster after recovering from an outage or planned maintenance. When failback is initiated on the old source cluster, it needs to determine where to truncate its log before starting to fetch from the new source cluster. If the new API is supported, the broker sends a LastMirrorredEpochs request to the new source cluster asking for the LME, and then truncates its local log to the last offset of the returned epoch. If the new API is not supported, the broker truncates to zero and starts mirroring from scratch.

Before transitioning a mirror partition from PREPARING to MIRRORING, the MirrorCoordinator must ensure that all in-sync replicas in the destination cluster have truncated their logs to the correct offset. If less than min ISR are available, we will skip and retry in the following fetch. This coordination step validates that every ISR member has completed truncation before the partition is allowed to begin actively fetching from the source cluster. Without it, the mirror leader could start appending new data from the source while local followers still hold divergent log segments, causing inconsistencies within the destination cluster. After truncation, reverse mirroring begins normally. Note that the log truncation on everse mirroring may cause the data loss if there are records that didn't get mirrored to the old destination cluster.

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

Existing 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 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 advances. For example, if a topic originally contained offsets 0-100 and retention deletes offsets 0-99, the source cluster's log start offset becomes 100. When the mirror leader fetches from the source, it discovers the new log start offset and updates its local log start offset to match, creating the same offset gap. If a mirror follower attempts to fetch from an offset below the source cluster's log start offset (e.g. fetching offset 50 when log start offset is 100), the source broker returns an OffsetOutOfRangeException. The mirror leader handles this by truncating its local log to the source's current log start offset and resuming fetching from that point. This ensures the destination cluster mirrors the current retention state of the source cluster without attempting to replicate already-deleted data.

Consumer Groups

Cluster Mirroring synchronizes consumer group offsets from the source cluster to the destination cluster, enabling consumers to resume consumption from their last committed offset after failover. The MirrorMetadataManager periodically fetches consumer group committed offsets from the source cluster and replicates it to the destination cluster's. This ensures that consumer groups maintain their consumption progress across both clusters. During offset synchronization, the committed offset in the destination cluster may temporarily exceed the current log end offset (LEO) of the mirror topic. For example, if a consumer commits offset 100 in the source cluster but the destination cluster has only mirrored up to offset 80 (LEO = 80), the MirrorMetadataManager still commits offset 100 to the destination cluster. This is acceptable because the mirror leader continues fetching data and the LEO will eventually advance to include offset 100. However, if a failover occurs before the mirrored data catches up, consumers attempting to resume from offset 100 will receive an OffsetOutOfRangeException. To handle this scenario gracefully, consumers should configure auto.offset.reset=latest when consuming from mirror topics. This ensures that if a committed offset is beyond the current LEO after failover, the consumer automatically resets to the latest available offset rather than failing or resetting to the earliest offset.

Security

Cluster Mirroring supports comprehensive security controls through both authorization and authentication mechanisms. This ensures that only authorized principals can establish and manage cluster mirrors. When configuring a mirror, operators specify ACLs that should be synchronized from the source cluster, and these ACLs are periodically replicated to the destination cluster to maintain consistent access control policies across both environments.

When connecting to the source cluster, Cluster Mirroring requires only the bootstrap server address and appropriate credentials, no other sensitive cluster information is exposed or required. The destination cluster's mirror configuration supports all standard Kafka authentication mechanisms including TLS/SSL for encrypted transport and SASL for client authentication. Each mirror can be configured with its own security settings, allowing different mirrors to connect to source clusters with varying security requirements. This enables secure cross-cluster replication even when source and destination clusters use different authentication protocols or when connecting across security boundaries such as on-premises to cloud environments. All credentials are stored as mirror configuration records in the destination cluster metadata log, and used exclusively for establishing authenticated connections to the source cluster.

Source cluster permissions (mirror principal):

...

RPC

...

Component

...

ACL Operation

...

ACL Resource

...

Purpose

...

Destination cluster permissions:

...

RPC

...

Component

...

ACL Operation

...

ACL Resource

...

Purpose

...

Existing 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 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 advances. For example, if a topic originally contained offsets 0-100 and retention deletes offsets 0-99, the source cluster's log start offset becomes 100. When the mirror leader fetches from the source, it discovers the new log start offset and updates its local log start offset to match, creating the same offset gap. If a mirror follower attempts to fetch from an offset below the source cluster's log start offset (e.g. fetching offset 50 when log start offset is 100), the source broker returns an OffsetOutOfRangeException. The mirror leader handles this by truncating its local log to the source's current log start offset and resuming fetching from that point. This ensures the destination cluster mirrors the current retention state of the source cluster without attempting to replicate already-deleted data.

Consumer Groups

Cluster Mirroring synchronizes consumer group offsets from the source cluster to the destination cluster, enabling consumers to resume consumption from their last committed offset after failover. The MirrorMetadataManager periodically fetches consumer group committed offsets from the source cluster and replicates it to the destination cluster's. This ensures that consumer groups maintain their consumption progress across both clusters. During offset synchronization, the committed offset in the destination cluster may temporarily exceed the current log end offset (LEO) of the mirror topic. For example, if a consumer commits offset 100 in the source cluster but the destination cluster has only mirrored up to offset 80 (LEO = 80), the MirrorMetadataManager still commits offset 100 to the destination cluster. This is acceptable because the mirror leader continues fetching data and the LEO will eventually advance to include offset 100. However, if a failover occurs before the mirrored data catches up, consumers attempting to resume from offset 100 will receive an OffsetOutOfRangeException. To handle this scenario gracefully, consumers should configure auto.offset.reset=latest when consuming from mirror topics. This ensures that if a committed offset is beyond the current LEO after failover, the consumer automatically resets to the latest available offset rather than failing or resetting to the earliest offset.

Security

Cluster Mirroring supports comprehensive security controls through both authorization and authentication mechanisms. This ensures that only authorized principals can establish and manage cluster mirrors. When configuring a mirror, operators specify ACLs that should be synchronized from the source cluster, and these ACLs are periodically replicated to the destination cluster to maintain consistent access control policies across both environments.

When connecting to the source cluster, Cluster Mirroring requires only the bootstrap server address and appropriate credentials, no other sensitive cluster information is exposed or required. The destination cluster's mirror configuration supports all standard Kafka authentication mechanisms including TLS/SSL for encrypted transport and SASL for client authentication. Each mirror can be configured with its own security settings, allowing different mirrors to connect to source clusters with varying security requirements. This enables secure cross-cluster replication even when source and destination clusters use different authentication protocols or when connecting across security boundaries such as on-premises to cloud environments. All credentials are stored as mirror configuration records in the destination cluster metadata log, and used exclusively for establishing authenticated connections to the source cluster.

Source cluster permissions (mirror principal):

Destination cluster permissions:

An operator can grant ClusterMirror:*:CREATE,ALTER,DESCRIBE for full mirror management, or scope it to specific mirrors like ClusterMirror:prod-dr:DESCRIBE for read-only monitoring of a single mirror, without granting any broker-level privileges.

Inter-broker coordinator RPCs (WriteMirrorStates, ReadMirrorStates, LastMirroredEpochs, BumpLeaderEpochs, and FindCoordinator) require CLUSTER_ACTION on the Cluster resource, as they are only issued by the broker service account.

CreatePartitions, OffsetCommit, IncrementalAlterConfigs, CreateAcls, and DeleteAcls are issued internally by MMM through the inter-broker channel or direct coordinator calls, bypassing normal ACL checks. No explicit ACL grants are needed for these operations.

Idempotent Producer

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

Image Added

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

• STOPPING: Remove fetchers, truncate to LSO, persist LME, write MIRROR_PID_RESET control batch.
• STOPPED: Partition is writable (terminal state, no actions).

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

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

The SourceClusterId field records which source cluster the mirrored data came from, enabling future validation (e.g. detecting unexpected source cluster changes) and data provenance tracing from the log itself.

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

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 MirrorPidResetRecord expires all PSM entries. Local producers get fresh PIDs from the coordinator with no collision risk.
• Failback (A to B, then B to A): After failover, B becomes writable. Later, A starts mirroring from B and truncates its log to the LSO. A then stores B's records as-is. B's MirrorPidResetRecord 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 batch is encountered during append or during log recovery, all producer entries are removed from the PSM. A will write its own MirrorPidResetRecord 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 MirrorPidResetRecord independently.
• Fan-in (A to C, B to C, different topics): Each topic's partitions have independent PSMs. The MirrorPidResetRecord 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 MirrorPidResetRecord expires all PSM entries on the stopping node. Longer chains work inductively by the same principle.

Exactly-Once Semantics

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

During the mirror stopping transition, the MirrorCoordinator performs a log truncation operation that resets each mirror partition to its LSO. This offset represents the point in the log where all transactions have been decided (committed or aborted), essentially the highest offset where data is known to be consistent from a transactional perspective. Any records beyond this point may belong to incomplete transactions and should not persist after mirroring stops. Note that the actual lag may be greater than what's reported by the metrics. This approach prevents a critical consistency issue: the destination cluster could retain partial transaction data that would never be completed since mirroring has stopped. This would leave the topic in an inconsistent state where read_committed consumers may be blocked due to incomplete transaction data. Additionally, the transaction coordinator would not be able to rollback these hanging transactions because there would be no __transaction_state metadata in the destination cluster.

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

Consider this source cluster log:

If replication reaches offset 4 and the source cluster fails, the destination cluster contains data records for transaction 4002 (offset 2) without the abort marker (offset 4). This creates a hanging transaction that can never be committed or aborted on the destination cluster. Note that this approach causes data loss for any in-flight transactions or non-mirrored completed transactions when we experiencing a lag during the failover and may result in already-processed records being lost if consumers on the destination cluster read uncommitted data.

Bandwidth Control

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

1. Destination Cluster Throttling: To avoid conflicts with intra-cluster replication controls, mirror-specific throttling configurations operate independently from standard replication throttling. The system provides two configuration levels: a broker-level rate limit (mirror.replication.throttled.rate) that sets the overall bandwidth ceiling for mirror replication traffic, and a topic-level replica list (mirror.replication.throttled.replicas) that specifies which partition-broker combinations should be throttled using the standard partition-index:broker-id notation. Operators can dynamically adjust throttling rates at runtime without restarting brokers, first setting a cluster-wide default rate, then fine-tuning specific topic partitions as mirroring progresses. This allows gradual bandwidth allocation as mirror relationships are established.
2. Source Cluster Throttling: The source cluster side requires a different approach because mirror fetch requests operate as consumer traffic rather than replication traffic. This design is intentional since the mirroring must fetch only up to the LSO to maintain transactional consistency, which is a consumer-level guarantee not available through the replication protocol. Consequently, 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.

Tiered Storage

Mirror topics in the destination cluster currently only replicate data from local storage on the source broker. Integrating with tiered storage would allow mirroring to handle data that has been offloaded to remote storage (e.g., S3, HDFS), enabling full replication of topics with long retention periods without requiring all data to reside in local broker storage. A detailed design of the metadata synchronization protocol, API schema, and state management will be provided in a follow-up KIP.

Share Group

Cluster Mirroring supports both traditional consumer groups and share consumer groups (Kafka Queue functionality) to ensure seamless failover for all consumer types. While the data mirroring mechanism remains identical, the offset synchronization strategy differs based on the group type. Share consumer groups use a different offset management model based on Share-Partition Start Offset (SPSO) and Share-Partition End Offset (SPEO) rather than traditional committed offsets. First we retrieve the current SPSO for each share group using the DescribeShareGroupOffsets API from the source cluster, and then we update the SPSO in the destination cluster using the AlterShareGroupOffsets API, which also initializes the group state in both the group coordinator and share coordinator. This means the API can initialize a share group in the destination cluster even if it doesn't exist yet, eliminating the need for pre-creation or complex state management.

Kafka enforces that consumer group and share group names must be unique within a single cluster. This creates a potential conflict scenario during mirroring. When such conflicts occur, the offset commit operation will fail with GroupIdNotFoundException. Users must resolve these conflicts manually by either deleting the conflicting group in the destination cluster before mirroring begins, or excluding the conflicting groups from offset synchronization. These conflicts affect only offset synchronization and do not impact data mirroring itself. The topic data continues to replicate normally, and only the automatic offset synchronization for the conflicting groups is blocked.

Active-Active Writes

Active-active topology is not initially supported in Cluster Mirroring, though it could potentially be achieved through topic prefixing and removing the reliance on topic ID for mirroring. This is a candidate for a future improvement KIP. Instead, bidirectional mirroring is supported, but only when mirroring different topics between clusters, allowing records produced to either cluster to be consumed from both. Unlike MirrorMaker 2, Cluster Mirroring does not need special cycle detection or prevention logic because the read-only enforcement inherently blocks the conditions that would create infinite replication loops.

Synchronous Mirroring

Currently, mirroring is asynchronous. The source cluster acknowledges the producer without waiting for the destination to replicate the data. Sync mirroring would guarantee that records are replicated to the destination cluster before the source acknowledges the produce request, providing stronger durability guarantees at the cost of higher latency. This would be useful for workloads where zero data loss across clusters is a strict requirement.

Future extensions to synchronous mirroring could enable preservation of transactional semantics across clusters. Streaming platforms using exactly-once mode (Apache Kafka Streams, Apache Flink, Apache Spark) rely on the source cluster's transactional protocol and coordination. During failover or migration scenarios, transactional metadata for pending transactions does not transfer to the destination cluster, potentially breaking exactly-once guarantees. Supporting transactional cross-cluster replication would require coordinating transactional metadata and ensuring transaction state consistency across clusters, something MM2's Connect-based architecture cannot support.

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.

Mirror Operations

Failover Process

Failover is initiated by calling the RemoveTopicsFromMirror API, which appends a ".removed" suffix to the mirror.name internal config. This transitions the mirror topics from read-only to writable state after the stopping process completes gracefully. When producers reconnect to the destination cluster after failover, they obtain new producer IDs which are separate from previously mirrored IDs, so they begin writing with fresh sequence numbers starting from 0. Consumers can reconnect to the destination cluster using the same group ID, resuming from the last synchronized offsets, minimizing data re-processing or gaps. The transition is transparent from the consumer's perspective and offset management continues normally through the destination's group coordinator.

# 9091 (source) -----> 9094 (destination)
# in case of disaster, the operator can failover by running the following command
bin/kafka-mirror.sh --bootstrap-server :9094 --remove --topic .* --mirror my-mirror
# 9091 (source) --x--> 9094 (destination)
# now all mirror topics are detached from the source cluster and accept writes (the two clusters are allowed to diverge)

Failback Process

Failback enables mirroring to be reversed after a failover, allowing the original source cluster to become the destination and vice versa. This is critical for scenarios where you want to fail back to the original cluster after recovering from an outage or planned maintenance. When failback is initiated on the old source cluster, it needs to determine where to truncate its log before starting to fetch from the new source cluster. If the new API is supported, the broker sends a LastMirrorredEpochs request to the new source cluster asking for the LME, and then truncates its local log to the last offset of the returned epoch. If the new API is not supported, the broker truncates to zero and starts mirroring from scratch.

Before transitioning a mirror partition from PREPARING to MIRRORING, the MirrorCoordinator must ensure that all in-sync replicas in the destination cluster have truncated their logs to the correct offset. If less than min ISR are available, we will skip and retry in the following fetch. This coordination step validates that every ISR member has completed truncation before the partition is allowed to begin actively fetching from the source cluster. Without it, the mirror leader could start appending new data from the source while local followers still hold divergent log segments, causing inconsistencies within the destination cluster. After truncation, reverse mirroring begins normally. Note that the log truncation on everse mirroring may cause the data loss if there are records that didn't get mirrored to the old destination cluster.

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

Create Mirror Workflow

...

An operator can grant ClusterMirror:*:CREATE,ALTER,DESCRIBE for full mirror management, or scope it to specific mirrors like ClusterMirror:prod-dr:DESCRIBE for read-only monitoring of a single mirror, without granting any broker-level privileges.

Inter-broker coordinator RPCs (WriteMirrorStates, ReadMirrorStates, LastMirroredEpochs, BumpLeaderEpochs, and FindCoordinator) require CLUSTER_ACTION on the Cluster resource, as they are only issued by the broker service account.

CreatePartitions, OffsetCommit, IncrementalAlterConfigs, CreateAcls, and DeleteAcls are issued internally by MMM through the inter-broker channel or direct coordinator calls, bypassing normal ACL checks. No explicit ACL grants are needed for these operations.

Idempotent Producer

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

Image Removed

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

• STOPPING: Remove fetchers, truncate to LSO, persist LME, write MIRROR_PID_RESET control batch.
• STOPPED: Partition is writable (terminal state, no actions).

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

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

The SourceClusterId field records which source cluster the mirrored data came from, enabling future validation (e.g. detecting unexpected source cluster changes) and data provenance tracing from the log itself.

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

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 MirrorPidResetRecord expires all PSM entries. Local producers get fresh PIDs from the coordinator with no collision risk.
• Failback (A to B, then B to A): After failover, B becomes writable. Later, A starts mirroring from B and truncates its log to the LSO. A then stores B's records as-is. B's MirrorPidResetRecord 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 batch is encountered during append or during log recovery, all producer entries are removed from the PSM. A will write its own MirrorPidResetRecord 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 MirrorPidResetRecord independently.
• Fan-in (A to C, B to C, different topics): Each topic's partitions have independent PSMs. The MirrorPidResetRecord 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 MirrorPidResetRecord expires all PSM entries on the stopping node. Longer chains work inductively by the same principle.

Exactly-Once Semantics

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

During the mirror stopping transition, the MirrorCoordinator performs a log truncation operation that resets each mirror partition to its LSO. This offset represents the point in the log where all transactions have been decided (committed or aborted), essentially the highest offset where data is known to be consistent from a transactional perspective. Any records beyond this point may belong to incomplete transactions and should not persist after mirroring stops. Note that the actual lag may be greater than what's reported by the metrics. This approach prevents a critical consistency issue: the destination cluster could retain partial transaction data that would never be completed since mirroring has stopped. This would leave the topic in an inconsistent state where read_committed consumers may be blocked due to incomplete transaction data. Additionally, the transaction coordinator would not be able to rollback these hanging transactions because there would be no __transaction_state metadata in the destination cluster.

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

Consider this source cluster log:

If replication reaches offset 4 and the source cluster fails, the destination cluster contains data records for transaction 4002 (offset 2) without the abort marker (offset 4). This creates a hanging transaction that can never be committed or aborted on the destination cluster. Note that this approach causes data loss for any in-flight transactions or non-mirrored completed transactions when we experiencing a lag during the failover and may result in already-processed records being lost if consumers on the destination cluster read uncommitted data.

Bandwidth Control

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

1. Destination Cluster Throttling: To avoid conflicts with intra-cluster replication controls, mirror-specific throttling configurations operate independently from standard replication throttling. The system provides two configuration levels: a broker-level rate limit (mirror.replication.throttled.rate) that sets the overall bandwidth ceiling for mirror replication traffic, and a topic-level replica list (mirror.replication.throttled.replicas) that specifies which partition-broker combinations should be throttled using the standard partition-index:broker-id notation. Operators can dynamically adjust throttling rates at runtime without restarting brokers, first setting a cluster-wide default rate, then fine-tuning specific topic partitions as mirroring progresses. This allows gradual bandwidth allocation as mirror relationships are established.
2. Source Cluster Throttling: The source cluster side requires a different approach because mirror fetch requests operate as consumer traffic rather than replication traffic. This design is intentional since the mirroring must fetch only up to the LSO to maintain transactional consistency, which is a consumer-level guarantee not available through the replication protocol. Consequently, 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.

Tiered Storage

Mirror topics in the destination cluster currently only replicate data from local storage on the source broker. Integrating with tiered storage would allow mirroring to handle data that has been offloaded to remote storage (e.g., S3, HDFS), enabling full replication of topics with long retention periods without requiring all data to reside in local broker storage. A detailed design of the metadata synchronization protocol, API schema, and state management will be provided in a follow-up KIP.

Share Group

Cluster Mirroring supports both traditional consumer groups and share consumer groups (Kafka Queue functionality) to ensure seamless failover for all consumer types. While the data mirroring mechanism remains identical, the offset synchronization strategy differs based on the group type. Share consumer groups use a different offset management model based on Share-Partition Start Offset (SPSO) and Share-Partition End Offset (SPEO) rather than traditional committed offsets. First we retrieve the current SPSO for each share group using the DescribeShareGroupOffsets API from the source cluster, and then we update the SPSO in the destination cluster using the AlterShareGroupOffsets API, which also initializes the group state in both the group coordinator and share coordinator. This means the API can initialize a share group in the destination cluster even if it doesn't exist yet, eliminating the need for pre-creation or complex state management.

Kafka enforces that consumer group and share group names must be unique within a single cluster. This creates a potential conflict scenario during mirroring. When such conflicts occur, the offset commit operation will fail with GroupIdNotFoundException. Users must resolve these conflicts manually by either deleting the conflicting group in the destination cluster before mirroring begins, or excluding the conflicting groups from offset synchronization. These conflicts affect only offset synchronization and do not impact data mirroring itself. The topic data continues to replicate normally, and only the automatic offset synchronization for the conflicting groups is blocked.

Active-Active Writes

Active-active topology is not initially supported in Cluster Mirroring, though it could potentially be achieved through topic prefixing and removing the reliance on topic ID for mirroring. This is a candidate for a future improvement KIP. Instead, bidirectional mirroring is supported, but only when mirroring different topics between clusters, allowing records produced to either cluster to be consumed from both. Unlike MirrorMaker 2, Cluster Mirroring does not need special cycle detection or prevention logic because the read-only enforcement inherently blocks the conditions that would create infinite replication loops.

Synchronous Mirroring

Currently, mirroring is asynchronous. The source cluster acknowledges the producer without waiting for the destination to replicate the data. Sync mirroring would guarantee that records are replicated to the destination cluster before the source acknowledges the produce request, providing stronger durability guarantees at the cost of higher latency. This would be useful for workloads where zero data loss across clusters is a strict requirement.

Future extensions to synchronous mirroring could enable preservation of transactional semantics across clusters. Streaming platforms using exactly-once mode (Apache Kafka Streams, Apache Flink, Apache Spark) rely on the source cluster's transactional protocol and coordination. During failover or migration scenarios, transactional metadata for pending transactions does not transfer to the destination cluster, potentially breaking exactly-once guarantees. Supporting transactional cross-cluster replication would require coordinating transactional metadata and ensuring transaction state consistency across clusters, something MM2's Connect-based architecture cannot support.

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.

Command Workflows

...

1. The user sends CreateMirror requests to any broker with the mirror name and mirror related properties (bootstrap servers, security settings, etc.).
2. The broker forwards the request to the active controller.
3. The controller saves the properties into the metadata log as ConfigRecord entries with ConfigResource(Type.MIRROR, mirrorName).
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 the MirrorMetadataManager registers the new mirror configuration.

Add Topics to Mirror Workflow

1. User sends AddTopicsToMirror request with topics and mirror name.

2. The broker forwards to the active controller.

3. The controller validates that each topic exists and is not already in a mirror. It then sets the topic config mirror.name=<mirrorName> for each topic, generating a ConfigRecord per topic into the metadata log.
4. Response is sent back to clients with per topic results.
5. When the MirrorMetadataManager in the partition leader node gets notified about the topic config update, it detects that mirror.name is not empty and has no .removed or .paused suffix. It then queries the current mirror partition state from the coordinator. The coordinator could be located on a different broker node, so a ReadMirrorStates inter broker RPC may be needed.
6. Based on the current mirror partition state, the state machine transitions the partition. In most cases, from UNKNOWN to PREPARING.
7. During PREPARING, the mirror fetcher performs Last Mirrored Epoch (LME) truncation. The LME is the greatest leader epoch that the source cluster recognizes from the destination. If the source has no LME knowledge (first time mirroring), it returns -1 and the destination truncates everything and replicates from scratch. Otherwise, the destination truncates at the start offset of the first epoch beyond the LME. It then waits until all ISR members (or all replicas if mirror.support.unclean.leader.election=true) complete the truncation.
8. Once all ISR members have completed truncation, the state transitions from PREPARING to MIRRORING. A MirrorFetcherThread is created and starts sending consumer Fetch requests (not follower requests) to the source cluster to replicate data. The Fetch protocol handles any offset level divergence by truncating to the exact offset where the source epoch ends.

9. The fetched batch retains its original leader epoch from the source. When the partition later becomes writable after failover, the leader epoch is bumped to ensure monotonically increasing epochs for new records.

10. The partition state is persisted to the __mirror_state topic on each state change via local append or WriteMirrorStates (when coordinator is remote) as MirrorPartitionStateKey/MirrorPartitionStateValue records, distributed by hash(mirrorName, topicId, partition) % numPartitions.
11. The MirrorMetadataManager also periodically synchronizes topic configs, consumer group offsets, and ACLs from the source cluster.

Remove Topics from Mirror Workflow

1. User sends RemoveTopicsFromMirror request with topics and mirror name.
2. The controller validates each topic belongs to the specified mirror and is in MIRRORING state. It then updates the topic config by appending the .removed suffix, e.g. mirror.name=cluster1.removed, generating a ConfigRecord.
3. When the MirrorMetadataManager in the partition leader node gets notified, it detects the .removed suffix on mirror.name. It queries the current mirror partition state from the coordinator, and transitions to STOPPING.
4. During STOPPING:
   1. The MirrorFetcherManager removes all fetcher threads for the affected partitions, stopping replication.
   2. Bump the leader epoch for the partitions to ensure monotonically increasing epochs for new records.
   3. The log is truncated to LSO for transactional consistency.
   4. The LME is recorded as LastMirroredEpochsKey/LastMirroredEpochsValue records into the __mirror_state topic for potential future failback.
   5. A MIRROR_PID_RESET control batch is written to the partition log, which expires all ProducerStateManager entries so that new producers get fresh PIDs with no collision risk.
5. The state transitions from STOPPING to STOPPED. The read only flag is cleared and the topic becomes writable. New producers can start producing with fresh PIDs starting at sequence 0 and a higher leader epoch.

Pause Topics Workflow

1. User sends PauseMirrorTopics request with topics and mirror name.
2. The controller validates each topic belongs to the specified mirror and is currently in MIRRORING state. It appends the .paused suffix to the mirror name config, e.g. mirror.name=cluster1.paused, generating a ConfigRecord.
3. When the MirrorMetadataManager in the partition leader node gets notified, it detects the .paused suffix. It transitions the state to PAUSING.
4. During PAUSING, the MirrorFetcherManager removes the fetcher threads for the affected partitions. No more data is replicated.
5. The state transitions from PAUSING to PAUSED. The partition remains read only. Metadata synchronization (configs, groups, ACLs) is also halted for the paused topics.
6. The partition state change is persisted to the __mirror_state topic.

Resume Topics Workflow

1. User sends ResumeMirrorTopics request with topics and mirror name.
2. The controller validates the topic is currently paused (has .paused suffix). It removes the .paused suffix, restoring the original mirror name, e.g. mirror.name=cluster1, generating a ConfigRecord.
3. When the MirrorMetadataManager in the partition leader node gets notified, it detects that mirror.name no longer has the .paused suffix.
4. The state transitions directly from PAUSED to MIRRORING. No log truncation is needed because the partition is already at the correct offset from before the pause.
5. New MirrorFetcherThread instances are created and resume replication from the current log end offset.
6. Metadata synchronization (configs, groups, ACLs) also resumes.

Delete Mirror Workflow

1. The user sends a DeleteMirror request with the mirror name.
2. The controller validates that the mirror is empty (no topics assigned) or all its partitions are in STOPPED state.
3. If valid, the controller tombstones the mirror configuration in the cluster metadata log, removing all ConfigRecord entries for the mirror.
4. All mirror state entries in __mirror_state for this mirror are cleaned up.
5. Any remaining coordinator state is shut down, source cluster connections are closed, and the mirror name becomes available for reuse.
6. After deletion, failback using this mirror configuration is no longer possible.

List Mirrors Workflow

1. The user sends ListMirrorsRequest to any broker (no parameters required).

2. The broker handler gets all configured mirror partitions from, which reads from the in memory metadata cache.
3. For each authorized mirror, the broker returns: mirror name, source cluster ID, source bootstrap servers, and topic count.
4. No metadata records are written. This is a read only operation against the local metadata cache.

Describe Mirrors Workflow

1. The user sends DescribeMirrorsRequest with optional mirror names (empty means all mirrors).

2. The broker handler queries two sources:

   1. The ReplicaManager which provides source offset, destination offset, and lag for each partition.

   2. The MirrorCoordinator which provides the current partition state from the metadata manager cache.

3. The request is forwarded to each broker that only reports partitions for which it has lag information or is the partition leader. This avoids duplicate reporting across brokers.

4. For each partition, the response includes: mirror name, topic name, partition ID, source offset, destination offset, lag, and current state.
5. No metadata records are written. This is a read only operation.

...

// new added field in PartitionData type
{ "name": "MirrorLeaderEpoch", "type": "int32", "versions": "19+", "default": "-1", "taggedVersions": "19+", "tag": 3, "ignorable": true,
  "about": "The latest known mirrored leader epoch." },

CreateMirror

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

...

{
  "apiKey": 103,
  "type": "response",
  "name": "ResumeMirrorTopicsResponse",
  // Version 0 is the initial version.
  "validVersions": "0",
  "flexibleVersions": "0+",
  "fields": [
    { "name": "ThrottleTimeMs", "type": "int32", "versions": "0+",
      "about": "The duration in milliseconds for which the request was throttled due to a quota violation, or zero if the request did not violate any quota." },
    { "name": "ErrorCode", "type": "int16", "versions": "0+",
      "about": "The error code, or 0 if there was no error." },
    { "name": "ErrorMessage", "type": "string", "versions": "0+", "nullableVersions": "0+", "default": "null",
      "about": "The top-level error message, or null if there was no error." },
    { "name": "MirrorName", "type": "string", "versions": "0+", "entityType": "mirrorName",
      "about": "The cluster mirror name." },
    { "name": "Topics", "type": "[]TopicResult", "versions": "0",
      "about": "The results for the topics.", "fields": [
      { "name": "Name", "type": "string", "versions": "0", "entityType": "topicName",
        "about": "The topic name." },
      { "name": "ErrorCode", "type": "int16", "versions": "0",
        "about": "The error code, or 0 if there was no error." }
    ]}
  ]
}

DeleteMirror

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": 104,
  "type": "response",
  "name": "DeleteMirrorResponse",
  // Version 0 is the initial version.
  "validVersions": "0",
  "flexibleVersions": "0+",
  "fields": [
    { "name": "ThrottleTimeMs", "type": "int32", "versions": "0+",
      "about": "The duration in milliseconds for which the request was throttled due to a quota violation, or zero if the request did not violate any quota." },
    { "name": "ErrorCode", "type": "int16", "versions": "0+",
      "about": "The error code, or 0 if there was no error." },
    { "name": "ErrorMessage", "type": "string", "versions": "0+", "nullableVersions": "0+",
      "about": "The error message, or null if there was no error." }
  ]
}

ListMirrors

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": "response",
  "name": "ListMirrorsResponse",
  // Version 0 is the initial version.
  "validVersions": "0",
  "flexibleVersions": "0+",
  "fields": [
    { "name": "ThrottleTimeMs", "type": "int32", "versions": "0+",
      "about": "The duration in milliseconds for which the request was throttled due to a quota violation, or zero if the request did not violate any quota." },
    { "name": "ErrorCode", "type": "int16", "versions": "0+",
      "about": "The error code, or 0 if there was no error." },
    { "name": "ErrorMessage", "type": "string", "versions": "0+", "nullableVersions": "0+", "default": "null",
      "about": "The top-level error message, or null if there was no error." },
    { "name": "Mirrors", "type": "[]ListedMirror", "versions": "0+",
      "about": "Each mirror in the response.", "fields": [
      { "name": "MirrorName", "type": "string", "versions": "0+", "entityType": "mirrorName",
        "about": "The cluster mirror name." },
      { "name": "SourceBootstrap", "type": "string", "versions": "0+",
        "about": "The source cluster bootstrap servers." },
      { "name": "SourceClusterId", "type": "string", "versions": "0+", "default": "",
        "about": "The source cluster ID, or empty if not yet resolved." },
      { "name": "TopicCount", "type": "int32", "versions": "0+", "default": "0",
        "about": "The number of topics configured for this mirror. 0 indicates an empty mirror with no topics." }
    ]}
  ]
}

DescribeMirrors

Returns the current mirroring status, state, and configuration for the specified mirror topics on the destination cluster.

...