• Created by Unknown User (showuon), last updated on Feb 13, 2026 17 minute read

You are viewing an old version of this page. View the current version.

Compare with Current View Page History

« Previous Version 2 Next »

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

This page is meant as a template for writing a KIP. To create a KIP choose Tools->Copy on this page and modify with your content and replace the heading with the next KIP number and a description of your issue. Replace anything in italics with your own description.

Status

Current state:"Under Discussion"

Discussion thread: here [Change the link from the KIP proposal email archive to your own email thread]

JIRA: here [Change the link from KAFKA-1 to your own ticket]

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.

Figure 1: Cluster Mirroring Setup.

  • Integrated Architecture: Replication logic runs within broker processes, eliminating external dependencies and reducing the operational footprint.
  • Simplified Configuration: Creating a cluster mirror requires a single command-line invocation or Admin API call with bootstrap servers and security credentials.
  • Metadata Synchronization: Topic configurations, consumer group offsets, and ACLs are periodically synchronized from source to destination cluster without additional configuration.
  • Unified Monitoring: Mirroring metrics are exposed through standard Kafka broker JMX metrics alongside existing replication metrics. Administrators use familiar tools and dashboards to monitor cross-cluster replication.
  • Faster Failover: The failover operation is simplified because metadata synchronization is continuous and automatic. Consumer applications can resume processing immediately after switching clusters without any offset translation.
  • Delta Failback: Destination leader acts as a follower with regards to source leader, so it will always fetch from the local log end offset to catch up with the leader, making it possible to mirror only the delta when failing back (reverse mirroring).
  • Version Compatibility: For migration or DR use cases where only failover is needed without failback, this proposal supports any Kafka version from v2.1 onward as the source cluster, leveraging the client/broker forward compatibility introduced in v4.0.

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.

Producers write to the source cluster and receive acknowledgments based on the source cluster's replication requirements (e.g. acks=all ensures replication to all in-sync replicas within the source cluster). Data is then asynchronously replicated to destination clusters with no impact on producer latency or throughput.

This decision reflects the reality that cross-datacenter network latency makes synchronous replication impractical for some deployments. Requiring synchronous acknowledgment from a geographically distant cluster would introduce significant latency (typically 50-200ms for inter-region replication), making it unsuitable for latency-sensitive applications.

Implications for DR use cases:

  • Potential Data Loss: In the event of a catastrophic failure of the source cluster, recently produced records that have not yet been replicated to the destination cluster will be lost. The amount of data loss depends on replication lag at the time of failure.
  • RPO (Recovery Point Objective): Organizations must handle a non-zero RPO determined by the replication lag between source and destination clusters. Typical replication lag ranges from seconds to minutes depending on network bandwidth, throughput, and geographic distance.

 

Asynchronous replication should provide the right balance for disaster recovery use cases where availability and performance of the primary cluster must not be compromised by cross-datacenter latency. Applications requiring zero data loss across cluster failures can wait for the follow-up KIP that will extend this design to support synchronous mirroring, or handle the lag using application-level caching.

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 Elections

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.

In normal Kafka operation, once a record is committed (part of the high watermark), it is immutable and will never be changed or removed. When a new leader is elected, followers use the epoch information to determine which records are safe to keep and which must be truncated to align with the new leader's log. Replicas eventually converge to the same data through epoch-based reconciliation. Unclean leader elections break this guarantee by allowing non-ISR brokers to become leaders, potentially with fewer records than were previously committed.

Source and destination clusters have completely independent controller architectures. Leadership changes in the source cluster happen independently of destination leadership changes. This means that epoch values diverge between clusters even though they represent the same logical topic partition. Source cluster epoch N and destination cluster epoch N have no inherent relationship, they represent different leadership events that happened at different times. This means that standard epoch comparison is insufficient because epochs are meaningful only within their originating cluster.

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.


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.

Figure 2: High Level Architecture.

The mirror name is stored as a topic-level configuration (mirror.name) that propagates through Kafka's metadata log as configuration change records. When topics are added to a mirror via the addTopicsToMirror API, the controller generates configuration records that are replicated to all brokers through the standard metadata update mechanism.

Brokers monitor these configuration changes to detect when partitions they lead belong to a mirror, triggering the creation of mirror fetchers and enforcement of read-only semantics. This design ensures that mirror associations are visible, auditable, and manageable through standard Kafka 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.

We use a composite key of mirror name, topic id, and partition number to distribute coordination work across the __mirror_state topic's partitions, which is the internal compacted topic used to store mirror metadata. Each mirror partition independently hashes to a coordinator, spreading the load across all brokers in the cluster. This means a mirror with hundreds of partitions will have its state management distributed evenly rather than concentrated on a single broker.

