Authors: Omnia Ibrahim, Gaurav Narula, Luke Chen, Federico Valeri 

Status

Current state: DRAFT

Discussion thread: here [#TODO]

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

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

Motivation

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

Asynchronous mirroring in KIP-1279: Cluster Mirroring carries a data-loss risk – as replication lag grows, records acknowledged on the source may not reach the destination before a failure. Today this is the application's problem to solve, typically through caching layers that replay data — adding application complexity and no single solution that fits all use cases.

Stretched clusters are sometimes proposed as an alternative, but they don't protect against software failures or configuration incidents, so they're not a real DR option. Most vendors position stretched clusters as HA rather than DR, and few offer them as a managed service — the operational cost is high and the DR coverage is limited.

This KIP introduces a synchronous mode of cluster mirroring that delays acknowledging a producer until the record has been replicated to all ISRs in every registered synchronous mirroring link. The mechanism extends the existing ISR-based ack predicate to also account for ISRs on registered destination clusters

End-to-End operator workflow enabled by Sync mirroring:

This section catalogs the end-to-end operator workflows that synchronous mirroring enables.

Disaster Recovery (zero-RPO)

Synchronous mirroring's primary use case is disaster recovery with zero recovery-point objective. When a topic is configured for sync mirroring, the source's high watermark advances only after every registered destination's ISR has caught up to the produced offset — every record acknowledged on the source is therefore durable on the destination at acknowledge time. On source failure (planned or unplanned), the destination holds the full acknowledge record appended to log file up to the last acknowledge offset, so operators can redirect producers and consumers to the destination as the new primary without losing any acknowledged record. The async data-loss window described in Motivation — where lag-tail records can be permanently lost on failure — is closed for any topic in sync mode.

Near Zero-Downtime Cluster Migration

Beyond disaster recovery, synchronous mirroring also enables a near zero-downtime cluster migrations. Traditional migration workflows require operators to stop all producers, wait for the asynchronous replication lag to drain to zero, and only then redirect producers to the new cluster. This "stop-drain-switch" approach introduces unavoidable downtime proportional to the replication lag at the time producers are stopped. With synchronous mirroring, the workflow becomes fundamentally different: operators first enable mirroring in async mode, let it catch up, then promote to sync mode (or enable sync from start which start as asynchronous then auto-promotion to sync triggers when lag reaches zero). Once sync mode is active, every acknowledged record is guaranteed to exist on the destination cluster. Producers can then be redirected at any time — there is no lag to drain and no window of data loss, eliminating the need for a coordinated shutdown.

Kafka Streams and Kafka Connect DR

Asynchronous mirroring is not a viable disaster-recovery story for Kafka Streams or Kafka Connect because both rely on internal Kafka topics — Streams changelog topics for state stores, Connect offset and config topics — whose contents are not derivable from any other source. In async mirroring any record acknowledged on the source but not yet replicated when a failure occurs is permanently lost on the destination, and a single missing record in a state-changelog or config topic corrupts the application's persisted state. There is no application-level workaround; the state is simply wrong on the destination, and the application cannot resume cleanly.

With sync mode, operators can configure these internal topics for sync mirroring along side sync mirroring of any input topics to the flow. Any record acknowledged is durable on the destination, so the application's state on failover matches what the source had. The corruption mode goes away.

One residual gap remains: the consumer group coordinator (which holds offset commits for Connect sink connectors and Streams' internal consumer state) is async-mirrored under KIP-1279 and this KIP — sync coordinators are out of scope (see Non-Goals). On failover, offset commits may lag behind the actual processed state, leading to bounded reprocessing of records that had already been processed but whose offsets hadn't been committed cross-cluster yet. This is a much smaller blast radius than corrupted application state — bounded reprocessing is a known and recoverable cost; corrupted state is not.

Operator mitigation: transfer source-committed offsets on planned failover. Because the input topics and internal state topics are sync-replicated, the destination has every record the source did up through any sync-acknowledged offset. The remaining gap is purely on the consumer-group side: the destination's __consumer_offsets reflects whatever has been async-replicated for consumer group state, which can lag the actual processing state on source. When the source is still reachable at failover time (planned cluster migration, coordinated failover, or post-incident rebuild) or have access to last committed offsets from metrics, an operator can transfer the source's last-committed input-topic offsets directly to the destination's consumer group for the Streams application or the Connect sink — for example by describe the source's last committed offset for the relevant group and using kafka-consumer-groups.sh --reset-offsets --to-offset <X> against the destination cluster. This eliminates the reprocessing window that async-replicated offsets would otherwise produce. In a true source-dead DR scenario where the source is unreachable, the operator is limited to metrics they have on application side or any 3rd-party services that report offsets to relay on or reset the offset to latest if applicable. 

Sync mode therefore moves Streams/Connect DR from "not supportable" to "supportable with extra operations on failover." Full coverage of the consumer-group-coordinator gap is the same follow-up KIP that addresses cross-cluster transactional exactly-once — both need a cross-cluster-aware coordinator.

Feature Improvements with Sync Mode

Beyond closing the async-lag data-loss window, sync mode unlocks several capabilities that async mirroring fundamentally cannot deliver. Each is detailed below.

Log Compaction and Retention Safety

Asynchronous mirroring has a subtle data-loss risk with log compaction and retention. Because the source cluster's log cleaner operates on records below the high watermark, and async mirroring does not gate the high watermark on the destination's replication progress, records on the source may be compacted away or deleted by retention policies before the async mirror fetcher has replicated them to the destination. Those records are permanently lost — the destination never sees them. This is particularly dangerous for compacted topics (e.g., configuration topics, Kafka Streams state stores, changelogs) where the latest value for each key is critical. With sync mirroring, the source's high watermark only advances after every registered destination's ISR has caught up. Since compaction and retention only operate on records below HW, every record is guaranteed to have been replicated to the destination before it becomes eligible for cleanup on the source.

Transactional Producer

Synchronous mirroring meaningfully improves the transactional-producer failover story compared to async mode, though full cross-cluster exactly-once still requires a follow-up KIP (see Non-Goals).

In async mode, on failover the destination aborts any in-flight transaction whose COMMIT/ABORT marker hadn't replicated. The aborted set can be large — proportional to async lag — because data records may have replicated but their markers may not have. Both data records and transaction markers are log records that flow through the same replication path, so any lag-tail of data without its corresponding marker becomes an aborted transaction on failover.

With sync mirroring, both data records and transaction markers are sync-replicated when the topic is in SYNC mode (markers are themselves append records, so sync gating applies to them automatically). At steady state with zero lag, the in-flight-and-incomplete set is bounded to the single currently in-flight transaction (if any) — vs unbounded in async mode. On failover:

• Marker absent on destination at failover: the transaction is aborted as part of the KIP-1279 transaction failover path. Bounded to smaller set of in-flight transaction at the moment of source crash.
• Marker present on destination at failover: the destination's local HW advances to or past the marker offset on leader election, exactly as a single-cluster leader handoff would; the transaction is visible as committed.

Caveat — mixed SYNC/ASYNC topics in one transaction. A single transactional producer can span multiple topics. transaction markers are written per-partition. If a transaction spans topic A (SYNC) and topic B (ASYNC), A's marker is sync-gated and B's is not. On source crash before B's marker replicates, the same transaction can appear committed for A's data (marker present on destination) and aborted for B's data (marker absent). For transactional consistency across topics, either make all participating topics SYNC, or avoid spanning multiple topics in a single transaction. 

Caveat — sendOffsetsToTransaction is not sync-supportable. The transactional producer API Producer.sendOffsetsToTransaction(offsets, consumerGroupMetadata) commits consumer group offsets atomically with the transaction's data records. The offset commits are routed through the consumer group coordinator and persisted to __consumer_offsets. Because consumer group coordinator state is explicitly out of scope for sync mirroring under this KIP (see Non-Goals), the offset-commit half of such transactions is async-replicated even when the data topics are SYNC. On source failover, the transaction's data records can be durable on the destination while its corresponding offset commits lag — consumers reading the destination after failover may reprocess records whose offsets had been committed on the source but not yet replicated. This is mechanically the same gap described under Kafka Streams and KafkaConnect DR, where an operator-side mitigation is also documented. Full atomicity for sendOffsetsToTransaction across clusters requires sync coordinators, deferred to the same follow-up KIP that covers cross-cluster exactly-once.

Goals

• Support synchronous mirroring at the broker level, providing a more robust solution for zero-RPO use cases.
• Enable zero-downtime cluster migrations by allowing producers to switch to a destination cluster at any time once synchronous mirroring is active, without requiring a "stop producers, drain lag, then switch" workflow.

Non-Goals

Synchronous Transaction Coordinator: Full cross-cluster exactly-once — where a producer initiates a transaction on the source and resumes the same transaction on the destination after failover with the same PID, epoch, and sequence — requires a cross-cluster-aware transaction coordinator. Deferred to a follow-up KIP. For the partial improvements sync mode delivers today, see Feature Improvements > Transactional Producer.

Synchronous Consumer Group: Consumer groups will remain asynchronous for the time being for multiple reasons:

• There is limited value in making consumer groups synchronous. In the majority of zero-RPO cases, the main concern is the topic data itself on the produce side, not occasional double processing.
• The only case where synchronous consumer groups become relevant is in the context of transactional producers, where a producer can also update the committed offset of a given consumer group. Since the full support of transactional producers are not part of this KIP, this can be deferred to follow-up KIPs for full transactional support in the future.

Proposed Changes

Cluster Mirroring Mode

Asynchronous mirroring: This is the default mode introduced in KIP-1279. Records are acknowledged to the producer once they are acknowledged by the source cluster, then replicated to the destination asynchronously.

• Maintains existing latency for source producers, but does not guarantee zero-RPO.
• A topic can have multiple asynchronous mirror links registered. Each destination cluster replicates independently without contributing to producer acknowledgment latency.

Synchronous mirroring: The source cluster will delay acknowledging records to the producer until the destination cluster acknowledges the records first.

• Increases latency for source producers in exchange for zero-RPO.
• A topic can have multiple synchronous mirror links registered. The source delays acknowledgment until every registered destination has reached ISR for the record, so zero-RPO holds across all of them.
• Note: Producer latency is bounded by the slowest synchronous destination cluster. Latency scales with the number of registered sync destinations.
• Sync-mode produces can fail with cross-cluster-specific errors (NOT_ENOUGH_REMOTE_REPLICAS_AFTER_APPEND, MIRROR_SYNC_REPLICATION_TIMEOUT); see Errors.

New cluster mirroring is always initialized in asynchronous mode. Once it has caught up, it can be promoted to synchronous mode; once the sync metadata converges across both clusters, every new record produced on the source is delivered to the destination before being acknowledged.

Cluster mirroring in synchronous mode can be downgraded to asynchronous via --alter --async. The demotion takes effect as soon as the destination commits MirrorTopicTypeChangeRecord(ASYNC) — source-side HW gating is dropped immediately. Any in-flight acks=all produces still waiting on sync gating then complete under the new ASYNC gating predicate, which drops the cross-cluster zero-RPO guarantee for those in-flight requests. Draining in-flight sync requests before flipping would defeat the purpose of demotion, since demotion is typically invoked when the destination is already slow.

Source clusters can also unilaterally release a synchronous mirror registration when the destination becomes unreachable. See Source-Side Release.

Registering Sync Replica

When a topic is added/updated to be in sync mode:

Destination Cluster:

1. MirrorCoordinator sets the desired mirror type for the topic to SYNC. The system auto-promotes to sync mode when lag reaches zero across all partitions of the topic — promotion is never rejected on a non-zero lag, just deferred. See ASYNC→SYNC Promotion for the unified path used by both --add --sync and --alter --sync.
2. MirrorCoordinator sends a RegisterMirrorSyncTopicRequest to the source cluster naming the topics being registered.

Source Cluster:

1. MirrorCoordinator on the source registers the topic for sync mode under the given mirror link name and emits MirrorSyncRegistrationChangeRecord(Registered=true) so source partition leaders see the new entry in the topic's registeredMirrors set on TopicImage. Per-replica LEOs, ISR membership, and MinISR for the destination flow from the MirrorReplicaInfo field of normal FetchRequests once the destination's fetcher restarts in SYNC mode.

Initial-Fetch latency window: Until the destination's first SYNC-mode Fetch arrives (bounded by the destination's mirror fetcher poll interval — typically sub-second), the source has no MirrorReplicaInfo for the new destination, so any acks=all produces in that window wait at the source HW gate. The wait is bounded by the fetch interval and well inside request.timeout.ms, so producers experience a brief latency bump rather than a failure. This applies equally to a fresh promotion and to mass re-registration after a controller failover (each destination's first Fetch is independent and arrives within one interval, so the window does not compound across topics).

Registration propagation window: When the source MirrorCoordinator handles RegisterMirrorSyncTopicRequest, it writes per-(mirror, topic) state to __mirror_state and asks the KRaft controller to commit MirrorSyncRegistrationChangeRecord to the quorum metadata log. Once a quorum of KRaft has the record, the coordinator returns OK to the destination. Source brokers replay the metadata log on their own cadence — the controller does not gate on broker application — so by the time the destination receives OK, some source partition leaders for the topic may still be using a stale topicImage.registeredMirrors. Two consequences for that residual window:

• Fetch-time: A SYNC-mode Fetch landing on a stale leader is rejected with MIRROR_REGISTRATION_PROPAGATING (distinct from INVALID_SYNC_REGISTRATION). The leader detects propagation lag using its own metadata-lag self-knowledge — if broker.appliedMetadataOffset < broker.latestKnownControllerCommitOffset and the mirror is unknown, return MIRROR_REGISTRATION_PROPAGATING; otherwise return INVALID_SYNC_REGISTRATION (broker is current, mirror genuinely not registered or was released). The destination treats both as per-partition signals — fetcher retries the affected partition in SYNC mode, the rest of the topic continues normally (see Source-Side Release for the full destination-side handling). Topic-wide demote only happens when every partition has been failing continuously for mirror.sync.topic.demote.timeout.ms (default 30s), or on a Register-time MIRROR_SYNC_RELEASED_REQUIRES_FRESH_PROMOTION.
• Producer-side: A producer acks=all request landing on a stale leader during the window will be acknowledged without sync gating, since the leader doesn't yet know to gate on this destination. This window is bounded by KRaft metadata propagation latency to non-coordinator brokers — the same limitation any KRaft-driven metadata change has. Operators should treat sync engagement as bounded-latency rather than instantaneous from Register OK; this matters more here than for other metadata changes because the lag has a durability consequence rather than just a behavioural one.

Unregistering (reverting to async):

A synchronous mirroring link can be unregistered in three ways:

1. Explicit demotion: The destination cluster can alter the mirrored topic back to async via AlterMirrorTopicTypeRequest. This triggers an UnregisterMirrorSyncTopicRequest to the source cluster. Coordinated through the destination, no special source-side state is left behind.
2. Topic removal from mirror: When RemoveTopicsFromMirror triggers STOPPING on a SYNC topic, sync-specific handling is needed because the source is gating producer acks on this destination:
   • Order: UnregisterMirrorSyncTopicRequest  is sent as the first STOPPING  step (before fetcher removal), so source producers unblock before the destination begins LME persistence, leader-epoch bump, ABORT  markers, and MIRROR_PID_RESET .
   • Brief in-flight rejection window: Between Unregister commit and fetcher removal, in-flight Fetches still carry MirrorReplicaInfo and the source returns INVALID_SYNC_REGISTRATION per-partition — harmless since the destination is stopping anyway.
   • Fallback if Unregister fails: STOPPING proceeds even if the source is unreachable; an operator can later run --release-sync on the source to clean up the stale registration.
   • Cleanup: STOPPING also clears the per-topic mirror-type state (both effective and desired) on the destination.
3. Source-side release: The source cluster can unilaterally release the sync link when the destination is unreachable. This is destructive — any record acknowledged under sync semantics that has not yet reached the destination is now effectively async-replicated and may be lost on source failure. The source marks each released (MirrorName, TopicId) tuple with a sticky Released=true flag; sync resumes automatically when the destination is reachable again and its auto-promoter completes a recovery promotion. See Source-Side Release below.

Source-Side Release

kafka-cluster-mirrors.sh --release-sync --mirror <mirror-name> [--topics <patterns>] exists for the case where the destination cluster is unreachable from the source (cluster down, network partition) and source producers are blocked because the destination's ISR can no longer be evaluated. In that situation the destination operator cannot intervene to issue an Unregister from destination side — so the source operator unilaterally releases the sync registration to restore producer availability. This is a transient unblock at the cost of zero-RPO for any record acknowledged under sync semantics that hadn't yet replicated; sync resumes automatically once the destination is reachable again and catches up (see Recovery below).

The complementary case — destination is reachable but underperforming, increasing replication latency — is handled on the destination side: the destination operator runs --alter --async and the destination's MirrorCoordinator issues a normal UnregisterMirrorSyncTopicRequest. No source-side release is needed in that case.

Initiated via: kafka-cluster-mirrors.sh --release-sync --mirror <mirror-name> [--topics <patterns>] on the source cluster, which routes through UnregisterMirrorSyncTopicRequest to the source MirrorCoordinator that owns the relevant __mirror_state partition. When --topics is omitted, every topic registered under that destination's mirror name on this source is released.

Source-side flow

1. The source MirrorCoordinator removes the affected topics from the destination's entry in MirrorSyncRegistrationValue and marks each released (MirrorName, TopicId) tuple with Released=true in the same coordinator record. The marker is sticky — it lives in __mirror_state and is rebuilt from there by any new coordinator instance on coordinator failover (i.e., when leadership of the __mirror_state partition changes).
2. For each released topic, the controller emits MirrorSyncRegistrationChangeRecord(Registered=false) for that (MirrorName, TopicId), removing the destination from the topic's registeredMirrors set on TopicImage. When the set becomes empty, source partition leaders stop gating the high watermark on the released topic and pending acks=all produces complete based on local-ISR alone.
3. The source's admission logic for subsequent RegisterMirrorSyncTopicRequest calls depends on the Released=true marker state and the request's RecoveryPromotion flag:

Destination-side handling

The destination learns that a release has occurred via two converging error signals:

• Fetch-time: INVALID_SYNC_REGISTRATION is returned per-partition by the source's Fetch handler whenever a Fetch carries MirrorReplicaInfo for a (MirrorName, TopicId) that has been released or is otherwise not present in MirrorSyncRegistrationValue for this destination — catching destinations that held stale local SYNC state when the release happened.
• Register-time: MIRROR_SYNC_RELEASED_REQUIRES_FRESH_PROMOTION is returned in the RegisterMirrorSyncTopicResponse whenever a destination's promotion-path Register lands on a (MirrorName, TopicId) tuple that the source coordinator has marked Released=true on MirrorSyncRegistrationValue (see Source-side flow step 1). The marker is sticky — it persists across coordinator failovers (rebuilt from __mirror_state by any new coordinator instance) and is cleared only when the source admits a recovery-path Register carrying RecoveryPromotion=true. This signal catches stale or in-flight Registers that were issued by the destination before it learned about the release.

When each signal fires - A destination that was actively sync-mirroring at the moment of release sees INVALID_SYNC_REGISTRATION on its next Fetch — its Fetch loop is still carrying MirrorReplicaInfo for the just-released tuple. The Fetch-time signal is per-partition: each affected partition's response carries it independently. Topic-wide demote happens only after every partition has been failing for the demote-timeout window (see the per-partition handling below). The Register-time MIRROR_SYNC_RELEASED_REQUIRES_FRESH_PROMOTION is for a different scenario: a stale RegisterMirrorSyncTopicRequest from the initial-promotion path was already in flight to the source when release happened (the destination hadn't yet set the recovery hint). That stale Register has RecoveryPromotion=false, lands against the sticky marker, and is rejected — being a topic-level RPC, this directly triggers the topic-wide demote path.

Both signals trigger handling on the destination, but the scope of the response differs:

Per-partition error (Fetch-time INVALID_SYNC_REGISTRATION or MIRROR_REGISTRATION_PROPAGATING). Treated as a per-partition health signal, not a topic-wide event:

1. WARN log with (mirrorName, topicName, partition, reason).
2. Counter metric SyncPartitionFetchFailures{mirror, topic, partition, reason} increments (see Metrics).
3. Fetcher keeps the affected partition in SYNC mode and retries on its next poll. Effective mirrorType for the topic stays SYNC. Other partitions of the topic continue gating normally.
4. Each failing partition tracks an in-memory failing-since timestamp. If every partition of the topic has been continuously failing for mirror.sync.topic.demote.timeout.ms (default 30s), the destination falls back to the topic-wide demote path (below) — this catches the case where the source genuinely released the topic and all leaders eventually return INVALID.

This per-partition handling means a single stale source broker doesn't demote the whole topic, eliminating the demote→re-promote oscillation that a topic-wide reaction would produce. The cost is intra-topic mixed gating during the stale-broker window — producers on source partitions whose leaders haven't applied the registration aren't sync-gated for that partition (the leader returns the per-partition error and isn't gating). Operators see this via SyncPartitionFetchFailures; persistent failures on a specific partition are a broker-health signal pointing at the leader. Operators who can't tolerate partial sync can demote the topic manually via --alter --async.

