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

Status

Current state: Under Discussion

...

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

Motivation

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.0 (MM2) provides cross-cluster replication capabilities, it presents significant operational challenges.

• External System Management: MM2 runs as standalone Connect workers external to Kafka brokers, requiring separate deployment, monitoring, and lifecycle management. Administrators must provision additional hosts, manage Connect-specific configurations, and coordinate MM2 upgrades independently from Kafka broker upgrades.
• Compression Cost: 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 target 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.

Goals

Cluster Mirroring addresses these operational challenges by integrating cross-cluster replication directly into Kafka brokers, providing a simpler and more robust solution for cross-cluster replication.

...

While Cluster Mirroring is optimized for geo-replication, disaster recovery and migration use cases where a single source cluster replicates to one or more destination clusters, its coordinator-based architecture provides a foundation for more complex topologies.

Non-Goals

Synchronous Mirroring

This proposal describes asynchronous replication between clusters. Support for synchronous replication is deferred to future work.

...

Stretched clusters are not suitable for disaster recovery scenarios because they provide no protection against software failures or configuration incidents. Vendors that recommend stretched cluster deployments typically position them for high availability (HA) rather than DR, and notably, most do not offer stretched clusters as a managed service option, further underscoring the operational challenges and limited DR effectiveness of this architecture.

Unclean Leader Election

This proposal does not support unclean leader elections because there is no way to reconcile log divergence between source and destination clusters without a shared leader epoch. When the unclean.leader.election.enable is set to true, the broker will log a warning at every configuration synchronization period.

...

Solving this issue would require creating a shared leader epoch between source and destination clusters. Every time there is a source leader election we would need to notify the destination cluster and append data only after receiving a reply. This means that the overall latency would be cross-cluster replication latency plus intra-cluster replication latency. Read more in the Rejected Alternatives section.

Proposed Changes

Cluster Mirroring introduces a coordinator-based architecture integrated into Kafka brokers for managing cross-cluster replication. The design consists of three primary components that work together to provide automatic metadata synchronization and data replication. The following diagram illustrates how these components are wired together.

...

Brokers monitor these configuration changes to detect when partitions they lead belong to a mirror, triggering the creation of mirror fetchers and enforcement of read-only semantics. This design ensures that mirror associations are visible, auditable, and manageable through standard Kafka configuration introspection tools while maintaining strict control over how mirroring relationships are established and modified.

Main Components

MirrorCoordinator

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

...

Restarting a stopped mirror (STOPPED -> PREPARING -> MIRRORING): The mirror.name config is set again. onMetadataUpdate sees the partition in STOPPED state and transitions to PREPARING, re-truncating and resuming replication.

MirrorMetadataManager

The MirrorMetadataManager (MMM) implements periodic metadata synchronization between source and destination clusters. It maintains persistent network connections to all source clusters.

...

• The mirror.acl.include config controls which ACLs are synced using semicolon-separated rules with the format resourceType;resourceName;operation;permissionType;principal. Use * as a wildcard for any field. The resourceName and principal fields support regex. Trailing wildcard fields can be omitted. Default: * (all ACLs).
• * (all ACLs by default)
• TOPIC;orders.* (all ACLs for topics matching orders.*)
• *;*;*;*;User:alice (all ACLs for principal User:alice)
• *;*;*;*;User:app-.* (all ACLs for principals matching User:app-.*)
• TOPIC;*;READ;ALLOW (all topic READ/ALLOW ACLs)
• GROUP;consumer-.*;READ;ALLOW;User:bob (READ/ALLOW ACLs on groups matching consumer-.* for User:bob)
• TOPIC;orders.*,*;*;*;*;User:alice (sync all topic ACLs for orders.* topics and all ACLs for User:alice)

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:

...

When users remove a topic from the mirror, the partition will be removed from the fetch thread, and any late fetch responses will be skipped because the partition is not registered anymore in the fetcher thread.

Failover Process 

Failover is initiated by calling the RemoveTopicsFromMirror API, which appends a .removed suffix into the mirror.name internal config. This transitions the mirror topics from read-only to writable state after the stopping process completes gracefully.

...

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.

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.

...

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 the reverse mirroring may cause the data loss for the records that didn’t get mirrored to the old destination cluster earlier.

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

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.

...

