DUE TO SPAM, SIGN-UP IS DISABLED. Goto Selfserve wiki signup and request an account.
...
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.
...
Group Offset
Cluster Mirroring synchronizes consumer group offsets from the source cluster to the destination cluster, enabling consumers to resume consumption from their last committed offset position after failover. The MirrorMetadataManager periodically fetches consumer group committed offsets Both traditional consumer groups and share consumer groups (Kafka Queue functionality) are supported, with offset synchronization running in two separate phases to prevent cross-type conflicts.
For consumer groups, committed offsets are periodically fetched from the source cluster and replicates it applied 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 Control
Cluster Mirroring supports comprehensive security controls through both authorization and authentication mechanisms. This ensures that only authorized principals can establish and manage cluster mirrors. When configuring a mirror, operators specify 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.
. For share groups, which use a different offset management model based on Share-Partition Start Offset (SPSO) and Share-Partition End Offset (SPEO), the current SPSO is retrieved from the source and applied to the destination, which also initializes the group state in both the group coordinator and share coordinator. This means a share group can be initialized in the destination cluster even if it doesn't exist yet, eliminating the need for pre-creation or complex state management.
Each phase lists only groups of its own type on both source and destination, preventing a consumer group on the source from overwriting a share group with the same name on the destination (or vice versa). Kafka enforces that consumer group and share group names must be unique within a single cluster. When a name conflict does occur across types, the offset commit operation will fail for the affected group, which is logged and skipped without affecting other groups. Users can resolve these conflicts by deleting the conflicting group in the destination cluster or excluding it from offset synchronization. These conflicts affect only offset synchronization and do not impact data mirroring itself.
During offset synchronization, the committed offset in the destination cluster may temporarily exceed the current log end offset (LEO) of the mirror topic. For example, if a consumer commits offset 100 in the source cluster but the destination cluster has only mirrored up to offset 80, the offset 100 is still committed to the destination. This is acceptable because the mirror leader continues fetching data and the LEO will eventually catch up. However, if a failover occurs before the mirrored data catches up, consumers attempting to resume from offset 100 will receive an OffsetOutOfRangeException. To handle this gracefully, consumers should configure auto.offset.reset=latest when consuming from mirror topics, ensuring that if a committed offset is beyond the current LEO after failover, the consumer automatically resets to the latest available offset.
Security Control
Cluster Mirroring supports comprehensive security controls through both authorization and authentication mechanisms. This ensures that only authorized principals can establish and manage cluster mirrors. When configuring a mirror, operators specify 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 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.
...
Transaction 4001 was committed at the source but aborted at the destination because the COMMIT marker (offset 3) had not yet been replicated. Transaction 4002 was correctly aborted at both clusters. Applications that require strict transactional guarantees across clusters should implement deduplication or reconciliation logic after failover. Additionally, the kafka-transactions tool can only abort transactions originated from the local cluster. It cannot abort transactions replicated via mirroring because the __transaction_state topic is not mirrored. Hanging transactions from mirrored data are resolved exclusively by the STOPPING transition flow described above.
Bandwidth Control
Cluster Mirroring adopts a dual-sided throttling mechanism that extends Kafka's existing bandwidth control capabilities to work across cluster boundaries.
- 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.
- Source Cluster Throttling: The source cluster side requires a different approach because mirror fetch requests operate as consumer traffic rather than replication traffic. Consequently, standard leader replication throttling mechanisms cannot apply to mirror traffic. Instead, the source cluster leverages Kafka's client quota system. Each mirror fetcher thread presents itself with a deterministic client identifier that encodes the broker ID, fetcher thread number, and mirror name. Operators can apply per-client byte rate quotas to these identifiers, effectively throttling the outbound mirror traffic from the source cluster. This approach integrates seamlessly with Kafka's existing quota enforcement infrastructure.
In a follow-up KIP we will add source-side throttling that allows source cluster leaders to limit bandwidth served to all mirror fetchers, similar to how leader.replication.throttled.rate controls intra-cluster replication. This provides independent control over mirror catch-up traffic without impacting local replication or consumer workloads. Combined with destination-side throttling, operators gain complete bidirectional bandwidth control for mirror traffic.
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 Groups
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.
via mirroring because the __transaction_state topic is not mirrored. Hanging transactions from mirrored data are resolved exclusively by the STOPPING transition flow described above.
Bandwidth Control
Cluster Mirroring adopts a dual-sided throttling mechanism that extends Kafka's existing bandwidth control capabilities to work across cluster boundaries.
- 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.
- Source Cluster Throttling: The source cluster side requires a different approach because mirror fetch requests operate as consumer traffic rather than replication traffic. Consequently, standard leader replication throttling mechanisms cannot apply to mirror traffic. Instead, the source cluster leverages Kafka's client quota system. Each mirror fetcher thread presents itself with a deterministic client identifier that encodes the broker ID, fetcher thread number, and mirror name. Operators can apply per-client byte rate quotas to these identifiers, effectively throttling the outbound mirror traffic from the source cluster. This approach integrates seamlessly with Kafka's existing quota enforcement infrastructure.
In a follow-up KIP we will add source-side throttling that allows source cluster leaders to limit bandwidth served to all mirror fetchers, similar to how leader.replication.throttled.rate controls intra-cluster replication. This provides independent control over mirror catch-up traffic without impacting local replication or consumer workloads. Combined with destination-side throttling, operators gain complete bidirectional bandwidth control for mirror traffic.
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 KIPKafka 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.
...