Responsibilities:

  • State Management: Mirror configuration and partition states are stored in the internal topic. The coordinator loads the state on startup and partition leadership changes.
  • Partition Assignment: Cluster mirrors are assigned to coordinator partitions using consistent hashing based on the mirror name. This distributes coordinator load across all brokers and allows for horizontal scaling. The number of coordinator partitions is configurable via mirror.topic.num.partitions.
  • Leader Election: When a broker becomes the leader for a __mirror_state partition, it loads the mirror metadata for all mirrors assigned to that partition and begins coordinating those mirrors. On resignation, it clears its in-memory state to avoid stale metadata.
  • Metadata Refresh Scheduling: The coordinator schedules periodic metadata refresh operations by invoking a metadata manager every 30 seconds by default. This ensures that topology changes, configuration updates, and offset commits in the source cluster are continuously propagated to the destination cluster.
  • State Transitions: The coordinator manages asynchronous state transitions for mirror partitions. Each partition is an independent replication unit with its own state. When the coordinator is the leader for a mirror partition, it writes the state updates directly to the internal topic. Remote brokers read and write partition state via new RPCs, enabling distributed coordination across the cluster. Both local and remote state updates trigger callbacks to execute appropriate actions for each state.

Figure 3: Mirror Partition Lifecycle.

States descriptions:

  • UNKNOWN: The partition has no cached state (broker just became leader, state not loaded yet). Not an explicit API-driven state, just the absence of state.
  • PREPARING: The coordinator for this partition detects via onMetadataUpdate that it leads a mirror partition. It fetches last mirrored offsets from the source cluster and schedules truncation to align the local log with the source. Valid from: null, UNKNOWN, STOPPED, FAILED.
  • MIRRORING: All ISR members have completed truncation. A MirrorFetcherThread is started to continuously replicate records from the source cluster. Valid from: PREPARING only.
  • STOPPING: Triggered by RemoveTopicsFromMirror API (user wants to fail over) or topic deletion on the source. The system records the last mirrored offset to the internal topic. Valid from: PREPARING, MIRRORING.
  • STOPPED: Last mirrored offsets have been persisted. The topic becomes writable on the destination cluster (the mirror fetcher is removed and the read-only flag is cleared). Valid from: STOPPING only.
  • FAILED: An error occurred. Can be entered from any state. Can transition back to PREPARING to retry. Valid from any state.


Examples:

Starting a mirror (UNKNOWN -> PREPARING -> MIRRORING): The addTopicsToMirror command sets mirror.name config via the controller. The metadata update propagates to brokers. The broker leading the partition sees it's the coordinator, finds no cached state (UNKNOWN), and transitions to PREPARING. After truncation completes, it moves to MIRRORING.

Failover (MIRRORING -> STOPPING -> STOPPED): The removeTopicsFromMirror command clears mirror.name. The coordinator detects the stop request, transitions to STOPPING, persists the last offset, then moves to STOPPED. The topic is now writable.

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.

Responsibilities:

  1. Connection Management: The manager maintains a connection pool with one blocking sender per source cluster. These connections are created lazily when the first topic for a mirror is added. Each sender uses the security credentials and network settings from the mirror configuration, allowing different mirrors to use different authentication mechanisms.
  2. Topic Metadata Synchronization: Every refresh cycle, the manager fetches topic metadata from source clusters using standard MetadataRequest calls. For each topic in the mirror configuration:
    1. Topic Creation: If a topic exists in the source but not the destination, the manager sends a CreateTopics request to the controller with identical partition count and configurations.
    2. Partition Expansion: If the source topic has more partitions than the destination, the manager sends a CreatePartitions request to scale up the destination topic to match.
    3. Configuration Sync: Topic configurations are compared between source and destination. Any differences trigger an IncrementalAlterConfigs request to align destination configs with the source.
  3. Consumer Group Offset Synchronization: The manager synchronizes classic and share consumer group offsets to enable seamless failover (no offset translation):
    1. Lists all consumer groups using ListGroups request.
    2. Fetches committed offsets for each group using OffsetFetch request or DescribeShareGroupOffsets request.
    3. Commits those offsets to the destination cluster’s group coordinator using the internal OffsetCommit or AlterShareGroupOffsets request.
  4. ACL Synchronization: Access control lists are mirrored from source to destination to maintain consistent security policies:
    1. Fetches all ACLs from the source using DescribeAcls request.
    2. Compares with the destination cluster’s current ACLs from the metadata image.
    3. Creates missing ACLs using CreateAcls request.
    4. Deletes ACLs that exist in destination but not in source using DeleteAcls request.


Cluster Mirroring allows users to modify configurations in the destination cluster, though these changes are periodically overridden by the topic configuration synchronization cycle. This design choice was made because while dynamic configuration changes could be blocked, static configuration changes via properties files cannot be prevented, making override inevitable. 