Aggregation mechanics. The per-partition failing-since tracking and the topic-wide demote trigger live on the destination's MirrorMetadataManager. Per-partition fetchers signal failure (and recovery) of their partition's Fetch to MirrorMetadataManager, which maintains an in-memory per-(mirror, topic, partition) failing-since timestamp; a partition's mark clears as soon as its next Fetch succeeds. The same monitoring loop that drives auto-promotion (watching the desired/effective gap and partition lag) evaluates the all-partitions-failing condition per topic — when every partition of a given topic has been continuously failing for mirror.sync.topic.demote.timeout.ms, MirrorMetadataManager triggers the topic-wide demote path described next. The state is in-memory only: on coordinator restart, per-partition fetchers report fresh failures on their next Fetch and the failing-since clocks restart — bounded extra latency before demote, no correctness impact.

Topic-wide demote (Register-time MIRROR_SYNC_RELEASED_REQUIRES_FRESH_PROMOTION, or all-partitions-failing timeout fired):

1. The destination controller writes MirrorTopicTypeChangeRecord(MirrorType=ASYNC), flipping the effective TopicImage.mirrorType() from SYNC to ASYNC.
2. MirrorFetcherManager.restartFetchersForTopics restarts the affected fetcher in ASYNC mode; subsequent Fetch requests no longer carry MirrorReplicaInfo for that topic.
3. The desired mirror type stays SYNC, reusing the desired-state pattern already established for ASYNC→SYNC auto-promotion. A pendingRecoveryPromotion hint is set on the partition's mirror state — it does not block the auto-promoter, it just marks the next Register issued by this destination to carry RecoveryPromotion=true. The hint is in-memory only; if the coordinator restarts while a topic is mid-recovery, the next auto-promoter attempt will land at the source with RecoveryPromotion=false (default) and be rejected, after which the destination re-establishes the hint and the following attempt carries the right flag — bounded by one extra rejection cycle per coordinator restart.
4. The demote is surfaced for operator visibility:
   • WARN log with (mirrorName, topicName, reason).
   • Counter metric SyncRegistrationRejected{reason=RELEASED_FRESH_PROMOTION_REQUIRED|ALL_PARTITIONS_FAILING} (see Metrics).
   • DescribeClusterMirrors annotates the topic as "released by source — recovery in flight" until a fresh Register is admitted.