To handle this scenario gracefully, consumers should configure auto.offset.reset=latest when consuming from mirrored 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. On the destination cluster, mirror-related operations (creating mirrors, adding/removing topics from mirrors, managing mirror configurations) require the CLUSTER_ACTION permission on the cluster resource. This ensures that only authorized principals can establish and manage cluster mirrors. When configuring a mirror, administrators 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.

...

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 in the destination cluster's mirror configuration and used exclusively for establishing authenticated connections to the source cluster.

Idempotent Producer

The idempotent producers rely on producer IDs to detect duplicate writes and ensure idempotent production. To avoid conflicts with the destination cluster's producer ID space, we rewrite source producer IDs to occupy the unused negative space by applying the formula: 

...

When a mirror topic becomes writable during failover, records with transformed producer IDs (<= -2) remain in the log with their original sequence numbers and epochs. Applications that reconnect to the destination cluster receive new producer IDs (>=0) from the destination's transaction coordinator, allowing them to continue producing.

Exactly-Once Semantics

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

...

Note that this approach causes data loss for any in-flight transactions 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 that sets the overall bandwidth ceiling for mirror replication traffic, and a topic-level replica list that specifies which partition-broker combinations should be throttled using the standard partition-index and 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.

Tiered Storage

Tiered Storage is not initially supported, but a detailed design of the metadata synchronization protocol, API schema, and state management will be provided in a follow-up KIP. A mirror follower that receives an OffsetMovedToTieredStorageException from the source leader handles it by marking the partition as failed, and also the mirror partition state will move to FAILED state.

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.

...

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.

Diskless Topics

At the time of writing, the Diskless Topics KIP (KIP-1500 and other sub-KIPs) are still under discussion, so there will be future KIPs to support this feature.

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.

Public Interfaces

Command-Line

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

...

$ bin/kafka-configs.sh --bootstrap-server :9091 --alter --add-config 'consumer_byte_rate=1024' \
  --entity-type clients --entity-name broker-4-fetcher-0-mirror-my-mirror
Completed updating config for client broker-4-fetcher-0-mirror-my-mirror.

Admin Client

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

CreateMirrorResult createMirror(String mirrorName, Map<String, String> configs, CreateMirrorOptions options);

AddTopicsToMirrorResult addTopicsToMirror(Map<String, String> topicToMirrorName, AddTopicsToMirrorOptions options);

RemoveTopicsFromMirrorResult removeTopicsFromMirror(String mirrorName, Set<String> topics, RemoveTopicsFromMirrorOptions options);

ListMirrorsResult listMirrors(ListMirrorsOptions options);

DescribeMirrorsResult describeMirrors(Collection<String> mirrorNames, DescribeMirrorsOptions options);

Protocol Changes

This KIP extends CreateTopic API, but also introduces some new APIs and metadata records.

CreateTopic

The CreateTopic API is updated to add information required for mirror topic creation.

...

1. This topic ID is not used by other topics in the current 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).

CreateMirror

The CreateMirror API 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 the .removed suffix. Once validated, the request is forwarded to the controller, which persists the configuration in the metadata log.

CreateMirrorRequest

{ 
  "apiKey": TBD, 
  "type": "request", 
  "listeners": ["broker", "controller"], 
  "name": "CreateMirrorRequest", 
  "validVersions": "0", 
  "flexibleVersions": "0+", 
  "fields": [ 
    { "name": "MirrorName", "type": "string", "versions": "0+", "nullableVersions": "0+", 
      "about": "The cluster mirror name."}, 
    { "name": "Config", "type": "[]MirrorConfig", "versions": "0+", 
      "about": "The cluster mirror configurations.",  "fields": [ 
      { "name": "Name", "type": "string", "versions": "0+", "mapKey": true, 
        "about": "The configuration key name." }, 
      { "name": "Value", "type": "string", "versions": "0+", "nullableVersions": "0+", 
        "about": "The value to set for the configuration key."} 
    ]} 
  ] 
} 

CreateMirrorResponse

{ 
  "apiKey": TBD, 
  "type": "response", 
  "name": "CreateMirrorResponse", 
  "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+", "ignorable": true, 
      "about": "The error message, or null if there was no error." } 

  ] 

}

AddTopicsToMirror

The AddTopicsToMirror API adds topics to a specified mirror. The broker validates that all target topic partitions are in either UNKNOWN or STOPPED state; otherwise, the request is rejected with an INVALID_REQUEST error. Once validated, the request is forwarded to the controller, which sets the mirror.name topic config to the specified mirror name.