However, this approach presents challenges in environments with external governing systems like the Strimzi operator, where the continuous reconciliation process conflicts with the refresh cycle, potentially causing performance impacts. More critically, temporary configuration mismatches such as reduced retention periods or altered partition counts could lead to data loss or missing partitions until the next synchronization cycle detects and corrects the discrepancy, highlighting the need for careful operational awareness when mixing mirroring with external cluster management solutions.

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

Each mirror can define its own filtering rules independently, loaded from the manager at each refresh cycle:


  • The mirror.groups.include config controls which consumer group offsets are synced using regex patterns.
    • .* (all groups by default)
    • app-.* (only sync groups starting with app-)
    • app-.*,service-.* (sync groups starting with app- or service-)


  • 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, mirror name) to organize threads, ensuring that:

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


The MirrorFetcherThread (MFT) is a specialized implementation of AbstractFetcherThread that handles cross-cluster data replication with consumer Fetch requests and different epoch semantics than standard intra-cluster replication, but keeping the same log consistency validations.

In Cluster Mirroring, destination partition leaders operate in a dual-role. They act as followers when fetching committed data (up to the LSO) from the source cluster leader, while simultaneously serving as leaders for their local replicas in the destination cluster. To maintain data consistency, destination partitions are read-only and reject produce requests from clients with ReadOnlyTopicException.

A mirror leader partition begins with an unknown source leader epoch. When it sends Fetch requests to the source cluster, the source leader may respond with a FencedLeaderEpochException. When such an error occurs, the mirror fetcher extracts the current source leader epoch from the error response and updates its internal fetch state to track the source cluster's actual leader epoch.  The last fetched epoch is always set to empty to disable log divergence checks due to unclean leader election (see non-goals section).

Figure 4: Mirror Leader Fetch State.

On subsequent Fetch requests:

  1. Fetch validation: The tracked source epoch ensures that fetched batches from the source cluster are validated against the correct source leader epoch, preventing acceptance of stale or invalid data (see KAFKA-18723).
  2. Epoch rewriting: When records are appended to the destination log, the batch epochs are rewritten to match the destination cluster's leader epochs, maintaining consistency within the destination cluster.


The source epoch tracking is purely for fetch validation, while the destination uses its own independent epoch sequence for replication and durability. This design keeps the two clusters' epoch spaces completely separate, allowing the destination to operate as a normal Kafka cluster with standard intra-cluster replication.

When the source partition's leader changes, a NotLeaderOrFollowerException is returned. At this point, the mirror fetcher thread queries the MirrorMetadataManager to get the updated endpoint and either creates a new fetcher thread or reuses one that is already connected to the new endpoint. This allows mirroring to continue seamlessly despite leadership changes in the source cluster.

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.

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.

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.

For each partition, we track the high watermark (HW) by storing it in the cluster metadata as Last Mirrored Offset (LMO) when removing a topic from a mirror (failover phase). The LMO represents the last record successfully mirrored from the original source cluster to the destination cluster before failover.

When reverse mirroring 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 LastMirrorredOffsets request to the new source cluster asking for the latest mirrored offset, and then truncates its local log to the returned offset. 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 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.

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

Public Interfaces

Briefly list any new interfaces that will be introduced as part of this proposal or any existing interfaces that will be removed or changed. The purpose of this section is to concisely call out the public contract that will come along with this feature.

A public interface is any change to the following:

  • Binary log format

  • The network protocol and api behavior

  • Any class in the public packages under clientsConfiguration, especially client configuration

    • org/apache/kafka/common/serialization

    • org/apache/kafka/common

    • org/apache/kafka/common/errors

    • org/apache/kafka/clients/producer

    • org/apache/kafka/clients/consumer (eventually, once stable)

  • Monitoring

  • Command line tools and arguments

  • Anything else that will likely break existing users in some way when they upgrade

Proposed Changes

Describe the new thing you want to do in appropriate detail. This may be fairly extensive and have large subsections of its own. Or it may be a few sentences. Use judgement based on the scope of the change.

Compatibility, Deprecation, and Migration Plan

  • What impact (if any) will there be on existing users?
  • If we are changing behavior how will we phase out the older behavior?
  • If we need special migration tools, describe them here.
  • When will we remove the existing behavior?

Test Plan

Describe in few sentences how the KIP will be tested. We are mostly interested in system tests (since unit-tests are specific to implementation details). How will we know that the implementation works as expected? How will we know nothing broke?

Rejected Alternatives

If there are alternative ways of accomplishing the same thing, what were they? The purpose of this section is to motivate why the design is the way it is and not some other way.

  • No labels