5. The mirror partition stays in MIRRORING state — replication is healthy in ASYNC mode while the auto-promoter waits for lag to reach zero. Not a FAILED transition.

Recovery is automatic — the destination's existing auto-promoter watches the desired/effective gap and fires when lag reaches zero, no operator action required. A destination operator who doesn't want auto-recovery can opt out by demoting the topic via --alter --async, which flips desired to ASYNC and stops the auto-promoter from firing.

Race with concurrent destination registration (partial connectivity)

The source's KRaft controller serializes Release and Register requests, so one always commits first. The dangerous interleaving is Release-then-stale-Register, which a naive handler would re-establish silently — erasing the operator's intent within milliseconds (with possibly unreplicated produces acknowledged under no-sync semantics in between). The sticky Released=true marker prevents this: a stale Register lands on the marker and is rejected with MIRROR_SYNC_RELEASED_REQUIRES_FRESH_PROMOTION; the reverse ordering (Register-then-Release) ends in a released state matching operator intent. When multiple destinations are registered against the same source topic, releasing one marks only that destination; remaining destinations continue to gate source HW, and the released one's subsequent Fetches receive INVALID_SYNC_REGISTRATION.

Recovery

Recovery of destination is automatic. The destination's auto-promoter monitors the desired/effective gap and the destination's own lag; the only difference from a normal initial promotion is that the pendingRecoveryPromotion hint marks the next Register to carry RecoveryPromotion=true.

1. The destination's auto-promoter fires when local LEO ≥ source HW on every partition (the same lag-zero condition as initial promotion). It issues a RegisterMirrorSyncTopicRequest with RecoveryPromotion=true.
2. The source admits the registration: with RecoveryPromotion=true, the marker is allowed to be cleared. The source clears the Released=true marker for that (MirrorName, TopicId) and emits a fresh MirrorSyncRegistrationChangeRecord(Registered=true), adding this destination back into the topic's registeredMirrors set so partition leaders resume HW gating against it. The destination clears its pendingRecoveryPromotion hint on success.
3. A Register without RecoveryPromotion=true arriving while the marker is set is rejected with MIRROR_SYNC_RELEASED_REQUIRES_FRESH_PROMOTION — this defends against stale in-flight Registers from before the release. The destination's existing rejection handling (above) flips effective to ASYNC and sets pendingRecoveryPromotion again, so the next auto-promoter attempt carries RecoveryPromotion=true for that topic.

If the source-side condition that prompted the release persists (destination still unreachable), the destination's Register attempts continue to fail. The existing RetryAttempt field on MirrorPartitionStateValue drives exponential backoff between auto-promoter Register attempts — the same backoff applied to other state-machine retries — so persistent unreachability does not produce Register spam.

Sync re-establishment always requires lag-zero on the destination plus admission by the source. The release-then-stale-Register race is still defeated by the marker — the stale Register carries RecoveryPromotion=false since it was issued before the destination handled any rejection, so it is rejected and does not clear the marker.

Auto-release is intentionally not supported in this KIP. Sync gating timeouts can be transient (slow destination, network hiccup, restart), and auto-release would convert these into hard divergences requiring operator-driven recovery on every flap. Operators should rely on metrics (MirrorSyncReplicationLatencyMs and similar) to decide when to release manually.

High Watermark Advancement

Single-cluster recap. Standard Kafka computes the leader's HW as min(LEOs of all replicas in its ISR) — the leader waits for every ISR member to fetch up to a given offset. With KIP-966 (Kafka 3.7+), HW additionally freezes when |ISR| < min.insync.replicas (the under-min-ISR HW freeze), preventing consumers from seeing records not backed by full durability. acks=all writes are rejected with NOT_ENOUGH_REPLICAS_AFTER_APPEND when |ISR| < min.insync.replicas at acknowledgment time.

Source HW. Sync mirror treats every registered destination's ISR replicas as participating in the source partition's HW computation. The source partition leader's HW is the minimum LEO across all participating ISR replicas — its own local ISR plus each registered destination's ISR:

participatingReplicas = sourceISR ∪ (∪ M.ISR for each M in topicImage.registeredMirrors(topicId))
sourceHW = min(r.LEO for each r in participatingReplicas)

Where the single-cluster leader waits for every local ISR replica, sync mirror also waits for every registered destination's ISR replica. There is no separate "source-local HW" that advances independently — source consumers and source producers both gate on this unified sourceHW, so source consumers only ever see records that are durable cross-cluster.

Each destination's per-replica LEOs and ISR membership reach the source via MirrorReplicaInfo carried in Fetch requests.

Under-min-ISR HW freeze applies on all sides. sourceHW freezes (at the last value where all sides were healthy) if |sourceISR| < source.min.insync.replicas OR if any M in topicImage.registeredMirrors(topicId) has |M.reported ISR| < that destination's MinISR. The freeze prevents HW from advancing past records that aren't fully durable everywhere.

Released mirrors are absent from the set. Source-side release is enforced at Register time by the source MirrorCoordinator (sticky Released=true marker on MirrorSyncRegistrationValue); the released destination is also removed from topicImage.registeredMirrors, so partition leaders simply iterate the current set without carrying any per-mirror Released flag. When topicImage.registeredMirrors(topicId) is empty, the predicate collapses to min(LEOs in sourceISR) — standard single-cluster behaviour.

Sync invariant. For any record at offset X that the source has acknowledged under sync semantics, every registered destination's local ISR was already at offset ≥ X before the ack — by definition of sourceHW. This invariant is what lets the Failover Process switch to local-ISR-driven HW without losing any source-acked record — every record acknowledged by the source is, by definition, already durable on each registered destination's local ISR.

Example. Source has 3 local ISR replicas at LEOs (100, 100, 99). Destination A reports ISR at LEOs (95, 97, 96). Destination B reports ISR at LEOs (90, 88).

participating LEOs = {100, 100, 99, 95, 97, 96, 90, 88}
sourceHW = min(...) = 88

Producers with acks=all are acknowledged for records up through offset 88; source consumers can read up through offset 88 — gated on the slowest participant (destination B).

Produce-ack precondition (extended): For acks=all to succeed, two conditions must hold at acknowledgment time:

1. sourceHW has reached the produced offset — which requires every participating ISR replica (source's local ISR plus each destination in topicImage.registeredMirrors(topicId)) to have it.
2. ISR-size checks:
   • |sourceISR| ≥ source.min.insync.replicas → if violated, returns NOT_ENOUGH_REPLICAS_AFTER_APPEND (single-cluster behavior).
   • For each M in topicImage.registeredMirrors(topicId): |M.reported ISR| ≥ M.MinISR → if violated, returns the new NOT_ENOUGH_REMOTE_REPLICAS_AFTER_APPEND with the offending mirror name in the message.

If condition 1 simply does not converge in time (slow replication somewhere, no MinISR violation), the producer's request.timeout.ms expires and MIRROR_SYNC_REPLICATION_TIMEOUT is returned. The MinISR checks in condition 2 are fast-feedback signals — the under-min-ISR HW freeze would also freeze sourceHW indirectly, eventually leading to a timeout, but the explicit checks give producers immediate error responses without waiting.

Destination HW:

• During MIRRORING: the destination's partition leader uses the tracked source HW (from the most recent FetchResponse) as its local HW. This value is exposed to destination consumers and propagated to destination's local followers in their FetchResponses. The destination's local ISR machinery continues to track per-follower LEOs for lag and ISR-membership purposes but does not independently drive HW advancement — source HW does. By the sync invariant, source HW is always at or below the destination's local-ISR min LEO at the time it's used, so every consumer-visible record is locally durable on the destination's ISR.

• On the STOPPING transition (failover): the destination's partition leader undergoes two coordinated changes:

  1. HW computation switches from source-HW-driven to standard local-ISR-driven (HW = min(local ISR LEOs), with under-min-ISR HW freeze applied). The new HW value is at least equal to the last tracked source HW (preserving zero-RPO for all source-acked records) and may be higher if the destination's local LEO had advanced past source HW from in-flight replication. Any such in-flight records become consumer-visible only as the destination's local ISR catches up to them — the standard local-ISR HW rule that drives every Kafka partition leader's HW.
  2. The destination's local leader epoch is bumped via the existing KIP-1279 STOPPING-transition machinery (BumpLeaderEpochs to the destination controller, with the current source leader epoch persisted as LastMirrorEpoch (LME) in __mirror_state). The bumped epoch draws a boundary in the destination's leader-epoch checkpoint between the mirrored era (records appended by the mirror fetcher under source's epochs) and the writable-primary era (records the destination now accepts from producers under the bumped epoch). This same bump and LME persistence are what KIP-1279 already relies on for reverse mirroring (failback): when the old source recovers and is reverse-mirrored from the destination, the recovered source's mirror fetcher uses the KIP-1279 two-phase truncation protocol (LME truncation, then replication-level truncation against the destination's leader-epoch history) to align its log against the new primary before resuming replication. From the perspective of this KIP, no new bump is introduced — the sync-mode HW computation simply switches at the same transition the existing bump already marks.

This is the HW value used for post-failover consumer visibility (see the Failover Process )

Interaction with mirror replication throttling: mirror.replication.throttled.rate applies uniformly to sync topics — an aggressive throttle below the inbound produce rate will surface as MirrorSyncReplicationLatencyMs rising and producers eventually receiving MIRROR_SYNC_REPLICATION_TIMEOUT.

ASYNC→SYNC Promotion

The mirror-type lifecycle across promotion, demotion, source-side release, and recovery.

States:

Transitions:

Both --add --sync (new topic) and --alter --sync (existing topic) use the same path. Promotion is never rejected on a non-zero lag — it is deferred until the destination has caught up.

1. The desired mirror type is set to SYNC, persisted as DesiredMirrorType on MirrorPartitionStateValue (see Mirror Metadata Records). This persistence is what lets the operator's intent survive coordinator failover when promotion is still pending.
2. While any partition has non-zero lag, the effective mirrorType stays ASYNC and the MirrorCoordinator monitors lag.
3. When lag reaches zero across all partitions, the coordinator commits MirrorTopicTypeChangeRecord(SYNC), the fetcher restarts in SYNC mode, and RegisterMirrorSyncTopicRequest is sent to the source.

The CLI returns immediately for both --add --sync and --alter --sync — the response reports whether each topic was promoted (lag was zero) or is pending (a per-topic warning indicates promotion will fire when lag drains)

Failover Process

Failover follows KIP-1279's STOPPING transition (mirror fetcher removed, leader epoch bumped, ABORT markers appended for in-flight transactions, last mirror epochs persisted, MIRROR_PID_RESET record appended). Sync mode's one addition at the same transition is the destination's HW computation switch from source-HW-driven to local-ISR-driven (HW = min(local ISR LEOs))

In producer-visible terms, after failover: records up through the destination's last-tracked source HW are immediately consumer-visible. Records past that point that are durable on the destination's local log become visible as the destination's local ISR catches up — the standard local-ISR-driven HW rule. No log truncation occurs; the change is purely in which rule drives HW advancement.

Why the local-ISR-driven HW rule preserves zero-RPO (worked example): at source-crash time

• the destination's local ISR has caught up to offset X+2 (the destination's mirror fetcher has been pulling records faster than source's local replicas, and the destination's local replicas have replicated more).
• The last FetchResponse the destination received from the source carried source HW = X-1.
• Source's actual HW at crash time is somewhere in [X-1 (last tracked HW), X+2(last fetched offset)] — the destination cannot determine where, because the next FetchResponse (which would have carried the updated source HW) never made it back.

For any record in the range X..X+2, two cases are indistinguishable from the destination's perspective:

• Case A: source advanced HW past this record and acked the producer before crashing. The record was committed cross-cluster.
• Case B: source crashed before advancing HW past this record (or advanced HW but the ack to the producer was lost in flight). The record was not committed.

Capping post-failover visibility at tracked source HW (X-1) would hide all of records X, X+1, X+2 — a zero-RPO violation for any of them that was actually in case A. Switching to local-ISR-driven HW makes the destination's HW reach X+2 (the local-min-ISR LEO), exposing all three records — honouring producer durability for every case-A record and at worst incurring a duplicate-on-retry for case-B records (if the producer was idempotent with a fresh PID after reconnecting, or non-idempotent and re-sending). The sync invariant guarantees that whenever the source advanced its HW to any offset, the destination's local ISR was already at that offset — so every source-acked record in the range is locally recoverable on the destination regardless of which acks made it back to the producer.

Multi-destination failover-target selection: 

When multiple sync destinations are registered for the same source topic, the operator should select the destination with the highest replicated log end offset to maximize preservation of in-flight records (the X..X+2 window above). Per-destination LEO is exposed via DescribeClusterMirrors  as the DestinationOffset  field (inherited from KIP-1279; this field carries the destination's LEO). The operator queries each candidate destination to compare and picks the one furthest ahead. 

Non-selected destinations: #TODO

Check this in KIP-1279. KIP-1279's LME-based truncation alignment is keyed by mirror name, and mirror names don't naturally compose when a destination's source cluster changes (whether on symmetric failback with a direction-specific name, or on cross-destination retargeting). Until KIP-1279 evolves to support LME continuity across mirror identity changes (e.g., by keying LME on a stable partition-or-cluster-pair identifier rather than operator-chosen mirror name), non-selected destinations after a sync-mirror failover follow the same options KIP-1279 provides for async multi-destination failover: release them, or recreate the mirror against the new primary with full re-replication.

Producer-side handling after failover — same semantics as a single-cluster leader handoff:

• Vanilla producer (no idempotence): if the producer replays records that landed on the destination via mirror replication, the destination appends them again, creating duplicates. Same as vanilla producers replaying after any leader handoff.

• Idempotent producer: PID and sequence state is per-session and broker-assigned. After failover the producer reconnects to the destination cluster; InitProducerId against the destination assigns a fresh PID (no cross-cluster PID-state mirroring under this KIP). Re-sent records appended under the new PID are fresh records — duplicates possible. Same behavior as an idempotent producer initiating a new session against the source after a connection drop.

• Transactional producer: failover handling depends on whether the COMMIT marker reached the destination before the source crashed:

  • COMMIT marker absent on destination at failover: the in-flight transaction is aborted as part of the standard leader-handoff transaction recovery. Sync mode bounds the abort set to small set of in-flight transaction at the source-crash moment.
  • COMMIT marker present on destination at failover: the destination's local HW advances to or past the marker offset as local ISR catches up; the transaction becomes visible as committed.

See Feature Improvements > Transactional Producer for the full story including the mixed SYNC/ASYNC caveat and what's deferred to a follow-up KIP.

Failback Process

Failback follows KIP-1279's reverse-mirror creation path: the operator creates the reversed mirror (old destination → old source) with the old source rejoining as the new destination; the new destination uses KIP-1279's two-phase truncation to align its log against the new source; and replication catches up asynchronously in MIRRORING. Sync mode adds nothing to the failback mechanics — the destination-side leader-epoch bump and LastMirrorEpoch persistence at the original STOPPING are KIP-1279 features that this KIP inherits unchanged.

Restoring SYNC after failback: uses the standard ASYNC→SYNC auto-promotion path (see ASYNC→SYNC Promotion): 

1. The reversed mirror is created (old destination → old source) with desired mirrorType=SYNC. Mirroring starts ASYNC regardless.
2. The old source (now destination) truncates its local log using the LME returned from the new source, then catches up asynchronously.
3. When lag reaches zero across all partitions of the topic, auto-promotion fires: the destination commits MirrorTopicTypeChangeRecord(SYNC), restarts its fetcher in SYNC mode, and sends RegisterMirrorSyncTopicRequest to the new source.
4. The new source begins HW gating on the new destination's reported ISR, restoring zero-RPO for further writes.

Unclean Leader Election on Source Cluster

When the source cluster experiences an unclean leader election (ULE), the new source leader may have a divergent (shorter) log compared to what the destination has already replicated. The mirrorLeaderEpoch mechanism inherited from KIP-1279 detects this divergence: the destination tracks the source's leader epoch in FetchRequest/FetchResponse, and when the source epoch advances unexpectedly (indicating ULE), the destination flags the divergence and triggers truncation.

In sync mode, source ULE is handled the same way as in async mode: the destination's fetcher truncates to the divergence point and re-syncs from the new source leader. Sync mode does not preserve zero-RPO through source ULE — any records the destination held above the divergence point are truncated, including records that were previously acked to producers under sync gating.

Kafka treats ULE as an opt-in-to-data-loss scenario (operators set unclean.leader.election.enable=true on the source topic); sync mirroring cannot reconstruct data the source has discarded.

Operators wanting zero-RPO across all failure modes — including ULE — should set unclean.leader.election.enable=false on the source topic. With that setting, the unclean election is blocked entirely; the source partition becomes unavailable until an ISR replica returns, and sync gating naturally pauses (no source leader = no acks). The operator can then choose:

1. Wait for source recovery. If an ISR replica is expected to return soon, the partition will heal with data intact and producers can resume against the original source.

2. Failover to the destination as the new primary. If availability cannot wait, run the standard stop-mirror flow on the destination (kafka-cluster-mirrors.sh --stop --topics ... --mirror M) — the destination topic becomes writable and producers can be redirected immediately. The destination holds all data the source had through the moment ULE was blocked, since sync gating prevented any later acks.

   Once the original source partition is healthy again, the operator can re-establish the mirror in the reverse direction (destination → original source, using KIP-1279's failback flow) to bring the original source up to date with whatever new data the destination accumulated while serving as primary; from there, the operator can fail back if desired.

Public Interfaces

Command-Line Tool

The kafka-cluster-mirrors.sh command-line tool is extended with new options for managing sync mirroring. All examples below assume bin/kafka-cluster-mirrors.sh --bootstrap-server localhost:9091; only the operation-specific arguments are shown.

Source compatibility check. Before submitting --add --sync or --alter --sync, MirrorCommand itself probes the source cluster via the Admin client (using node ApiVersions for Fetch v20 support and Admin.describeFeatures for mirror.sync.version enablement). If the source is incompatible, the CLI fails the command immediately with a clear error and exits non-zero — no desired state is set. If the source is unreachable at probe time, the CLI emits a warning and proceeds; the broker-side Layer-2 check (MIRROR_SYNC_VERSION_MISMATCH at Register time) catches any real incompatibility once the source becomes reachable.

Add topics to a mirror as SYNC:

--add --topic '.*' --mirror my-mirror --sync

Topics are added and then immediately set to SYNC mirror type. The system uses auto-promotion: mirroring starts in ASYNC and automatically promotes to SYNC when lag reaches zero.

Alter existing topics from ASYNC to SYNC:

--alter --topic '.*' --mirror my-mirror --sync

Sets the desired mirror type to SYNC for matched topics. Topics whose partitions all have lag = 0 are promoted immediately. Topics with any partition still lagging stay in ASYNC and auto-promote when lag reaches zero — the CLI returns a per-topic warning indicating the promotion is pending lag drain. Both --add --sync and --alter --sync use the same auto-promotion path described in ASYNC→SYNC Promotion.

Demote topics from SYNC to ASYNC:

--alter --topic '.*' --mirror my-mirror --async

Immediately demotes topics to ASYNC mode. No lag check is performed. In-flight acks=all produces waiting on sync gating complete with source-local-ISR semantics as soon as the demotion commits — they get successful acks under the new (ASYNC) gating predicate, dropping the cross-cluster zero-RPO guarantee for those in-flight requests.

Release a sync registration on the source cluster (destructive):

--release-sync --mirror my-mirror --topics 'orders-.*'

Run on the source cluster when the destination is unreachable. Releasing breaks zero-RPO for any data acked under sync semantics that has not yet reached the destination. Marks each released (MirrorName, TopicId) with a sticky Released=true flag; the destination's auto-promoter resumes sync automatically once it is reachable again and lag reaches zero. Omit --topics to release every topic this destination has registered.

New CLI options:

Admin Client

Three new methods are added to the Admin interface:

// Register topics for sync mirroring on the source cluster
RegisterMirrorSyncTopicsResult registerMirrorSyncTopics(
    String mirrorName, Set<String> topicNames, RegisterMirrorSyncTopicsOptions options);

// Unregister topics from sync mirroring on the source cluster
UnregisterMirrorSyncTopicsResult unregisterMirrorSyncTopics(
    String mirrorName, Set<String> topicNames, UnregisterMirrorSyncTopicsOptions options);

// Alter the mirror type of topics (ASYNC <-> SYNC). Promotion to SYNC is deferred per-topic until lag reaches zero.
AlterMirrorTopicTypeResult alterMirrorTopicType(
    String mirrorName, Map<String, AlterMirrorTopicTypeEntry> topics, AlterMirrorTopicTypeOptions options);

Each method has a convenience overload without the Options parameter.

AlterMirrorTopicTypeEntry:

public class AlterMirrorTopicTypeEntry {
    private final MirrorType mirrorType;

    public AlterMirrorTopicTypeEntry(MirrorType mirrorType) { this.mirrorType = mirrorType; }

    public MirrorType mirrorType() { return mirrorType; }
}

Promotion to SYNC always requires lag=0 across all partitions of the topic; there is no configurable threshold.

Result classes:

AlterMirrorTopicTypeResult provides per-topic futures so callers can observe per-topic outcomes: success when the desired type is set (the future completes when the metadata record is committed, regardless of whether effective promotion has happened yet), or per-topic failure for cases like topic-not-found or unauthorized.

Source compatibility — programmatic callers. MirrorCommand runs an eager source-compatibility probe before submitting alterMirrorTopicType or addTopicsToMirror with MirrorType=SYNC (see Command-LineTool). Direct callers of the Admin API bypass this check. Direct callers who want immediate feedback should probe the source themselves (using node ApiVersions for Fetch v20 and Admin.describeFeatures for mirror.sync.version) before issuing the request. Without a pre-probe, compatibility issues surface asynchronously via the SyncRegistrationRejected{reason=VERSION_MISMATCH} metric and a DescribeClusterMirrors annotation once the auto-promoter attempts to register.

Authorization

The new operations introduced by this KIP reuse the CLUSTER_MIRRORS resource type from KIP-1279. Permissions:

UnregisterMirrorSyncTopicRequest is the only RPC schema with two distinct authorization contexts depending on caller: broker-issued from the destination's MirrorCoordinator requires CLUSTER_ACTION (inter-broker), while operator-issued via --release-sync on the source CLI requires ALTER on the CLUSTER_MIRRORS resource.

MirrorType

public enum MirrorType {
    ASYNC((byte) 0),
    SYNC((byte) 1);
    private final byte id;
    MirrorType(byte id) { this.id = id; }
    public byte id() { return id; }
    public static MirrorType fromId(byte id) {
        for (MirrorType t : values()) {
            if (t.id == id) return t;
        }
        throw new IllegalArgumentException("Unknown MirrorType id: " + id);
    }
    public static MirrorType fromString(String value) {
        if (value == null || value.isEmpty()) return ASYNC;
        return valueOf(value.toUpperCase());
    }
}

The int8 id is the wire-format representation used in all RPC schemas and the MirrorTopicTypeChangeRecord metadata record. The fromString helper is for CLI parsing of user-supplied "ASYNC"/"SYNC" strings.

Protocol APIs

AddTopicsToMirror (API Key 95) — Extended

The AddTopicsToMirrorRequest is extended to include a MirrorType field per topic:

{
  "apiKey": 95,
  "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+" },
      { "name": "TopicName", "type": "string", "versions": "0+", "mapKey": true,
        "entityType": "topicName" },
      { "name": "MirrorName", "type": "string", "versions": "0+",
        "nullableVersions": "0+" },
      { "name": "MirrorType", "type": "int8", "versions": "0+", "default": "0",
        "about": "The desired mirror link type as an int8: 0 = ASYNC (default), 1 = SYNC. Set to 1 to add the topic with desired SYNC; the topic still starts in effective ASYNC and auto-promotes once lag reaches zero (immediate for newly-added topics with no records). See the MirrorType enum." }
    ]}
  ]
}