AddTopicsToMirrorRequest

{
  "apiKey":TBD,
  "type": "request",
  "listeners": ["broker", "controller"],
  "name": "AddTopicsToMirrorRequest",
  "validVersions": "0",
  "flexibleVersions": "0+",
  "fields": [
    { "name": "Topics", "type": "[]TopicState", "versions": "0+", "about": "The topic state.",
      "fields": [
        { "name": "TopicId", "type": "uuid", "versions": "0+", "about": "The unique topic ID."},
        { "name": "TopicName", "type": "string", "versions": "0+", "mapKey": true, "entityType": "topicName",
          "about": "The topic name." },
        { "name": "MirrorName", "type": "string", "versions": "0+", "nullableVersions": "0+",
          "about": "The mirror name."}
      ]}
  ]
}

AddTopicsToMirrorResponse

{
  "apiKey":TBD,
  "type": "response",
  "name": "AddTopicsToMirrorResponse",
  "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": "TopicId", "type": "uuid", "versions": "0+", "about": "The unique topic ID."},
    { "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+", "ignorable": true,
      "about": "The error message, or null if there was no error." }
  ]
}

RemoveTopicsFromMirror

The RemoveTopicsFromMirror API allows users to detach topics from their associated mirror. The broker validates that all target topic partitions are in either PREPARING or MIRRORING state. Once validated, the request is forwarded to the controller, which appends the .removed suffix to the mirror.name topic config to mark the topics as no longer mirrored.

RemoveTopicsFromMirrorRequest

{
  "apiKey": TBD,
  "type": "request",
  "listeners": ["broker", "controller"],
  "name": "RemoveTopicsFromMirrorRequest",
  "validVersions": "0",
  "flexibleVersions": "0+",
  "fields": [
    { "name": "MirrorName", "type": "string", "versions": "0+", "ignorable": true,
      "about": "The cluster mirror name." },
    { "name": "Topics", "type": "[]TopicState", "versions": "0+", "about": "The topic state.",
      "fields": [
        { "name": "TopicId", "type": "uuid", "versions": "0+", "about": "The unique topic ID."},
        { "name": "TopicName", "type": "string", "versions": "0+", "mapKey": true, "entityType": "topicName",
          "about": "The topic name." }
      ]}
  ]
}

RemoveTopicsFromMirrorResponse

{
  "apiKey": TBD,
  "type": "response",
  "name": "RemoveTopicsFromMirrorResponse",
  "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": "TopicId", "type": "uuid", "versions": "0+", "about": "The unique topic ID."},
    { "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+", "ignorable": true,
      "about": "The error message, or null if there was no error." }
  ]
}

LastMirroredOffset

The LastMirroredOffset API allows destination cluster partition leaders in PREPARING state to query the last mirrored offset from the source cluster. If the source cluster has no record of this offset in its internal topic, it returns 0, meaning the log must be truncated to the beginning and mirroring starts from scratch. This is particularly important during failback. The last mirrored offset identifies where mirrored data ends and un-mirrored data begins. Records beyond this offset must be truncated before mirroring new data from the new source cluster; otherwise, the two clusters would contain inconsistent data.

ListMirroredOffsetsRequest

{
  "apiKey": TBD,
  "type": "request",
  "listeners": ["broker", "controller"],
  "name": "LastMirroredOffsetsRequest",
  "validVersions": "0",
  "flexibleVersions": "0+",
  "fields": [
    { "name": "MirrorName", "type": "string", "versions": "0+", "about": "The mirror name." },
    { "name": "Topics", "type": "[]TopicState", "versions": "0",
      "about": "The responses per topic.", "fields": [
      { "name": "Name", "type": "string", "versions": "0", "entityType": "topicName",
        "about": "The topic name." },
      { "name": "Partitions", "type": "[]PartitionState", "versions": "0",
        "about": "The responses per partition.", "fields": [
        { "name": "PartitionIndex", "type": "int32", "versions": "0",
          "about": "The partition index." }
      ]}
    ]}
  ]
}

LastMirroredOffsetsResponse