RegisterMirrorSyncTopicRequest (API Key TBD)

Sent from the destination cluster to the source cluster to register topics for sync mirroring. 

{
  "apiKey": TBD,
  "type": "request",
  "listeners": ["broker", "controller"],
  "name": "RegisterMirrorSyncTopicRequest",
  "latestVersionUnstable": true,
  "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 topics to register for sync mirroring.", "fields": [
      { "name": "TopicId", "type": "uuid", "versions": "0+" },
      { "name": "TopicName", "type": "string", "versions": "0+", "mapKey": true,
        "entityType": "topicName" },
      { "name": "RecoveryPromotion", "type": "bool", "versions": "0+", "default": "false",
        "about": "True when this topic's Register entry is from a post-release recovery (the destination's pendingRecoveryPromotion hint was set on this topic after a prior MIRROR_SYNC_RELEASED_REQUIRES_FRESH_PROMOTION or INVALID_SYNC_REGISTRATION). Authorizes the source to clear a sticky Released=true marker for this (mirror, topic) tuple on admission. Set to false for non-recovery promotions. Per-topic so a single Register can batch a mix of non-recovery promotions and recoveries." }
    ]}
  ]
}

Response:

{
  "apiKey": TBD,
  "type": "response",
  "name": "RegisterMirrorSyncTopicResponse",
  "validVersions": "0",
  "flexibleVersions": "0+",
  "fields": [
    { "name": "ThrottleTimeMs", "type": "int32", "versions": "0+" },
    { "name": "Topics", "type": "[]TopicResult", "versions": "0+",
      "about": "Per-topic results so the destination can distinguish admitted topics from rejected ones (e.g., MIRROR_SYNC_RELEASED_REQUIRES_FRESH_PROMOTION on a subset).", "fields": [
      { "name": "TopicName", "type": "string", "versions": "0+",
        "entityType": "topicName" },
      { "name": "ErrorCode", "type": "int16", "versions": "0+" },
      { "name": "ErrorMessage", "type": "string", "versions": "0+",
        "nullableVersions": "0+", "ignorable": true }
    ]}
  ]
}

UnregisterMirrorSyncTopicRequest (API Key TBD)

Sent from the destination cluster to the source cluster to unregister topics from sync mirroring. Structurally identical to RegisterMirrorSyncTopicRequest.

{
  "apiKey": TBD,
  "type": "request",
  "listeners": ["broker", "controller"],
  "name": "UnregisterMirrorSyncTopicRequest",
  "latestVersionUnstable": true,
  "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 topics to unregister from sync mirroring.", "fields": [
      { "name": "TopicId", "type": "uuid", "versions": "0+" },
      { "name": "TopicName", "type": "string", "versions": "0+", "mapKey": true,
        "entityType": "topicName" }
    ]}
  ]
}

Response:

{
  "apiKey": TBD,
  "type": "response",
  "name": "UnregisterMirrorSyncTopicResponse",
  "validVersions": "0",
  "flexibleVersions": "0+",
  "fields": [
    { "name": "ThrottleTimeMs", "type": "int32", "versions": "0+" },
    { "name": "ErrorCode", "type": "int16", "versions": "0+" },
    { "name": "ErrorMessage", "type": "string", "versions": "0+",
      "nullableVersions": "0+", "ignorable": true }
  ]
}

AlterMirrorTopicTypeRequest (API Key TBD)

Alters the desired mirror type of topics in a mirror link. Promotion from ASYNC to SYNC is deferred per-topic until lag reaches zero (never rejected on lag); demotion from SYNC to ASYNC takes effect immediately.

{
  "apiKey": TBD,
  "type": "request",
  "listeners": ["broker"],
  "name": "AlterMirrorTopicTypeRequest",
  "latestVersionUnstable": true,
  "validVersions": "0",
  "flexibleVersions": "0+",
  "fields": [
    { "name": "MirrorName", "type": "string", "versions": "0+",
      "about": "The cluster mirror name." },
    { "name": "Topics", "type": "[]AlterMirrorTopic", "versions": "0+",
      "about": "The topics to alter.", "fields": [
      { "name": "TopicId", "type": "uuid", "versions": "0+" },
      { "name": "TopicName", "type": "string", "versions": "0+", "mapKey": true,
        "entityType": "topicName" },
      { "name": "MirrorType", "type": "int8", "versions": "0+",
        "about": "The desired mirror type as an int8: 0 = ASYNC, 1 = SYNC. See the MirrorType enum. Promotion to SYNC is deferred per-topic until lag reaches zero across all partitions." }
    ]}
  ]
}

Response (per-topic errors for partial success):

{
  "apiKey": TBD,
  "type": "response",
  "name": "AlterMirrorTopicTypeResponse",
  "validVersions": "0",
  "flexibleVersions": "0+",
  "fields": [
    { "name": "ThrottleTimeMs", "type": "int32", "versions": "0+" },
    { "name": "Topics", "type": "[]TopicResult", "versions": "0+",
      "about": "Per-topic results.", "fields": [
      { "name": "TopicName", "type": "string", "versions": "0+",
        "entityType": "topicName" },
      { "name": "ErrorCode", "type": "int16", "versions": "0+" },
      { "name": "ErrorMessage", "type": "string", "versions": "0+",
        "nullableVersions": "0+", "ignorable": true }
    ]}
  ]
}

Produce (API Key 0) — Extended to v14

The Produce API version is bumped to v14 to register two new per-partition error codes — MIRROR_SYNC_REPLICATION_TIMEOUT and NOT_ENOUGH_REMOTE_REPLICAS_AFTER_APPEND (see Errors). The response schema records the introducing version following Kafka's existing convention for new error codes (analogous to the v4 annotation on ProduceResponse.json for KAFKA_STORAGE_ERROR):

// Version 14 added MIRROR_SYNC_REPLICATION_TIMEOUT and NOT_ENOUGH_REMOTE_REPLICAS_AFTER_APPEND
// as possible per-partition error codes.

No request-side schema changes — the version bump exists solely to gate the new error codes via the ApiVersions handshake. Brokers receiving a ProduceRequest whose negotiated version is below v14 down-map both new error codes to NOT_ENOUGH_REPLICAS_AFTER_APPEND before responding (see Producer Client Compatibility for the rationale, the precedent, and the per-code mapping).

Fetch (API Key 1) — Extended to v20

The FetchRequest is extended to version 20 with a new top-level MirrorReplicaInfo field. This field carries the destination cluster's mirror name and replica state back to the source, enabling the source to identify which registered destination is reporting and to gate high-watermark advancement on remote ISR.

{ "name": "MirrorReplicaInfo", "type": "MirrorReplicaInfo", "versions": "20+",
  "ignorable": true,
  "about": "Details of replicas on the destination cluster for sync mirroring.",
  "fields": [
    { "name": "MirrorName", "type": "string", "versions": "20+",
      "entityType": "mirrorName",
      "about": "The destination cluster's mirror name. The source uses this to identify which registered destination is reporting replica state, to track per-destination ISR when multiple destinations register the same source topic, and to reject MirrorReplicaInfo from destinations whose registration has been released or is not yet visible on this broker (returns INVALID_SYNC_REGISTRATION or MIRROR_REGISTRATION_PROPAGATING per-partition; see [Errors](#errors))." },
    { "name": "Topics", "type": "[]MirroredTopics", "versions": "20+",
      "about": "The mirrored topics with replica info.", "fields": [
      { "name": "Topic", "type": "string", "versions": "20+",
        "entityType": "topicName" },
      { "name": "TopicId", "type": "uuid", "versions": "20+" },
      { "name": "Partitions", "type": "[]MirroredTopicPartition", "versions": "20+",
        "about": "The partitions with replica info.", "fields": [
        { "name": "Partition", "type": "int32", "versions": "20+" },
        { "name": "ReplicaLastOffsets", "type": "[]ReplicaLastOffsets", "versions": "20+",
          "about": "Replicas last offsets on the destination cluster.", "fields": [
          { "name": "ReplicaId", "type": "int32", "versions": "20+" },
          { "name": "LastFetchOffset", "type": "int64", "versions": "20+" }
        ]},
        { "name": "ISR", "type": "[]int32", "versions": "20+",
          "about": "List of current ISR replica IDs on the destination cluster." },
        { "name": "MinISR", "type": "int32", "versions": "20+",
          "about": "The configured min.insync.replicas on the destination cluster." }
      ]}
    ]}
  ]}

The MirrorReplicaInfo field is marked ignorable, so older brokers (< v20) will simply ignore it. This ensures backward compatibility.

MinISR is reported per-partition for schema uniformity even though min.insync.replicas is a topic-level config — all partitions of a topic carry the same value. Same redundancy pattern as DesiredMirrorType on MirrorPartitionStateValue (see MirrorMetadata Records).

DescribeClusterMirrors — Extended

The destination's DescribeClusterMirrorsResponse (from KIP-1279) is extended with three per-topic fields so operators can see both the operator-requested mirror type and the currently effective one, and detect topics that are currently auto-recovering after a source-side release:

{ "name": "DesiredMirrorType", "type": "int8", "versions": "0+", "default": "0",
  "about": "The operator-requested mirror type: 0 = ASYNC, 1 = SYNC. Differs from EffectiveMirrorType when promotion is pending lag drain, or when RecoveryPending is true." },
{ "name": "EffectiveMirrorType", "type": "int8", "versions": "0+", "default": "0",
  "about": "The currently effective mirror type: 0 = ASYNC, 1 = SYNC. Drives fetcher behavior on destination brokers." },
{ "name": "RecoveryPending", "type": "bool", "versions": "0+", "default": "false",
  "about": "True if this destination has a pending recovery promotion for the topic — sync was rejected by the source's released-marker, the auto-promoter is waiting for lag to reach zero, and the next RegisterMirrorSyncTopicRequest will carry RecoveryPromotion=true. Clears on successful Register admission." }

Interpreting the three values together:

Cluster Metadata Records

Sync cluster mirroring state is propagated to brokers via dedicated KRaft metadata records, following the pattern KIP-1279 already established (mirrorName carried by MirrorTopicStateChangeRecord, surfaced as a first-class field on TopicImage).

This KIP extends the same pattern with two new KRaft records — one on each cluster role:

• Destination cluster: MirrorTopicTypeChangeRecord carries the per-topic mirror type (ASYNC or SYNC) and drives fetcher mode.
• Source cluster: MirrorSyncRegistrationChangeRecord carries one entry per (TopicId, MirrorName) registration change. Source brokers materialize these as Set<MirrorName> registeredMirrors per topic on TopicImage, and partition leaders use that set to decide which destinations to gate HW on. Sync mode is derived state — a topic is sync-gated iff its registeredMirrors set is non-empty — so there is no separate operator-settable "sync required" config.

MirrorTopicTypeChangeRecord (Destination Cluster)

{
  "apiKey": 31,
  "type": "metadata",
  "name": "MirrorTopicTypeChangeRecord",
  "validVersions": "0",
  "flexibleVersions": "0+",
  "fields": [
    { "name": "TopicId", "type": "uuid", "versions": "0+",
      "about": "The topic whose mirror type is being altered." },
    { "name": "MirrorType", "type": "int8", "versions": "0+",
      "about": "0 = ASYNC, 1 = SYNC. Controls fetcher mode on destination brokers." }
  ]
}

The controller commits this record synchronously with the AlterMirrorTopicType API response, so by the time the API returns every destination broker converges on the new type within one metadata propagation cycle.

MirrorSyncRegistrationChangeRecord (Source Cluster)

{
  "apiKey": 32,
  "type": "metadata",
  "name": "MirrorSyncRegistrationChangeRecord",
  "validVersions": "0",
  "flexibleVersions": "0+",
  "fields": [
    { "name": "TopicId", "type": "uuid", "versions": "0+",
      "about": "The source topic whose registration set is being updated." },
    { "name": "MirrorName", "type": "string", "versions": "0+", "entityType": "mirrorName",
      "about": "The destination cluster's mirror name being added to or removed from the topic's registration set." },
    { "name": "Registered", "type": "bool", "versions": "0+",
      "about": "True when this destination is being added to the topic's registration set; false when it is being removed (clean unregister or source-side release)." }
  ]
}

Source brokers replay this record to maintain a Set<MirrorName> registeredMirrors per topic on TopicImage. Partition leaders read that set on every produce-ack evaluation: if non-empty, HW is gated on the union of source-local ISR LEOs and each destination's reported ISR LEOs (MirrorReplicaInfo from Fetch v20 — see High Watermark Advancement); if empty, the topic falls back to single-cluster ack semantics on the produce path. Sync mode is therefore derived state — there is no separate operator-settable "sync required" config.

The record is the propagation channel only. The source's MirrorCoordinator is the writer — it owns the per-(mirror, topic) registration state on __mirror_state (including sticky Released=true markers, described next), and on each Register/Unregister/Release transition it asks the source controller to emit a corresponding MirrorSyncRegistrationChangeRecord so source brokers' TopicImage view stays in sync. Released markers themselves do not appear in this KRaft record — they are coordinator-only state that gates the next Register attempt; once a destination is released, it is simply absent from the topic's registeredMirrors set, and partition leaders need no Released-aware logic.

Mirror Metadata Records

This section describes the coordinator records added or extended by this KIP. Records live in the __mirror_state internal topic on each cluster as applicable.

• MirrorPartitionStateValue (apiKey 2) is extended on the destination with a DesiredMirrorType field so the deferred ASYNC→SYNC promotion intent survives coordinator failover.
• MirrorSyncRegistration (apiKey 3) is a new coordinator record on the source for per-destination registration tracking.

MirrorPartitionStateValue (Destination Cluster, Extended)

A new DesiredMirrorType field is added in version 1 of the record. The destination's MirrorCoordinator writes it whenever an operator's AlterMirrorTopicType request changes the desired type. The effective type lives in the KRaft MirrorTopicTypeChangeRecord (apiKey 31); the gap between desired and effective is what the coordinator monitors to drive auto-promotion when lag reaches zero. The value persists across coordinator failovers, so deferred promotions still fire after a restart.

All partitions of a topic carry the same DesiredMirrorType value — accepted as a small redundancy cost to reuse the existing per-partition record.

Extended Value:

{
  "apiKey": 2,
  "type": "coordinator-value",
  "name": "MirrorPartitionStateValue",
  "validVersions": "0-1",
  "flexibleVersions": "0+",
  "fields": [
    // existing fields (TopicName, Partition, State, PreviousState, RetryAttempt) unchanged
    { "name": "DesiredMirrorType", "type": "int8", "versions": "1+", "default": "0",
      "about": "The desired mirror type for this partition's topic (0 = ASYNC, 1 = SYNC). Set when an operator's AlterMirrorTopicType request changes the desired type; the destination's MirrorCoordinator triggers ASYNC→SYNC promotion when this differs from the effective mirrorType in TopicImage and lag reaches zero. Defaults to 0 (ASYNC) for records written under version 0." }
  ]
}

Brokers replaying older (v0) records default to 0 (ASYNC).

MirrorSyncRegistration (Source Cluster)

Tracks which topics each destination cluster has registered for sync mirroring against this source cluster, and which (MirrorName, TopicId) tuples are marked Released=true by source-side operator action. The source's MirrorCoordinator owns this record and rebuilds the per-mirror registration ledger on startup or partition leadership change. On each Register/Unregister/Release transition the coordinator also emits a MirrorSyncRegistrationChangeRecord (described above) so source partition leaders see the change via metadata propagation. The sticky Released flag is coordinator-only state — it gates the next RegisterMirrorSyncTopicRequest for that tuple but does not appear in the KRaft record.

Key:

{
  "apiKey": 3,
  "type": "coordinator-key",
  "name": "MirrorSyncRegistrationKey",
  "validVersions": "0",
  "flexibleVersions": "0+",
  "fields": [
    { "name": "MirrorName", "type": "string", "versions": "0+", "entityType": "mirrorName",
      "about": "The destination cluster's mirror name (registered by the destination via RegisterMirrorSyncTopicRequest)." }
  ]
}

Value:

{
  "apiKey": 3,
  "type": "coordinator-value",
  "name": "MirrorSyncRegistrationValue",
  "validVersions": "0",
  "flexibleVersions": "0+",
  "fields": [
    { "name": "Topics", "type": "[]SyncTopic", "versions": "0+",
      "about": "Topics this destination has registered (or had released) under this mirror name.", "fields": [
      { "name": "Name", "type": "string", "versions": "0+",
        "about": "The topic name." },
      { "name": "TopicId", "type": "uuid", "versions": "0+",
        "about": "The source topic ID." },
      { "name": "Released", "type": "bool", "versions": "0+", "default": "false",
        "about": "True if the source operator has released this (mirror, topic) tuple. While true, RegisterMirrorSyncTopicRequest for this tuple returns MIRROR_SYNC_RELEASED_REQUIRES_FRESH_PROMOTION; the marker is cleared when the source admits a RegisterMirrorSyncTopicRequest carrying RecoveryPromotion=true (issued by the destination's auto-promoter once it observes lag has reached zero)." }
    ]}
  ]
}

Errors

This section lists the protocol-level errors introduced by this KIP. Error codes are assigned during KIP acceptance; entries here use TBD as placeholders.

MIRROR_SYNC_REPLICATION_TIMEOUT is returned when the destination's ISR fails to catch up to the produced offset before the producer's request.timeout.ms expires. The mirror name in the message identifies the bottleneck; when multiple sync mirrors are registered, it names the slowest. Compare with NOT_ENOUGH_REPLICAS_AFTER_APPEND, which is returned when the source's local ISR fails to catch up — distinguishing local from cross-cluster bottlenecks. Retriable, same as NOT_ENOUGH_REPLICAS_AFTER_APPEND. This error code is only returned to producers that negotiated Produce v14 or higher; producers on older negotiated versions receive NOT_ENOUGH_REPLICAS_AFTER_APPEND instead (see Producer Client Compatibility).

NOT_ENOUGH_REMOTE_REPLICAS_AFTER_APPEND is the fast-feedback variant of the under-min-ISR HW freeze on the destination: returned immediately when any registered destination's reported ISR is below its MinISR at ack time, instead of waiting for the timeout. The mirror name in the message names the offending destination. This error code is only returned to producers that negotiated Produce v14 or higher; producers on older negotiated versions receive NOT_ENOUGH_REPLICAS_AFTER_APPEND instead (see Producer Client Compatibility).

INVALID_SYNC_REGISTRATION and MIRROR_SYNC_RELEASED_REQUIRES_FRESH_PROMOTION are described in the Source-Side Release section.

Metrics

This KIP introduces the following metrics. All are JMX-accessible under the kafka.server domain, organized by cluster role (source vs destination — a single Kafka cluster can play both roles for different mirrors and expose both sets).

Source-side metrics (visible on a cluster that is the source of one or more sync mirrors):

Destination-side metrics (visible on a cluster that is a destination of one or more mirrors):

Compatibility, Deprecation, and Migration Plan

Synchronous cluster mirroring will be introduced after KIP-1279 reaches the "Preview" stage. Any topic already in asynchronous cluster mirroring mode can be promoted to sync mode via the standard --alter --sync path; the auto-promotion mechanism handles catch-up.

One new destination-broker configuration. This KIP introduces mirror.sync.topic.demote.timeout.ms (destination-side, type long, default 30000 = 30 seconds) controlling how long every partition of a sync-mirrored topic must be continuously failing (returning INVALID_SYNC_REGISTRATION or MIRROR_REGISTRATION_PROPAGATING) before the destination demotes the topic to ASYNC. Per-partition errors do not by themselves trigger a topic-wide demote; the timeout exists to detect topic-wide release scenarios where eventually every leader returns INVALID. The default is generously larger than typical KRaft metadata propagation latency. All other gating uses existing settings: min.insync.replicas on each side (with the under-min-ISR HW freeze), the producer's request.timeout.ms (propagated to the broker via ProduceRequest.timeoutMs), and KIP-1279's existing mirror configurations. The ASYNC→SYNC promotion precondition is always lag=0; there is no configurable threshold for that.

Release Phases

Sync mirroring follows the same phased rollout pattern as KIP-1279.

• Early Access. Disabled by default. Requires KIP-1279 to be at Preview or GA on both source and destination. Both clusters must explicitly enable unstable API/feature versions (unstable.api.versions.enable=true, unstable.feature.versions.enable=true) and the mirror.sync.version=1 feature flag. APIs and schemas may change without compatibility guarantees. For evaluation in non-production environments only.
• Preview. Both clusters still opt in to the mirror.sync.version feature. Upgrade from Early Access is supported with compatibility guarantees. Suitable for pre-production and pilot deployments.
• General Availability. Enabled by default when both clusters reach the corresponding production metadata version. Supported under Kafka's standard compatibility guarantees. Preview clusters upgrade seamlessly to GA without migration steps.

Compatibility Matrix

Producer Client Compatibility

This KIP introduces two new per-partition error codes on the Produce response — MIRROR_SYNC_REPLICATION_TIMEOUT and NOT_ENOUGH_REMOTE_REPLICAS_AFTER_APPEND (see Errors). Producer clients built before the sync-mirroring release do not recognize these codes; if a broker returned them directly to such clients, they would surface as UNKNOWN_SERVER_ERROR, which most producer clients treat as non-retriable and fail the produce.

Down-map for older negotiated Produce versions. To preserve the existing retriable semantics for old producers, the source broker translates both new error codes to NOT_ENOUGH_REPLICAS_AFTER_APPEND for any ProduceRequest whose negotiated version is below v14:

Both new error codes are cross-cluster variants of NOT_ENOUGH_REPLICAS_AFTER_APPEND (one timeout-style, one immediate under-min-ISR feedback), so the down-map preserves retriability and the producer's retry loop continues to function. The cost is that an old producer client cannot distinguish a cross-cluster bottleneck from a source-local bottleneck in its error messages — operators who need per-mirror diagnosis on the producer side must upgrade clients.

Precedent. This follows Kafka's established pattern for adding new error codes to existing APIs: bump the API version, annotate the response schema with the introducing version, and down-map the new code to a semantically-closest older code in the broker handler when the negotiated request version predates the bump. Concrete precedents:

• Fetch: KAFKA_STORAGE_ERROR → NOT_LEADER_OR_FOLLOWER for negotiated Fetch version ≤ 5 (KafkaApis.maybeDownConvertStorageError).
• InitProducerId: PRODUCER_FENCED → INVALID_PRODUCER_EPOCH for negotiated InitProducerId version < 4.
• EndTxn, AddPartitionsToTxn: same PRODUCER_FENCED → INVALID_PRODUCER_EPOCH translation for negotiated version < 2.

The Produce down-map introduced here will be implemented as a maybeDownConvertSyncMirrorError helper on the broker, called on each per-partition response before the ProduceResponse is serialized.

No required producer upgrade. Old producers continue to operate against sync-mirrored topics without schema or behavioral breakage. Producers that want to surface the new diagnostic error codes upgrade to a client version that supports Produce v14.

Migration from Async-Only Cluster Mirroring (KIP-1279)

No migration steps required for existing async mirrors — they continue to operate unchanged. To enable sync on an existing topic, run --alter --sync (or call AlterMirrorTopicType with MirrorType=SYNC). The standard auto-promotion path handles catch-up: the topic remains in ASYNC until lag reaches zero, then auto-promotes.

Performance

Sync mirroring trades cross-cluster latency for zero-RPO. The shape of that trade-off:

• Acknowledgment latency is bounded below by network RTT (Round-Trip-Time) to the slowest registered destination plus that destination's own internal ISR catchup — there is no way to ack faster than the slowest path.
• Multiple sync destinations: latency tracks the slowest destination, so each additional sync mirror can only make ack latency worse, never better. Async mirrors registered alongside do not affect produce latency.
• Throughput on the source is not directly reduced by sync gating — the produce path itself runs at source-cluster speed. What changes is in-flight depth: producers waiting on sync gating hold pending requests longer, so achieved throughput at fixed max.in.flight.requests.per.connection drops with longer ack latency.
  • Tuning: size max.in.flight.requests.per.connection and producer batching against expected sync-gated latency (RTT to slowest destination + that destination's internal ISR catchup), not against source-local-only latency.
• Throttling (mirror.replication.throttled.rate) interacts with sync mode the same way as async: throttle below inbound produce rate, and MirrorSyncReplicationLatencyMs rises until producers receive MIRROR_SYNC_REPLICATION_TIMEOUT. See High Watermark Advancement.

Test Plan

Unit Tests

Unit tests will cover individual component behavior:

• MirrorCoordinator and MirrorMetadataManager on both source and destination clusters
• MirrorFetcherThread sending replica info in sync mode
• MirrorCommand can alter the mirror type, add topics with --sync, and defer promotion (with a warning) when lag is non-zero — promotion is automatic once lag reaches zero, never rejected
• ReplicaManager and Partition on source waits for remote replicas
• AlterMirrorTopicType deferred-promotion: when called with lag > 0, topic stays ASYNC with desired=SYNC and auto-promotes when lag reaches zero (never rejects on lag)
• STOPPING transition unregisters sync topics and cleans up mirror type state — specifically, UnregisterMirrorSyncTopicRequest is sent as the first step (before fetcher removal); STOPPING proceeds even if the Unregister fails
• Source MirrorCoordinator admits a normal Register (RecoveryPromotion=false) when no Released=true marker is present, emits MirrorSyncRegistrationChangeRecord(Registered=true), and partition leaders begin gating after the destination's first SYNC-mode Fetch arrives
• Source-side Released=true marker semantics: stale Register (RecoveryPromotion=false) rejected with MIRROR_SYNC_RELEASED_REQUIRES_FRESH_PROMOTION; fresh recovery Register (RecoveryPromotion=true) clears the marker
• MirrorCommand source-compatibility probe: rejects --alter --sync / --add --sync with a clear error when the source's Fetch version < 20 or mirror.sync.version is disabled; emits a warning and proceeds when the source is unreachable; succeeds when the source is compatible
• Source MirrorCoordinator Layer-2 validation: returns MIRROR_SYNC_VERSION_MISMATCH per-topic on RegisterMirrorSyncTopicRequest when its own broker can't gate sync; destination handler keeps effective ASYNC and retries with RetryAttempt-driven backoff
• Source partition leader propagation-race detection: returns MIRROR_REGISTRATION_PROPAGATING when appliedMetadataOffset < latestKnownControllerCommitOffset and the mirror is unknown; returns INVALID_SYNC_REGISTRATION otherwise
• Destination MirrorMetadataManager handles per-partition INVALID_SYNC_REGISTRATION / MIRROR_REGISTRATION_PROPAGATING as per-partition signals: log + SyncPartitionFetchFailures metric increment + fetcher retries that partition in SYNC mode. Topic effective mirrorType stays SYNC. Topic-wide demote fires only when every partition has been continuously failing for mirror.sync.topic.demote.timeout.ms (default 30s), or on Register-time MIRROR_SYNC_RELEASED_REQUIRES_FRESH_PROMOTION. On topic-wide demote, sets pendingRecoveryPromotion hint and partition state stays MIRRORING.
• Source HW: sourceHW = min(LEOs over participating ISR replicas) correctly excludes released mirrors and applies the under-min-ISR HW freeze when source or any destination's |ISR| < min.insync.replicas
• Produce per-partition error down-mapping: when the ProduceRequest negotiated version is below v14, both MIRROR_SYNC_REPLICATION_TIMEOUT and NOT_ENOUGH_REMOTE_REPLICAS_AFTER_APPEND are rewritten to NOT_ENOUGH_REPLICAS_AFTER_APPEND on the per-partition response; for Produce v14 or higher both codes are returned as-is; other error codes are unaffected at every negotiated version

Integration Tests

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

• CLI Workflow: Add topics with --sync, alter mirror type, and verify with kafka-cluster-mirrors.sh
• Sync Replication: Create mirror via API, replicate topic in sync mode, verify data consistency across clusters
• Lag-Deferred Promotion: Trigger --alter --sync while lag > 0, verify the topic stays ASYNC with desired=SYNC and CLI returns a "pending lag drain" warning; produce until lag drains; verify auto-promotion fires when lag reaches 0.
• Auto-Promotion: Add topic as SYNC, verify it starts ASYNC and auto-promotes to SYNC when lag reaches zero
• Topic Removal: Remove sync topic from mirror, verify source unregistration as first STOPPING step and mirror type state cleanup
• Sync Replication Timeout: Throttle the destination such that lag cannot close within the producer's request.timeout.ms; verify producer receives MIRROR_SYNC_REPLICATION_TIMEOUT with the mirror name in the error message
• Source MinISR Violation: Bring destination's ISR below its MinISR during sustained produce; verify producer receives NOT_ENOUGH_REMOTE_REPLICAS_AFTER_APPEND with the mirror name; verify sourceHW freezes (under-min-ISR HW freeze) and resumes when destination's ISR recovers
• Producer Client Backward Compatibility (Produce < v14): Run a producer pinned to a pre-sync-mirroring Produce API version (v13 or below) against a sync-mirrored topic, exercising both new error paths:
  • Timeout path: throttle the destination so lag cannot close within the producer's request.timeout.ms; verify the old producer receives NOT_ENOUGH_REPLICAS_AFTER_APPEND (the down-mapped equivalent of MIRROR_SYNC_REPLICATION_TIMEOUT), the producer retries under its existing retriable-error semantics, and the produce eventually succeeds once the throttle is lifted.
  • Under-MinISR path: bring the destination's ISR below its MinISR during sustained produce; verify the old producer receives NOT_ENOUGH_REPLICAS_AFTER_APPEND (the down-mapped equivalent of NOT_ENOUGH_REMOTE_REPLICAS_AFTER_APPEND), retries under its existing retriable-error semantics, and the produce eventually succeeds once destination ISR recovers.
  • Control case: repeat both scenarios with a producer that negotiates Produce v14 or higher; verify the producer receives the original MIRROR_SYNC_REPLICATION_TIMEOUT and NOT_ENOUGH_REMOTE_REPLICAS_AFTER_APPEND codes (no down-map), and that the mirror name is present in the error message in both cases.
• Source-Side Release — Concurrent Register Race: Initiate --release-sync on the source while a stale RegisterMirrorSyncTopicRequest (issued by the destination's initial-promotion path with RecoveryPromotion=false) is in flight; verify the Register lands against the sticky Released=true marker and is rejected with MIRROR_SYNC_RELEASED_REQUIRES_FRESH_PROMOTION; verify source does not silently re-establish sync; verify destination demotes effective mirrorType to ASYNC and sets the pendingRecoveryPromotion hint
• Source-Side Release — Multi-Destination: With two destinations registered against the same source topic, release one; verify only the released destination is removed from registeredMirrors, the remaining destination continues to gate source HW, and the released destination's subsequent Fetches receive INVALID_SYNC_REGISTRATION
• Source-Side Release — Auto-Recovery: Make the destination temporarily unreachable; release on source; restore reachability; verify the destination's auto-promoter fires when lag reaches zero, the recovery-path Register carries RecoveryPromotion=true, the source admits it and clears the Released=true marker, and sync resumes — no operator action on the destination required
• Source-Side Release — Persistent Source Hold: Release on source; keep the source-side condition active so the marker stays held on subsequent admission attempts; verify the destination's auto-promoter applies exponential backoff via RetryAttempt and does not spam Registers
• Auto-Demote Keeps Partition in MIRRORING: Force the destination to issue Register against a source with Released=true; verify destination auto-flips effective mirrorType to ASYNC, sets the pendingRecoveryPromotion hint, partition state stays MIRRORING (not FAILED), replication continues in ASYNC mode, and the auto-promoter then re-attempts a recovery Register when lag reaches zero
• Source ULE Handling (sync mode): Trigger ULE on the source with unclean.leader.election.enable=true; verify destination detects divergence via mirrorLeaderEpoch, truncates to the divergence point, and re-syncs from the new source leader — same as async behavior
• Source ULE Blocked: With unclean.leader.election.enable=false on the source, simulate the conditions that would cause ULE; verify the source partition becomes unavailable, sync gating naturally pauses (no source leader = no acks), and the operator can failover to the destination as new primary via the standard stop-mirror flow
• CLI Source-Compatibility Probe — Compatible: Run --alter --sync against a sync-supporting source; verify the broker call is made and DesiredMirrorType=SYNC is set
• CLI Source-Compatibility Probe — Incompatible: Run --alter --sync against a source with mirror.sync.version disabled; verify the CLI fails with a clear error, exits non-zero, and DesiredMirrorType was not changed at the broker
• CLI Source-Compatibility Probe — Unreachable: Run --alter --sync while the source is partitioned away; verify the CLI emits a warning and the broker call still proceeds. Then bring the source back up incompatible; verify the auto-promoter eventually attempts Register and is rejected with MIRROR_SYNC_VERSION_MISMATCH, the topic stays at effective ASYNC, and SyncRegistrationRejected{reason=VERSION_MISMATCH} increments
• Registration Propagation Race: Inject artificial metadata-propagation lag on a source partition leader during Register commit; verify the leader returns MIRROR_REGISTRATION_PROPAGATING on the destination's Fetch for that partition; verify destination treats it as per-partition (increments SyncPartitionFetchFailures{partition=N, reason=PROPAGATING}, retries that partition, leaves topic effective at SYNC, other partitions continue normally); verify the next Fetch succeeds once the leader applies the record and the partition resumes gating
• Single-Partition Stale Leader Doesn't Demote Topic: Force one source partition leader to stay badly behind on metadata fetches such that it returns INVALID_SYNC_REGISTRATION for its partition; verify (a) the destination's topic effective mirrorType stays SYNC, (b) SyncPartitionFetchFailures{partition=N} increments, (c) other partitions of the topic continue gating normally, (d) once the stale leader catches up, the failing partition resumes normal sync gating without any demote/promote cycle
• All-Partitions-Failing Triggers Topic Demote: Release the topic from source (all leaders apply removal); verify after mirror.sync.topic.demote.timeout.ms the destination demotes the topic to ASYNC, sets pendingRecoveryPromotion, and auto-recovers via the standard path

System Tests

System tests will validate behavior under realistic production conditions:

• Performance Benchmark: Measure replication throughput and latency across WAN; verify MirrorSyncReplicationLatencyMs distribution under representative load
• Scalability Test: Replicate 1,000 topics with 100,000 partitions across clusters in sync mode
• Long-Running Stability: Run continuous replication for 7 days, verify no memory leaks or performance degradation
• Sync Failover and Failback: Simulate source crash mid-sync with destination caught up; verify destination retains all acked records on failover (zero-RPO); reverse-mirror to bring the recovered source back as the new destination; verify failback into sync uses auto-promotion when lag reaches zero
• Source-Side Release Under Load: Sustained produce workload across a sync mirror; trigger --release-sync mid-stream; verify source unblocks producers immediately, all destination partitions begin receiving INVALID_SYNC_REGISTRATION, SyncPartitionsFailing{mirror,topic} rises to the full partition count, and after mirror.sync.topic.demote.timeout.ms the destination demotes the topic to ASYNC and auto-recovers via the standard path without data loss
• Mixed-Mode Transaction Failover: Run a transactional producer writing to topic A (SYNC) and topic B (ASYNC) in the same transaction; throttle B's async replication so its COMMIT marker hasn't reached the destination at source-crash time; force source crash mid-transaction; verify A's destination shows the transaction as committed (marker present) while B's destination shows it as aborted (marker absent). This documents the cross-topic consistency hazard described in Feature Improvements > Transactional Producer — the test serves as documentation of the expected behavior, not a passing-vs-failing condition

Rejected Alternatives

Rejected Alternatives to the Whole KIP

Application-Level Synchronous Replication

Leave synchronous cross-cluster replication to be handled at the application level, where applications manage their own dual-write logic or use caching/buffering to replay data to destination clusters.

Why Rejected:

• Adds significant complexity to application code
• No standardized solution that fits all use cases
• Difficult to guarantee consistency and exactly-once semantics
• Each application team would need to reinvent the wheel
• Error handling and retry logic becomes application responsibility
• Broker-level solution provides better performance and reliability guarantees

Promote Stretch Clusters for Zero-RPO in DR Situations

Recommend the use of stretched Kafka clusters (single logical cluster spanning multiple data centers) instead of separate clusters with mirroring.

Why Rejected:

• Single failure domain — no isolation from software, configuration, or operational incidents. A stretched cluster is one logical cluster: a bad broker upgrade, a corrupted KRaft metadata log, a misapplied ACL change, a bad config rollout, a cascading failure, an inter-DC network partition that splits the controller quorum, or a runaway tenant propagates to every data center simultaneously, with no independent failover target to fall back to. There is no separate KRaft controller quorum, no separate metadata log — nothing to insulate failures, and no way to canary an upgrade in one DC for a release cycle before rolling it to the other. DR with two separate logical clusters exists primarily to protect against exactly these classes of incidents, which are the dominant cause of data-platform outages in practice — well above raw hardware failure or full-DC loss. With separate clusters plus sync mirroring, operators get the best of both worlds: zero data loss on failover and genuine fault isolation between primary and DR.
• Cross-DC RTT becomes both a per-produce cost and an operational ceiling. Stretched clusters place replicas in different DCs, so any topic with acks=all and min.insync.replicas requiring cross-DC quorum blocks every produce on cross-DC RTT — regardless of whether that topic actually needs cross-DC durability. The same constraint imposes a hard operational ceiling on daily cluster work: partition rebalances, admin-client metadata traversals across nodes, consumer-group coordinator interactions, and similar control-plane operations all incur cross-DC round-trips and become correspondingly slow or unstable. Sync mirroring surfaces WAN-scale latencies to producers as longer ack times on the opted-in topics rather than as cluster instability or as limits on cluster operations; it is opt-in per topic, and works across regions and providers.
• Vendor reality. Because of these operational limitations, no managed Kafka service offers stretched clusters as a managed product — operators who want this topology run it self-hosted at their own risk. Even in the self-hosted positioning, stretched clusters are framed as an HA mitigation for unstable hardware within a single cluster, not as a DR architecture. The operational requirements — dedicated low-latency interconnect, controller-quorum placement across DCs, ISR-aware client routing, careful tenant isolation — make them difficult to operate at scale, and the recovery semantics fundamentally do not match what operators expect from a DR setup.

Rejected Alternatives to Part of the KIP Decisions

Source Polling Destination for ISR Status

Instead of extending the FetchRequest to include remote replica information, have the source cluster actively poll the destination cluster to check if it has reached ISR for a given offset before acknowledging the producer.

Why Rejected:

• Introduces a new request/response pattern instead of leveraging existing Kafka protocols
• Less efficient than piggybacking on existing FetchRequest that already flows from destination to source
• The existing producer acknowledgment mechanism already delays acks until all ISR replicas fetch the required offset; extending this to include remote replicas is a natural fit
• In KIP-1279, the leader on the destination is already treated as a replica from the source's perspective, so extending the existing replication path is clearer and more consistent
• Polling would add unnecessary network overhead and complexity
• Harder to reason about timing and consistency with an additional RPC pattern

Sync/Async Selection in Producer Client API 

Allow producers to specify on a per-message basis whether to use synchronous or asynchronous mirroring (via configs or headers).

Why Rejected:

• Would require significant changes to the producer API and force producers to upgrade their clients to achieve this functionality
• If configuration is not consistent across all topic producers, it could lead to confusing behavior where some messages in a partition have different guarantees
• Topic-level configuration is simpler and sufficient for most use cases

Automatic Degradation to Async Mode

Automatically downgrade from sync to async mode when destination cluster latency exceeds a threshold, without manual intervention.

Why Rejected:

• Silent degradation could violate zero-RPO guarantees without operator awareness
• May cause unexpected data loss in disaster scenarios. Operators should make explicit decisions about when to accept data loss risk
• Difficult to determine appropriate thresholds that work for all use cases
• Manual control provides clearer operational semantics
• However this can be a follow-up improvement later to this KIP

Replica Fetch with Synthetic Replica ID for Sync Mirror

Use FetchRequest.forReplica with a synthetic negative replica ID (e.g., -3) for sync mirror fetches, so the source cluster treats the destination as a follower replica and returns data up to LEO.

Why Rejected:

• The synthetic replica ID would create a fake entry in the source's remoteReplicasMap, which could confuse ISR shrink/expand logic, partition reassignment, and leader election
• Multiple destination brokers fetching with the same synthetic ID (or different synthetic IDs) adds complexity to the source's replica tracking
• Risk of unintended side effects in metrics, monitoring, and operational tooling that assumes replica IDs correspond to actual brokers
• The source would need special-case handling in followerReplicaOrThrow and other validation paths to accept the synthetic ID
• Cleaner alternative: use consumer-style fetch (replicaId=-1) and modify the source's read path to return data up to LEO when MirrorReplicaInfo is present in the FetchRequest. This keeps mirror fetches clearly separated from real replica fetches while still allowing the destination to catch up to LEO. The MirrorReplicaInfo becomes the single mechanism for both data flow (read up to LEO) and HW gating (destination ISR state), avoiding any pollution of the source's replica tracking

Mirror-Level ULE Pause Policy

Introduce a mirror-level configuration (e.g., mirror.sync.unclean.leader.election.policy with values PAUSE and FOLLOW) that controls what happens on the destination when the source experiences an unclean leader election. The PAUSE value (intended as default) would have the destination retain its authoritative data (potentially ahead of the new source leader), pause sync mirroring, and require the operator to decide between failing over, demoting to ASYNC, or releasing the sync link. The FOLLOW value would silently truncate to match the new source leader.

Why Rejected:

• Re-invents the existing source-side unclean.leader.election.enable=false knob, which already prevents the data loss by blocking the unclean election entirely. Operators already have this primitive available without adding a new config surface
• The PAUSE option's recovery path (operator chooses between failover-to-destination, demote-to-ASYNC, or source-side release) is already achievable today: blocking the unclean election on the source makes the partition unavailable, sync gating naturally pauses, and the operator can failover to the destination as new primary and later reverse-mirror to fail back (see the Unclean Leader Election on Source Cluster section above)
• PAUSE-state semantics add substantial spec surface for a rare edge case: a new partition sub-state, paused-partition metric, operator decision tree, interaction with the source-side release flow, and recovery RPCs
• Mirror-level naming would risk colliding with KIP-1279's existing mirror.support.unclean.leader.election topic config (which controls LME-truncation behavior on the destination), creating two configs with similar names but different semantics on the same resource
• For V1, deferring this keeps sync and async modes consistent on source-ULE handling and limits the operational decision tree to existing Kafka primitives
• This can be revisited in a follow-up KIP if the community demands explicit mirror-level handling beyond what the existing source-side unclean.leader.election.enable=false provides

Unify Register/Unregister into a Single RPC

A single MirrorSyncRegistrationOpRequest with an Operation enum (REGISTER/UNREGISTER) could replace the separate RegisterMirrorSyncTopicRequest and UnregisterMirrorSyncTopicRequest, saving one apiKey.

Why Rejected:

• Separate RPCs make operation intent explicit at the protocol level — audit logs and wire traces show what was attempted without parsing an Operation discriminator
• Register carries fields specific to admission control (RecoveryPromotion) that are meaningless on Unregister; merging them would make the unified schema harder to reason about
• The apiKey conservation argument is minor; the protocol-key space is not constrained for mirror-related RPCs
• Separate RPCs allow independent schema evolution in the future

Operator-Driven Recovery from Source-Side Release

Require an explicit operator action on the destination (--alter --sync after a release) to clear a destination-local "released-by-source" suppression flag and trigger re-promotion, rather than letting the destination's auto-promoter retry recovery automatically once lag reaches zero.

Why Rejected:

• The two cases for source-side release have natural division of responsibility: destination unreachable from source is the only case --release-sync is for (the source operator is the only operator who can act); destination reachable but slow is handled by the destination operator running --alter --async. Once a destination becomes reachable again, the original justification for the source-side release no longer applies, and there is no operator intent to keep sync off
• Operator-driven recovery would require a destination-side operator to manually clear suppression after every transient unreachability event — adding toil for a case the system can resolve on its own
• The marker's defensive purpose (rejecting stale in-flight Registers from before release) is preserved either way: a stale Register carries RecoveryPromotion=false and is rejected; only a fresh recovery Register issued by the destination's auto-promoter (which the destination only fires after observing lag-zero) carries RecoveryPromotion=true and clears the marker
• A destination operator who explicitly doesn't want auto-recovery can opt out by demoting the topic via --alter --async, which is the same primitive used for the reachable-but-slow case