{
  "apiKey": TBD,
  "type": "response",
  "name": "LastMirroredOffsetsResponse",
  "validVersions": "0",
  "flexibleVersions": "0+",
  "fields": [
    { "name": "ThrottleTimeMs", "type": "int32", "versions": "0+",
      "about": "The duration in milliseconds for which the request was throttled due to a quota violation, or zero if the request did not violate any quota." },
    { "name": "ErrorCode", "type": "int16", "versions": "0+",
      "about": "The error code, or 0 if there was no error." },
    { "name": "Topics", "type": "[]OffsetResponseTopic", "versions": "0",
      "about": "The responses per topic.", "fields": [
      { "name": "Name", "type": "string", "versions": "0", "entityType": "topicName",
        "about": "The topic name." },
      { "name": "Partitions", "type": "[]OffsetResponsePartition", "versions": "0",
        "about": "The responses per partition.", "fields": [
        { "name": "PartitionIndex", "type": "int32", "versions": "0",
          "about": "The partition index." },
        { "name": "LastMirroredOffset", "type": "int64", "versions": "0",
          "about": "The last mirrored record offset." },
        { "name": "ErrorCode", "type": "int16", "versions": "0",
          "about": "The error code, or 0 if there was no error." }
      ]}
    ]}
  ]
}

ListMirrors

The ListMirrors API returns the current mirror names and their associated topic counts in the cluster.

ListMirrorsRequest

{
  "apiKey": TBD,
  "type": "request",
  "listeners": ["broker"],
  "name": "ListMirrorsRequest",
  // Version 0 is the initial version.
  "validVersions": "0",
  "flexibleVersions": "0+",
  "fields": []
}

ListMirrorsResponse

{
  "apiKey": TBD,
  "type": "response",
  "name": "ListMirrorsResponse",
  // Version 0 is the initial version.
  "validVersions": "0",
  "flexibleVersions": "0+",
  "fields": [
    { "name": "ThrottleTimeMs", "type": "int32", "versions": "0+", "ignorable": true,
      "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": "Mirrors", "type": "[]ListedMirror", "versions": "0+",
      "about": "Each mirror in the response.", "fields": [
      { "name": "MirrorName", "type": "string", "versions": "0+",
        "about": "The mirror name." },
      { "name": "SourceBootstrap", "type": "string", "versions": "0+",
        "about": "The source cluster bootstrap servers." },
      { "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

The DescribeMirrors API is to retrieve the information about the mirror names, including the partition state, lag, source offset and destination offset.

DescribeMirrorsRequest

{
  "apiKey": TBD,
  "type": "request",
  "listeners": ["broker"],
  "name": "DescribeMirrorsRequest",
  // Version 0 is the initial version.
  "validVersions": "0",
  "flexibleVersions": "0+",
  "fields": [
    { "name": "MirrorNames", "type": "[]string", "versions": "0+",
      "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." }
  ]
}

DescribeMirrorsResponse

{
  "apiKey": TBD,
  "type": "response",
  "name": "DescribeMirrorsResponse",
  // Version 0 is the initial version.
  "validVersions": "0",
  "flexibleVersions": "0+",
  "fields": [
    { "name": "ThrottleTimeMs", "type": "int32", "versions": "0+",
      "about": "The duration in milliseconds for which the request was throttled due to a quota violation, or zero if the request did not violate any quota." },
    { "name": "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+",
        "about": "The mirror name." },
      { "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+",
            "about": "The high watermark offset from the source cluster leader." },
          { "name": "DestinationOffset", "type": "int64", "versions": "0+",
            "about": "The log end offset on the destination cluster." },
          { "name": "Lag", "type": "int64", "versions": "0+",
            "about": "The lag (source offset - destination offset)." },
          { "name": "State", "type": "string", "versions": "0+",
            "about": "The partition state (INITIALIZING, PREPARING, MIRRORING, STOPPING, STOPPED, FAILED)." }
        ]}
      ]},
      { "name": "AuthorizedOperations", "type": "int32", "versions": "0+", "default": "-2147483648",
        "about": "32-bit bitfield to represent authorized operations for this mirror." }
    ]}
  ]
}

ReadMirrorStates

The ReadMirrorStates RPC reads mirror states from the coordinator broker when it resides on a different node than the requesting broker.

ReadMirrorStatesRequest

{
  "apiKey": TBD,
  "type": "request",
  "listeners": ["broker", "controller"],
  "name": "ReadMirrorStatesRequest",
  "validVersions": "0",
  "flexibleVersions": "0+",
  "fields": [
    { "name": "MirrorName", "type": "string", "versions": "0+", "about": "The mirror name." },
    { "name": "Topics", "type": "[]TopicState", "versions": "0",
      "about": "The responses per topic.", "fields": [
      { "name": "Name", "type": "string", "versions": "0", "entityType": "topicName",
        "about": "The topic name." },
      { "name": "Partitions", "type": "[]PartitionState", "versions": "0",
        "about": "The responses per partition.", "fields": [
        { "name": "PartitionIndex", "type": "int32", "versions": "0",
          "about": "The partition index." }
        ]}
      ]},
    { "name": "NeedPartitionStates", "type": "bool", "versions": "0+", "default": "true",
      "about": "Need the partition states or only topics states needed." }
  ]
}

ReadMirrorStatesResponse

{
  "apiKey": TBD,
  "type": "response",
  "name": "ReadMirrorStatesResponse",
  "validVersions": "0",
  "flexibleVersions": "0+",
  "fields": [
    { "name": "ThrottleTimeMs", "type": "int32", "versions": "0+",
      "about": "The duration in milliseconds for which the request was throttled due to a quota violation, or zero if the request did not violate any quota." },
    { "name": "ErrorCode", "type": "int16", "versions": "0+",
      "about": "The error code, or 0 if there was no error." },
    { "name": "Topics", "type": "[]TopicState", "versions": "0",
      "about": "The responses per topic.", "fields": [
      { "name": "Name", "type": "string", "versions": "0", "entityType": "topicName",
        "about": "The topic name." },
      { "name": "Partitions", "type": "[]PartitionState", "versions": "0",
        "about": "The responses per partition.", "fields": [
        { "name": "PartitionIndex", "type": "int32", "versions": "0",
          "about": "The partition index." },
        { "name": "LastMirroredOffset", "type": "int64", "versions": "0",
          "about": "The last mirrored record offset." },
        { "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." }
      ]}
    ]}
  ]
}

WriteMirrorStates

The WriteMirrorStates RPC writes mirror state updates to the coordinator broker when it resides on a different node than the requesting broker.

WriteMirrorStatesRequest

{
  "apiKey": TBD,
  "type": "request",
  "listeners": ["broker", "controller"],
  "name": "WriteMirrorStatesRequest",
  "validVersions": "0",
  "flexibleVersions": "0+",
  "fields": [
    { "name": "MirrorName", "type": "string", "versions": "0+", "about": "The mirror name." },
    { "name": "TopicsUpdated", "type": "[]TopicState", "versions": "0",
      "about": "The topics to be updated.", "fields": [
      { "name": "Name", "type": "string", "versions": "0", "entityType": "topicName",
        "about": "The topic name." },
      { "name": "Partitions", "type": "[]PartitionState", "versions": "0",
        "about": "The responses per partition.", "fields": [
        { "name": "PartitionIndex", "type": "int32", "versions": "0",
          "about": "The partition index." },
        { "name": "LastMirroredOffset", "type": "int64", "versions": "0",
          "about": "The last mirrored record offset." },
        { "name": "state", "type": "int8", "versions": "0+",
          "about": "The mirror partition state." }
      ]}
    ]},
    { "name": "RemovedTopics", "type": "[]string", "versions": "0+", "about": "The topic names to be removed." }
  ]
}

WriteMirrorStatesResponse

{
  "apiKey": TBD,
  "type": "response",
  "name": "WriteMirrorStatesResponse",
  "validVersions": "0",
  "flexibleVersions": "0+",
  "fields": [
    { "name": "ThrottleTimeMs", "type": "int32", "versions": "0+",
      "about": "The duration in milliseconds for which the request was throttled due to a quota violation, or zero if the request did not violate any quota." },
    { "name": "ErrorCode", "type": "int16", "versions": "0+",
      "about": "The error code, or 0 if there was no error." },
    { "name": "Topics", "type": "[]TopicState", "versions": "0",
      "about": "The responses per topic.", "fields": [
      { "name": "Name", "type": "string", "versions": "0", "entityType": "topicName",
        "about": "The topic name." },
      { "name": "Partitions", "type": "[]PartitionState", "versions": "0",
        "about": "The responses per partition.", "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." }
      ]}
    ]}
  ]
}

FindCoordinatorRequest

The FindCoordinatorRequest object is extended to support a new coordinator type:

public enum CoordinatorType { 
    GROUP((byte) 0), 
    TRANSACTION((byte) 1), 
    MIRROR((byte) 2); // New type
}

Mirror Metadata Records

LastMirroredOffsets

LastMirroredOffsets record tracks the latest successfully mirrored offset for each partition.

{
  "apiKey": 1,
  "type": "coordinator-key",
  "name": "LastMirroredOffsetsKey",
  "validVersions": "0",
  "flexibleVersions": "none",
  "fields": [
    { "name": "MirrorName", "type": "string", "versions": "0",
      "about": "The cluster mirror name."}
  ]
}

{
  "apiKey": 1,
  "type": "coordinator-value",
  "name": "LastMirroredOffsetsValue",
  "validVersions": "0",
  "flexibleVersions": "0+",
  "fields": [
    { "name": "Topics", "type": "[]Topic", "versions": "0+",
      "about": "The mirror topics for which we want to store the last mirrored offsets.",  "fields": [
      { "name": "Name", "type": "string", "versions": "0",
        "about": "The topic name." },
      { "name": "Partitions", "type": "[]Partition", "versions": "0+",
        "about": "Each partition to record the last mirrored offsets.", "fields": [
        { "name": "PartitionIndex", "type": "int32", "versions": "0+",
          "about": "The partition index." },
        { "name": "LastMirroredOffset", "type": "int64", "versions": "0+",
          "about": "The last mirrored offset for this partition." }
      ]}
    ]}
  ]
}

MirrorPartitionState

MirrorPartitionState record represents the lifecycle states of a mirrored partition.

{
  "apiKey": 2,
  "type": "coordinator-key",
  "name": "MirrorPartitionStateKey",
  "validVersions": "0",
  "flexibleVersions": "none",
  "fields": [
    { "name": "MirrorName", "type": "string", "versions": "0",
      "about": "The cluster mirror name."}
  ]
}

{
  "apiKey": 2,
  "type": "coordinator-value",
  "name": "MirrorPartitionStateValue",
  "validVersions": "0",
  "flexibleVersions": "0+",
  "fields": [
    { "name": "TopicName", "type": "string", "versions": "0",
      "about": "The topic name."},
    { "name": "Partition", "type": "int32", "versions": "0",
      "about": "The partition index."},
    { "name": "State", "type": "int8", "versions": "0+",
      "about": "The mirror partition state." }
  ]
}

Configuration

A new configuration resource type is added for cluster mirrors, which is stored in the cluster metadata internal log:

...

Cluster mirrors can be configured using the following properties:

Topic Configuration

Broker Configuration

Mirror Configuration

Metrics

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

Compatibility, Deprecation, and Migration Plan

Cluster Mirroring will be introduced through a phased rollout across multiple Kafka releases to ensure stability and gather community feedback. 

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 and unstable feature versions in all configuration files. After starting the cluster with a minimum metadata version, administrators can dynamically enable the mirror version feature to activate Cluster Mirroring. 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 administrators 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.

Migration From MirrorMaker 2

Cluster Mirror is not compatible with MirrorMaker 2 (MM2). This is a critical consideration for users planning to migrate from MirrorMaker 2 to Cluster Mirroring.

...

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

Compatibility Matrix

Note that some features require support from the source cluster.

Performance

MirrorFetcherThread uses the same fetch protocol optimizations as ReplicaFetcherThread:

...

• 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, administrators 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.

Future Work

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

...

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

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-mirrors.sh, add topics, verify replication
• Basic Replication: Create mirror via API, replicate topic, verify data consistency
• 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:

• Performance Benchmark: Measure replication throughput and latency across WAN.
• Scalability Test: Replicate 1000 topics with 100,000 partitions across clusters.
• Failover Test: Simulate source cluster failure, measure consumer recovery time.
• Long-Running Stability: Run continuous replication for 7 days, verify no memory leaks or performance degradation.
• Security Validation: Test all authentication mechanisms (SASL PLAIN, SCRAM, Kerberos, mTLS) via kafka-mirrors.sh config files.

Rejected Alternatives

Keep Using MirrorMaker 2

This KIP introduces native cluster mirroring to address the limitations of MirrorMaker 2 described in the motivation section.

Support Unclean Leader Election

As described in the non-goal section, since there's no shared leader epoch between source and destination cluster, supporting unclean leader election becomes very tricky.

...