Hi Luke, Federico, Omnia, and the whole team, Before anything else - thank you. We have been iterating on this KIP for months now, and I deeply appreciate how thoroughly and seriously you have engaged with every architectural and security concern raised throughout this review process. Digging into each edge case, updating the design specification, and opening the tracking JIRAs has ensured a robust design for a change of this scale.
*VK12 (Data-Safety Clamp): Confirmed Closed.*I see the defensive clamp fully integrated into both the MirrorMetadataManager section and the Group Offsets section: max(dest log start offset, min(dest LEO, source committed offset)) The inclusion of the worked execution example completely resolves my data-safety concerns regarding offset boundaries. Thank you for folding this critical validation layer into the core design. *VK10 (Group-Namespace Validation & KAFKA-20799): Resolved.*Your reasoning is structurally sound, and I verified the behavior against the core implementation to be certain. Because group names are unique across consumer/share/streams types by design (with GroupMetadataManager throwing a GroupIdNotFoundException for cross-type access) and synced offset commits route through the standard OffsetCommit path, an isolated pre-flight gate in mirroring would be incomplete. The correct structural fix belongs at the group-namespace level in core Kafka. I am completely aligned with dropping the localized gate framework; KAFKA-20799 is precisely the right architectural home for this ecosystem fix. To assist with the implementation of KAFKA-20799, here is how I view the priority structure for production readiness (non-blocking for this KIP): *Actionable Error Telemetry:* Enhance the exception payload to explicitly include the colliding group name, its type in each cluster, and the associated mirror name, rather than suppressing it behind a bare GroupIdNotFoundException. This transforms an ambiguous failover-time skip into a self-explaining, actionable diagnostic event. *Operational Metrics: *Introduce a dedicated metric counter tracking group-offset-sync collisions/errors so platform operators can alert on collisions deterministically instead of parsing distributed logs. *On-Demand Diagnostics (Lowest Priority):* Expose an on-demand describe/list capability to surface point-in-time name collisions, which is highly valuable for identifying pre-existing conflicts during initial environment setup. Item (1) delivers the primary architectural value; the remaining items can land according to your core roadmap. I would be glad to serve as the reviewer for the KAFKA-20799 PR once it is ready for code review. With VK12 verified and VK10 properly decoupled and tracked in KAFKA-20799, I have no remaining architectural concerns. You have my full endorsement on KIP-1279. This is a highly significant advancement for Kafka's distributed mirroring capabilities, executed with excellent rigor. Outstanding work by the entire team. Regards, Vaquar Khan https://www.linkedin.com/in/vaquar-khan-b695577/ On Mon, 13 Jul 2026 at 05:26, Luke Chen <[email protected]> wrote: > Hi Vaquar, > > Thanks for your comments. > > VK12: > Sorry, it's my bad that I forgot to update in the discussion thread and the > KIP. > Yes, we adopted your suggestion but forgot to update the KIP. > I've just updated it now. Sorry for the confusion. > > VK10: > "(a) returning a precise error for cross-type collisions" > As you know, this is an existing logic, not related to this KIP. > I've opened KAFKA-20799 <https://issues.apache.org/jira/browse/KAFKA-20799> > to > improve the error handling for group name collisions. > > "(b) adding a pre-flight validation check to the CLI utility" > I'm wondering how much it helps because the group name is not fixed during > the mirroring period. > Even if we did a validation before starting the mirror, it is still > possible the newly created consumer group name, either in source or > destination cluster, will have a collision. Also, the > "mirror.groups.include" > is a dynamic configuration. Users can update it anytime. > For example: source cluster has 0 consumer/share group, destination cluster > has 1 "share group" called: "test" > 1. "mirror.groups.include=.*": Users start the mirror, and later create a > consumer group "test" in the source cluster, which collides with the one in > the destination cluster when sync up. > 2. "mirror.groups.include=dummy": Users start the mirror, and later create > a consumer group "test" in the source cluster. No group offset sync up > happens. But later, the "mirror.groups.include" config is updated to > "mirror.groups.include=test", > so the name collision still happens. > > For case (1), when mirror starts, we don't know if the source cluster will > create a classic consumer group named "test" in the future. There is > nothing we can do. > For case (2), even if we add a validation each time the > "mirror.groups.include/exclude" > config updates. That still doesn't resolve this issue because we don't have > a way to distinguish an offset commit is from mirror sync up or from user > creation. For example, "mirror.groups.include=.*" changes to > "mirror.groups.include=test". > When the change triggers, should we treat the existing "test" group in the > destination cluster as "name collision" or we should pass the validation > because the "test" could be created and synced in the mirror? We have no > way to distinguish it. > > My thought is to improve the option (a) in KAFKA-20799 > <https://issues.apache.org/jira/browse/KAFKA-20799> and make the error log > clear to users. > What do you think? > > Thank you, > Luke > > On Mon, Jul 13, 2026 at 7:54 AM vaquar khan <[email protected]> wrote: > > > Hi Federico, Luke, Omnia, and all, > > > > Thanks for the continued work on this. The KIP has come a long way. I > want > > to close out the items we have settled so the record is clean, then > address > > the remaining data-safety item where the current text does not yet > > implement the necessary architectural guard. > > > > > > *RESOLVED / CLOSED ON MY END* > > - VK2 (Poison pill blast radius): Resolved. Partition-level FAILED > > isolation in the fetcher pool is the correct model. > > > > - VK5 (Infinite loops): Resolved. Read-only enforcement on mirrored > > topics makes A->B->A physically impossible. > > > > - VK6 (Unclean-election divergence): Accepted as a documented > > limitation, and the updated ULE sub-design thoroughly addresses my > initial > > concerns. > > > > - VK7 (Tiered storage): Accepted for the initial release; the follow-up > > plan is reasonable. > > > > - VK1 (Thundering herd / OOM): Resolved. Bounding peak memory > > by num_fetcher_threads * response_max_bytes, rather than partitions * > > 1MB, ensures predictable resource consumption. > > > > - VK3 (Control-plane hotspot): Resolved. Distributing > > __mirror_state transitions via the (mirrorName, topicId, partition) hash > > mirrors the proven scalability model of __consumer_offsets. > > > > - VK4 / VK8 (PID rewriting): Resolved, and superior to my original > > proposal. Dropping PID rewriting via the MIRROR_PID_RESET control record > > completely bypasses the hasProducerId() issue (where > > AbstractRecordBatch.hasProducerId() evaluates NO_PRODUCER_ID < > > producerId(), meaning a negative PID would have bypassed > > ProducerStateManager and incorrectly defaulted the LSO to the high > > watermark). > > > > - VK11 (TransactionIndex): Resolved. Confirmed the index is rebuilt > > locally on append rather than copied over the wire. > > > > - VK9 (CRC / zero-copy): Moot once PID rewriting was removed. > > > > Thank you for working through these eight architectural items. > > > > > > *STILL OPEN: VK12 (Offset-sync read-side data loss)* > > Luke noted on June 2 that the proposed formula makes sense, which I > > appreciate. However, the current Group Offsets text only implements the > > active-group guard ("skip groups that are active on the destination in > any > > state other than EMPTY or DEAD"). While this guard is necessary, it is > > insufficient to prevent data loss for INACTIVE or dead groups during > > replication lag. > > > > Verifying this against the core coordinator implementation reveals a > > critical vulnerability when operators follow the KIP's guidance to > > set auto.offset.reset=latest: > > > > 1. The synced offset for an inactive group can legitimately exceed the > > destination log end offset (LEO) due to replication lag. The > > OffsetCommit path in the group coordinator does not validate a committed > > offset against the partition LEO, meaning an out-of-range offset is > > persisted as-is. > > > > 2. Upon failover, a resuming consumer reading from this offset receives > an > > OFFSET_OUT_OF_RANGE exception. > > > > 3. Under the auto.offset.reset=latest strategy, the reset resolves via > > AutoOffsetResetStrategy.timestamp() to > > ListOffsetsRequest.LATEST_TIMESTAMP, which returns the current > destination > > LEO. > > > > 4. Consequently, the consumer skips every record that arrived between > > the stale committed position and the reset point. This results in > > silent, unrecoverable read-side data loss precisely during a cluster > > failover the exact scenario this KIP is designed to protect against. > > > > Because the affected group is inactive on the destination, the > > active-group guard offers no protection here. The only way to guarantee > > data safety is to enforce a structural invariant at sync time using a > > double-clamp: > > > > syncedOffset = max(destinationLogStartOffset, min(sourceCommitted, > > destinationLEO)) > > > > - The min(..., destinationLEO) upper bound eliminates the > > OFFSET_OUT_OF_RANGE exception window entirely, preventing forced resets > and > > data skipping. > > - The max(destinationLogStartOffset, ...) lower bound prevents > > group rewinding if an offset falls below the destination retention log > > start offset. > > > > Under this invariant, the worst-case scenario is bounded > > re-processing proportional to the replication lag at failover, which > aligns > > with Kafka's at-least-once delivery contract. This is strictly safer than > > endorsing a configuration path that silently drops data. > > > > Can we update the Group Offsets section to explicitly define this > > double-clamp as a coordinator-enforced invariant alongside the > active-group > > guard? > > > > > > *STILL OPEN: VK10 (Group-name namespace collision / JR18)* > > The current text specifies that an offset commit fails > > with GroupIdNotFoundException when a consumer group and a share group > share > > a name. > > As confirmed in the coordinator logic, share groups and consumer groups > > occupy a single namespace, and cross-type access throws this exception. > > This introduces two issues during operational failovers: > > > > 1. The error semantics are incorrect. The group ID does exist; it is > > simply the wrong type. Throwing "not found" misleads operators under > fire. > > I agree with Jun's point in JR18 that an exception like > > GroupIdInUseException is required. > > > > 2. Discovering name collisions during an emergency failover introduces > > extreme operational risk. Introducing a pre-migration validation check > to > > the CLI (e.g., kafka-cluster-mirrors.sh) to scan and flag source group > > and share-group collisions before mirroring begins would move this risk > > from production runtime to set up-time. > > > > Can we commit to (a) returning a precise error for cross-type > > collisions,and (b) adding a pre-flight validation check to the CLI > utility? > > > > Once the double-clamp invariant for VK12 is integrated into the design > text > > and we have a clear resolution for VK10, I have no further blockers and > > will gladly support the vote. > > > > Regards, > > Vaquar Khan > > https://www.linkedin.com/in/vaquar-khan-b695577/ > > > > On Sun, 12 Jul 2026 at 12:54, Jun Rao via dev <[email protected]> > > wrote: > > > > > Hi, Federico, > > > > > > Thanks for the updated KIP. A few more comments. > > > > > > JR20. "Groups that are active on the destination cluster (i.e., in any > > > state other than EMPTY or DEAD) are skipped to avoid regressing offsets > > > after failover." > > > Does MirrorMetadataManager need to send an RPC to the group coordinator > > to > > > know the group's state? > > > > > > JR21. Are mirrored records validated against > > > log.message.timestamp.before.max.ms and > > log.message.timestamp.after.max.ms > > > on the destination cluster? > > > > > > JR22. "Leader Epoch Bump (LEB): The leader epoch in the destination > > cluster > > > remains unchanged during replication. That means it is possible that > the > > > fetched batch carries a leader epoch that does not match with the local > > > leader epoch." > > > Is that correct? In the Leader Epoch Invariant section, there is logic > to > > > makes sure SLE > DLE can't happen. > > > > > > JR23. "A new dynamic topic level configuration > > > mirror.support.unclean.leader.election (boolean, default false) > controls > > > whether LME truncation waits for all assigned replicas to converge, not > > > just current ISR members. When enabled, the leader will not complete > > > truncation until every replica has joined the ISR and truncated past > the > > > LME offset. This prevents subsequent unclean leader elections from > > > introducing undetectable log divergence between clusters." > > > Is this to prevent issues with ULE on the source or destination? > > > > > > JR24. "KIP-1327 introduces immutable creation timestamps on both > > partitions > > > and consumer groups to establish a happen-before relationship that > > > determines which offset reset policy applies when a consumer group > > > encounters a partition with no committed offset." > > > This needs adjustment since we decided not to pursue KIP-1327. > > > > > > JR25. IncludePatterns/ExcludePatterns: Which one takes precedence if > both > > > are specified? > > > > > > JR26. For each record and RPC involving a state field, could you add a > > > description of the all possible states? > > > > > > JR27. WriteMirrorStatesRequest > > > { "name": "", "type": "int32", "versions": "0+", "default": > "-1", > > > "about": "The last mirror leader epoch, or -1 if not > > available." > > > }, > > > The field name is empty. > > > > > > JR28. Could you describe how the StateValidationOffset field is being > > used? > > > > > > JR29. "The ReplicaManager which provides source offset, destination > > offset, > > > and lag for each partition." > > > Which RPC is used to query the ReplicaManager? > > > > > > JR30. Failover: The original destination b stores LME under the mirror > > name > > > a-to-b. During fallback, the original source a creates a new mirror > name > > > b-to-a and uses it in DescribeClusterMirrors to obtain LME. How can the > > > original destination cluster b find the right LME if the stored and > > > requested mirror names don't match? > > > > > > JR31. Start Mirror Topics: > > > JR31.1 "Controller persists include/exclude patterns as ConfigRecord > > > entries on the MIRROR resource in the metadata log." > > > Should we persist topics too? It's possible that the source topic > doesn't > > > exist yet when a mirror is started. > > > JR31.2 "For each topic, the controller creates it on the destination if > > it > > > does not already exist, updates include/exclude topic pattern configs" > > > What are include/exclude topic pattern configs? In the previous step, > the > > > controller already persists include/exclude patterns as ConfigRecord > > > entries on the MIRROR resource in the metadata log. > > > JR31.3 "Brokers receive the metadata update. The MirrorMetadataManager > > > detects the new mirrorName and DesiredState from the TopicImage and > > queries > > > the coordinator for the current partition state. > > > Partitions transition from UNKNOWN to LOG_TRUNCATION. During this > state, > > > LME truncation runs and waits for all ISR members (or all replicas if > ULE > > > is enabled)." > > > The MirrorMetadataManager instance responsible for a mirror is not > > > necessarily the same broker that leads a mirror partition. How can the > > > MirrorMetadataManager instance perform truncation? Ditto for Stop > Mirror > > > Topics. > > > > > > JR32. Stop Mirror Topics: "For each topic, the controller writes a > > > MirrorTopicStateChangeRecord with DesiredState=1 (stop)." > > > Is this record kept forever in the metadata snapshot? > > > > > > JR33. Delete Mirror > > > JR33.1 he controller tombstones => The controller tombstones > > > JR34. "The mirror state records in __mirror_state internal topic are > also > > > tombstoned." > > > Could you describe how MirrorTopicStateChangeRecord is tombstoned since > > it > > > doesn't have a key? > > > > > > JR34. AdminClient: > > > JR34.1 Could you define the public facing classes such as > > > CreateClusterMirrorResult, StartMirrorTopicsResult, > > > PauseMirrorTopicsOptions, etc? > > > JR34.2 describeClusterMirrors: How does it populate the clusterId field > > in > > > DescribeClusterMirrorsRequest? > > > > > > JR35. CreateTopic : "The replicas for the partition assignment are all > > > active and not in fenced or controlled shutdown. This is to make sure > > when > > > a topic gets deleted and re-created with the same topic ID, the stale > > > offline log dir won’t be treated as the active log dir after it becomes > > > online" > > > Could you explain why this is only needed for mirror topics, but not > > > regular topics? > > > > > > JR36. Fetch: Could you explain why we need to add the MirrorLeaderEpoch > > > field? I thought the destination cluster just issues a regular fetch > > > request like a consumer. > > > > > > JR37. CreateClusterMirror: "The broker validates that the mirror name > is > > > not already in use, contains only permitted characters, and does not > end > > > with .stopped or .paused suffix." > > > Could you explain why the suffix restriction is necessary? > > > > > > JR38. MirrorMetadataManager: Could you explain how mirrors are assigned > > to > > > MirrorMetadataManager instances? Is it based on the _mrror_state topic > > > partitions? > > > > > > JR39. StopMirrorTopics: Do we need TopicId? How does the AdminClient > get > > > it? > > > > > > JR40. ReadMirrorStatesResponse: Could you explain how PreviousState is > > > being used? > > > > > > JR41. PartitionChangeRecord.MinLeaderEpoch: Currently, leader epochs > are > > > implicitly derived from PartitionChangeRecord. By adding this field, > the > > > controller will sometimes derive the epoch and sometimes use the > > explicitly > > > stored epoch, which can be confusing. Would it be better to always > store > > > the leader epoch explicitly in PartitionChangeRecord? > > > > > > JR42. MirrorPartitionStateValue: Why does it use topicName, not > topicId? > > > > > > JR43. Mirror config: What about authentication (sasl, ssl, etc) related > > > configs? > > > > > > JR44. Could you also describe how a partition leader picks up the > > mirroring > > > flow when it changes? > > > > > > JR45. Broker config: sasl.mechanism.mirror.admin.protocol > > > Is that for mirror-related RPCs between brokers and controllers? > > > > > > JR46. Compatibility > > > Since this KIP changes the metadata record format, we need an MV bump. > > For > > > mirror-related records and internal RPCs, should we introduce a > > > mirror-specific feature like GV/TV to gate them? > > > > > > Jun > > > > > > > > > On Tue, Jul 7, 2026 at 6:40 AM Federico Valeri <[email protected]> > > > wrote: > > > > > > > Hi Michael, thanks for your review. > > > > > > > > MM1: It was a bit confusing but the coordinator is the scheduler and > > > > the manager is the executor. That said, given that the manager > already > > > > has a scheduler, we moved that logic there and updated the KIP. > > > > > > > > MM2: Good point. A deleted source topic would now cause the mirror > > > > partitions on the destination cluster to move to FAILED state. You > can > > > > easily identify them using the describe command with the --failed > > > > flag. > > > > > > > > MM3: We have a log message at warn level. I think this is enough. > > > > > > > > MM4: Correct. KIP updated. > > > > > > > > MM5: Correct. KIP updated. > > > > > > > > MM6: The isolation level is hardcoded to READ_UNCOMMITTED. Not sure > > > > about the actual use case, but we can certainly make it configurable > > > > in a follow-up KIP. > > > > > > > > MM7: For quota related topic configurations, > > > > follower.replication.throttled.replicas and > > > > leader.replication.throttled.replicas are excluded by default in MM2 > > > > and we apply the same logic. mirror.support.unclean.leader.election > > > > and mirror.replication.throttled.replicas are included by default > > > > because they are mirror related configurations. > > > > > > > > MM8: Updated the field description to: Topic patterns to remove from > > > > mirror.topics.include or add to mirror.topics.exclude. > > > > > > > > MM9: Not needed. KIP updated. > > > > > > > > MM10: The config is needed because the listener can have multiple > > > > enabled SASL mechanisms and the listener name alone doesn't select > > > > one. The naming follows the existing > > > > sasl.mechanism.inter.broker.protocol convention, so renaming would be > > > > inconsistent with that precedent. > > > > > > > > MM11: Correct. KIP updated. > > > > > > > > MM12: No, when mirror.support.unclean.leader.election=true, we’ll > wait > > > > for all replicas, not just ISR replicas to complete truncation. > > > > > > > > MM13: Correct. KIP updated. > > > > > > > > MM14: We missed the “MIRROR_TOPIC_NOT_STOPPED”, so now: start on non > > > > stopped throws MIRROR_TOPIC_NOT_STOPPED. Stop on paused throws > > > > MIRROR_TOPIC_ALREADY_PAUSED. Pause on stopped throws > > > > MIRROR_TOPIC_BEING_STOPPED. Resume on non paused throws > > > > MIRROR_TOPIC_NOT_PAUSED. Stop on stopped throws > > > > TOPIC_NOT_IN_CLUSTER_MIRROR. > > > > > > > > > > > > > > > > On Fri, Jul 3, 2026 at 5:26 PM Mickael Maison < > > [email protected]> > > > > wrote: > > > > > > > > > > Hi, > > > > > > > > > > Thanks for the KIP! > > > > > This will definitively be a great addition to Kafka. > > > > > > > > > > MM1: In the description of the main components, > > > > > ClusterMirrorCoordinator has this responsibility: > > > > > "Metadata Refresh Scheduling: The coordinator schedules periodic > > > > > metadata refresh operations by invoking a metadata manager every 30 > > > > > seconds by default. This ensures that configuration updates and > group > > > > > offset commits in the source cluster are continuously propagated to > > > > > the destination cluster. The refresh interval is configurable via > > > > > mirror.metadata.refresh.interval.ms." > > > > > Is that accurate because MirrorMetadataManager has pretty much the > > > > > same responsibility. > > > > > > > > > > MM2: About topic deletions, the KIP states: "the operator would > need > > > > > to manually remove the topic from the mirror" > > > > > Is there a way to find topics that have been deleted in the source > > > > > cluster for an operator to do that action? > > > > > > > > > > MM3: About group offsets: It states: "Groups with active members on > > > > > the destination are skipped." > > > > > Should these be logged too? > > > > > > > > > > MM4: In the security control section, the DeleteClusterMirror RPC > > > > > requires Alter on the ClusterMirror resource. Should we use Delete > > > > > instead of Alter? > > > > > On non-cluster resources, we typically use Delete for deletion. See > > > > > > > > > > > > > > > https://urldefense.com/v3/__https://kafka.apache.org/43/security/authorization-and-acls/*operations-and-resources-on-protocols__;Iw!!Ayb5sqE7!tihx6ElDOI77AtiHliFlPPSZqZJwpnkf9ZzYWGySbnmOir-2Y0uKuum9iDrYrvt-HCTR3A8oaxqU3AKr-48$ > > > > > > > > > > MM5: In Destination cluster permissions, I think FindCoordinator > > > > > should also need Describe on the Group resources for regular > consumer > > > > > groups. > > > > > > > > > > MM6: In Exactly-Once Semantics, it states: "The mirror fetcher > thread > > > > > uses READ_UNCOMMITTED isolation" > > > > > Could this be configurable? If so we can do this in a follow-up KIP > > > > > > > > > > MM7: It seems quotas are not mirrored between the clusters? > > > > > Is that omission intentional? > > > > > > > > > > MM8: In StopMirrorTopicsRequest I'm not sure I understand what the > > > > > Patterns field is? > > > > > Is it a pattern for the topics to stop? If so the description seems > > odd > > > > > > > > > > MM9: In > > > > > > > > > > StopMirrorTopicsResponse/StartMirrorTopicsResponse/PauseMirrorTopicsResponse/ResumeMirrorTopicsResponse > > > > > do we really need the mirror name? > > > > > > > > > > MM10: In Configuration / Broker: > > > > > Can you clarify why we need sasl.mechanism.mirror.admin.protocol? > > > > > Can't we retrieve the protocol from mirror.admin.listener.name or > > > > > inter.broker.listener.name? > > > > > If we really need it, should it be named > > > > > mirror.admin.sasl.mechanism.protocol to match the other names? > > > > > > > > > > MM11: In Configuration / Mirror > > > > > Should it be mirror.acls.include instead of mirror.acl.include to > > > > > match mirror.groups.include and mirror.topics.include? > > > > > > > > > > MM12: Is the description of mirror.support.unclean.leader.election > > > > inverted? > > > > > > > > > > MM13: The description of CLUSTER_MIRROR_NOT_EMPTY states: "The > > cluster > > > > > mirror still has active or non-removed topics" > > > > > Should it be "The cluster mirror still has active or non-stopped > > > topics"? > > > > > > > > > > MM14: We have the MIRROR_TOPIC_BEING_STOPPED error when stopping > > topic > > > > > already in stopping state? What happens when pausing topics already > > in > > > > > pausing state? > > > > > Related, I assume when stopping a stopped topic, you get > > > > > TOPIC_NOT_IN_CLUSTER_MIRROR > > > > > > > > > > Thanks, > > > > > Mickael > > > > > > > > > > On Mon, Jun 22, 2026 at 7:47 PM Andrew Schofield < > > > [email protected]> > > > > wrote: > > > > > > > > > > > > Hi Fede, > > > > > > Thanks for your replies. > > > > > > > > > > > > AS30: Yes, that's right. Two different end markers for the same > > > > transaction, and essentially unbounded duration of the inconsistency > > > since > > > > KIP-939. I agree that it's a class of transaction state divergence, > > like > > > > KAFKA-20716. For KIP-1279, the most important thing is that consumers > > of > > > > the mirrored topic must not get confused by the markers. For example, > > > when > > > > the Fetch response contains the list of aborted transactions, > provided > > we > > > > are sure that it can be built properly without confusion from the > extra > > > > marker, then we are good. It came to mind because the way this is > > > > represented in the protocol for the responses is somewhat fragile I > > feel, > > > > so I'm a bit paranoid that a mis-alignment could break it. > > > > > > > > > > > > Thanks, > > > > > > Andrew > > > > > > > > > > > > On 2026/06/19 12:16:52 Federico Valeri wrote: > > > > > > > Hi Andrew, me, again. > > > > > > > > > > > > > > Given that READ_COMMITTED consumers are unaffected because the > > > ABORT > > > > > > > is already recorded in the transaction index and the dangling > > > COMMIT > > > > > > > produces no CompletedTxn (no index and LSO updates), the > > practical > > > > > > > impact is limited. However, __transaction_state records the > > > > > > > transaction as COMMITTED while the partition log has it as > > aborted. > > > > > > > While analyzing this, we discovered that this type of > divergence > > is > > > > > > > not unique to mirroring. KAFKA-20716 documents the same > category > > of > > > > > > > inconsistency after unclean leader elections, where lost > > > transaction > > > > > > > markers cause the LSO to get stuck (much bigger problem). We > > > believe > > > > > > > that solving the transaction state divergence issue would also > > > close > > > > > > > this mirroring scenario. Wdyt? > > > > > > > > > > > > > > On Thu, Jun 18, 2026 at 10:36 PM Federico Valeri < > > > > [email protected]> wrote: > > > > > > > > > > > > > > > > Hi Andrew, about AS30, I had a closer look at KIP-939, and I > > > think > > > > > > > > this is what you mean: > > > > > > > > > > > > > > > > The problem is that a WriteTxnMarker request for a PID > unknown > > to > > > > the > > > > > > > > ProducerStateManager (cleared by MIRROR_PID_RESET) does not > > fail. > > > > The > > > > > > > > end transaction marker is still appended, but transaction > index > > > and > > > > > > > > LSO are unaffected (ProducerStateManager.prepareUpdate, > > > > > > > > ProducerAppendInfo.appendEndTxnMarker). This may lead to a > > > > situation > > > > > > > > where we have two different end txn markers for the same > > > > transaction: > > > > > > > > ABORT from the stop mirroring operation, followed by a COMMIT > > > from > > > > the > > > > > > > > local Txn Coordinator retrying its WriteTxnMarker after the > > > > partition > > > > > > > > becomes writable again. > > > > > > > > > > > > > > > > READ_COMMITTED consumers still see the earlier ABORT in the > > > > > > > > transaction index and skip the aborted data. The visible > > outcome > > > > for > > > > > > > > consumers is correct. However, __transaction_state on the > > > original > > > > > > > > source cluster now records the transaction as COMMITTED while > > the > > > > > > > > partition log has it as aborted, creating a silent > > inconsistency > > > > that > > > > > > > > tools querying transaction state will not detect. The window > > for > > > > this > > > > > > > > race is bounded by transaction.max.timeout.ms (default 15 > > > > minutes), > > > > > > > > but becomes unbounded with KIP-939 because it allows prepared > > > > > > > > transactions to sit indefinitely. > > > > > > > > > > > > > > > > We could move the coordinator state to “complete abort” at > the > > > > cost of > > > > > > > > an RPC. This would prevent any retry, but that would break > the > > > 2PC > > > > > > > > contract (Kafka must hold the prepared state until the > external > > > > > > > > coordinator decides). Also, before stop there would be a > number > > > of > > > > > > > > retries constantly going on. We definitely need a better > > > solution. > > > > > > > > > > > > > > > > On Thu, Jun 18, 2026 at 11:36 AM Federico Valeri < > > > > [email protected]> wrote: > > > > > > > > > > > > > > > > > > Hi Andrew, thanks for your careful review. > > > > > > > > > > > > > > > > > > AS28: We reworked that section and added some diagrams to > > > > illustrate > > > > > > > > > problem and solutions. Let us know if it is better. > > > > > > > > > > > > > > > > > > AS30: We updated that table, but we are not sure about the > > > > problem you > > > > > > > > > mention. Can you clarify and send us the sequence of > actions > > > that > > > > > > > > > would lead to that problem? > > > > > > > > > > > > > > > > > > AS32: You are right. Replaced > > > > StartMirrorTopicsRequestData.TopicData > > > > > > > > > with a Controller.MirrorTopicMetadata record. Also renamed > > > > TopicData > > > > > > > > > to TopicMetadata in all schemas. > > > > > > > > > > > > > > > > > > AS33: Good idea. Added per-topic error messages in the RPC > > > > responses > > > > > > > > > you mentioned. > > > > > > > > > > > > > > > > > > AS29, AS31, AS34 (both of them), AS35, AS36: These small > > issues > > > > are > > > > > > > > > all fixed now. > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > On Tue, Jun 9, 2026 at 9:58 PM Andrew Schofield < > > > > [email protected]> wrote: > > > > > > > > > > > > > > > > > > > > Hi Fede and friends, > > > > > > > > > > This KIP is shaping up nicely. The change to the way that > > > > state management for the mirrors is done is much more elegant. The > > > removal > > > > of the PID mapping is also a nice improvement. > > > > > > > > > > > > > > > > > > > > Inevitably, I have some more comments. > > > > > > > > > > > > > > > > > > > > AS28: I understand KAFKA-18723 and how leader epochs work > > in > > > > that case, but the enhancements that you made to the Fetch RPC and > how > > > they > > > > work escapes me. Please could we have some diagrams to illustrate the > > > > various epochs? > > > > > > > > > > > > > > > > > > > > AS29: I like the addition of share group offset syncing. > > You > > > > need AlterShareGroupOffsets and DescribeShareGroupOffsets in the > tables > > > of > > > > RPCs and permissions. > > > > > > > > > > > > > > > > > > > > AS30: It is certainly true that EOS is not supported. > Using > > > > cluster mirroring on transactional data does risk breaking > > transactional > > > > atomicity for transactions in flight at the time that mirroring was > > > > stopped. That's a known limitation, and it's an effect of per-topic > > async > > > > replication. Writing ABORT markers when stop mirroring is triggered > is > > > > interesting. I just wonder whether it's truly safe. > > > > > > > > > > > > > > > > > > > > Could you enhance the tables of records in the > Exactly-Once > > > > Semantics section to include leader epochs? I think that the leader > > epoch > > > > for the ABORT markers will have been bumped as the mirroring stopped > > and > > > > the partitions became writable. However, I worry that someone might > > > switch > > > > the mirroring back in the other direction before the transaction > which > > > was > > > > rudely partially aborted has completed on the original source > cluster, > > > and > > > > that the effect of the transaction coordinator completing the > > > transaction, > > > > potentially writing duplicate transaction markers after the ABORT > > markers > > > > which will have been replicated back again. > > > > > > > > > > > > > > > > > > > > This is probably only a realistic possibility when > KIP-939 > > is > > > > present, but it is accepted and I'm sure it will land soon enough. I > > > would > > > > say this KIP should be prepared for the existence of long-running > > > > transactions so we don't dig a hole to fall into later. > > > > > > > > > > > > > > > > > > > > AS31: In the section on the Admin Client, the description > > for > > > > Admin.createClusterMirror(String, Map, CreateClusterMirrorOptions) is > > > > duplicated. > > > > > > > > > > > > > > > > > > > > AS32: The use of StartMirrorTopicsRequestData.TopicData > in > > an > > > > external interface looks odd. > > > > > > > > > > > > > > > > > > > > AS33: I suggest per-topic error messages in the RPC > > responses > > > > such as PauseMirrorTopics so you can pass back more context, such as > > the > > > > state if the request failed due to a state mismatch. > > > > > > > > > > > > > > > > > > > > AS34: There is an unnamed field in > > > > DescribeClusterMirrorsResponse. The same in LastMirrorEpochsValue. > > > > > > > > > > > > > > > > > > > > AS34: Some of the fields in BumpLeaderEpochsRequest start > > > with > > > > lower-case letters. Also, there are quite a few fields in the RPCs in > > > > general that should be version "0+" which are "0". > > > > > > > > > > > > > > > > > > > > AS35: Sometimes the partition states in the RPCs are > > string, > > > > but sometimes int8. > > > > > > > > > > > > > > > > > > > > AS36: The base level for share group support is probably > > 4.1. > > > > All of the RPCs were present in 4.1, although they were only enabled > by > > > > default in 4.2. > > > > > > > > > > > > > > > > > > > > Thanks, > > > > > > > > > > Andrew > > > > > > > > > > > > > > > > > > > > On 2026/06/04 11:01:36 Federico Valeri wrote: > > > > > > > > > > > Hello, additional updates and replies: > > > > > > > > > > > > > > > > > > > > > > VK12: Consumer data loss risk is a good point. We > updated > > > > the "Group > > > > > > > > > > > Offsets" paragraph. > > > > > > > > > > > > > > > > > > > > > > Some of you raised the point that using mirror.name > > > > suffixes for pause > > > > > > > > > > > and stop operations is a bit inelegant and we agreed. > Now > > > we > > > > propose a > > > > > > > > > > > better approach: > > > > > > > > > > > > > > > > > > > > > > We replaced the topic config based approach ( > mirror.name > > > and > > > > > > > > > > > .stopped/.paused suffix conventions) with a first class > > > > metadata > > > > > > > > > > > record for tracking mirror topic state changes. A new > > > > > > > > > > > MirrorTopicStateChangeRecord is added to the metadata > log > > > > with three > > > > > > > > > > > fields: TopicId (uuid), MirrorName (string), and > > > > DesiredState (int8, > > > > > > > > > > > where 0 = start mirroring, 1 = stop, 2 = pause). When a > > > user > > > > calls > > > > > > > > > > > startMirrorTopics, stopMirrorTopics, or > > pauseMirrorTopics, > > > > the > > > > > > > > > > > controller writes this record to the metadata log > instead > > > of > > > > > > > > > > > manipulating config suffixes. Brokers receive the > record > > > via > > > > metadata > > > > > > > > > > > updates and react accordingly, triggering partition > level > > > > state > > > > > > > > > > > transitions through the existing MirrorPartitionState > > state > > > > machine. > > > > > > > > > > > More details in the updated KIP. > > > > > > > > > > > > > > > > > > > > > > Let us know what you think. > > > > > > > > > > > > > > > > > > > > > > Thanks > > > > > > > > > > > Fede > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > On Tue, Jun 2, 2026 at 8:50 AM Luke Chen < > > > [email protected]> > > > > wrote: > > > > > > > > > > > > > > > > > > > > > > > > Hi Viquar, > > > > > > > > > > > > > > > > > > > > > > > > Thanks for the comment. > > > > > > > > > > > > > > > > > > > > > > > > Regarding VK12, thanks for raising the issue. > > > > > > > > > > > > We didn't think about that. > > > > > > > > > > > > I agree that compared with data re-processing, it's > > worse > > > > to have data loss. > > > > > > > > > > > > The proposed formula makes sense to me. > > > > > > > > > > > > *syncedOffset = max(destinationLogStartOffset, > > > > min(sourceCommitted, > > > > > > > > > > > > destinationLEO))* > > > > > > > > > > > > > > > > > > > > > > > > Let us have some discussion internally and then > update > > > the > > > > KIP and reply to > > > > > > > > > > > > the thread. > > > > > > > > > > > > > > > > > > > > > > > > Thanks, > > > > > > > > > > > > Luke > > > > > > > > > > > > > > > > > > > > > > > > On Wed, May 20, 2026 at 2:39 AM Federico Valeri < > > > > [email protected]> > > > > > > > > > > > > wrote: > > > > > > > > > > > > > > > > > > > > > > > > > Hi Rajini, > > > > > > > > > > > > > > > > > > > > > > > > > > RS14: Done. Missed that, sorry. > > > > > > > > > > > > > > > > > > > > > > > > > > Thanks again. > > > > > > > > > > > > > > > > > > > > > > > > > > On Tue, May 19, 2026 at 7:28 PM Rajini Sivaram < > > > > [email protected]> > > > > > > > > > > > > > wrote: > > > > > > > > > > > > > > > > > > > > > > > > > > > > Hi Federico, > > > > > > > > > > > > > > > > > > > > > > > > > > > > Thanks for the updates. Just one minor point, > apart > > > > from that, looks > > > > > > > > > > > > > good. > > > > > > > > > > > > > > > > > > > > > > > > > > > > RS14: KIP still shows "bin/kafka-configs.sh > > > > --bootstrap-server :9094 > > > > > > > > > > > > > > --entity-type mirrors". > > > > > > > > > > > > > > Will be good to change that to `--entity-type > > > > cluster-mirrors`. > > > > > > > > > > > > > > > > > > > > > > > > > > > > Thanks, > > > > > > > > > > > > > > > > > > > > > > > > > > > > Rajini > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > On Tue, May 19, 2026 at 5:28 PM vaquar khan < > > > > [email protected]> > > > > > > > > > > > > > wrote: > > > > > > > > > > > > > > > > > > > > > > > > > > > > > Hi Federico and team, > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > Thank you for your detailed response on May 11, > > > > 2026. I greatly > > > > > > > > > > > > > appreciate > > > > > > > > > > > > > > > the collaborative effort over the past few > months > > > to > > > > harden the > > > > > > > > > > > > > KIP-1279 > > > > > > > > > > > > > > > architecture. > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > * 1. Resolved Architectural & Stability > > > > Vulnerabilities* > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > I am pleased to confirm that the KIP has > > > > successfully integrated fixes > > > > > > > > > > > > > for > > > > > > > > > > > > > > > the critical vulnerabilities I flagged, thereby > > > > protecting the > > > > > > > > > > > > > cluster's > > > > > > > > > > > > > > > state machine, memory bounds, and control > plane. > > > > Consider the following > > > > > > > > > > > > > > > items fully resolved on my end: > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > - > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > *VK4 / VK8 (Negative PID Bug & State Machine > > > > Failure):* I previously > > > > > > > > > > > > > > > identified that the proposed > > -(sourceProducerId > > > + > > > > 2) PID rewriting > > > > > > > > > > > > > > > formula would fundamentally break > > > hasProducerId() > > > > in > > > > > > > > > > > > > > > AbstractRecordBatch.java, causing the broker > > to > > > > bypass > > > > > > > > > > > > > > > ProducerStateManager.update() and default > the > > > > Last Stable Offset > > > > > > > > > > > > > (LSO) > > > > > > > > > > > > > > > to the High Watermark. I am glad to see the > > KIP > > > > authors abandoned > > > > > > > > > > > > > PID > > > > > > > > > > > > > > > mapping entirely and adopted the > deterministic > > > > MIRROR_PID_RESET > > > > > > > > > > > > > control > > > > > > > > > > > > > > > record barrier. This perfectly protects > > Kafka's > > > > exactly-once > > > > > > > > > > > > > semantics. > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > - > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > *VK11 (TransactionIndex Rebuild Ambiguity):* > > > > Thank you for > > > > > > > > > > > > > officially > > > > > > > > > > > > > > > confirming that the TransactionIndex is > > strictly > > > > rebuilt locally > > > > > > > > > > > > > during > > > > > > > > > > > > > > > log appends rather than copied > byte-for-byte. > > > > This resolves my > > > > > > > > > > > > > concern > > > > > > > > > > > > > > > regarding PID mismatches inducing consumer > > read > > > > anomalies. > > > > > > > > > > > > > > > - > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > *VK1 (Thundering Herd / OOM Heap > Allocation):* > > > > Your clarification > > > > > > > > > > > > > that > > > > > > > > > > > > > > > fetcher threads multiplex partitions meaning > > the > > > > memory footprint is > > > > > > > > > > > > > > > strictly bounded by num_fetcher_threads * > > > > response_max_bytes rather > > > > > > > > > > > > > than > > > > > > > > > > > > > > > a concurrent 1MB buffer per partition fully > > > > resolves my concern > > > > > > > > > > > > > > > regarding > > > > > > > > > > > > > > > 50GB broker-wide OOM spikes during mass > > > partition > > > > wake-ups. > > > > > > > > > > > > > > > - > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > *VK3 (Control Plane Hotspots):* I had > flagged > > > the > > > > severe risk of > > > > > > > > > > > > > > > metadata saturation on a single broker > during > > a > > > > "link flap" event. > > > > > > > > > > > > > Your > > > > > > > > > > > > > > > confirmation that the __mirror_state topic > > > > utilizes a compound hash > > > > > > > > > > > > > > > of (mirrorName, > > > > > > > > > > > > > > > topicId, partition number) mathematically > > > ensures > > > > that the 50,000+ > > > > > > > > > > > > > state > > > > > > > > > > > > > > > transitions will be safely distributed > across > > > the > > > > cluster, > > > > > > > > > > > > > neutralizing > > > > > > > > > > > > > > > the > > > > > > > > > > > > > > > single-node hotspot risk. > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > 2. Outstanding Critical Blocker: VK12 (Offset > > Sync > > > > Data Loss) > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > Regarding VK12, there is a fundamental > > > > misunderstanding in your > > > > > > > > > > > > > previous > > > > > > > > > > > > > > > reply. You stated: *"The scenario described can > > > only > > > > occur if offsets > > > > > > > > > > > > > are > > > > > > > > > > > > > > > force-written to an active group, which the > > design > > > > prevents."* > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > My concern has absolutely nothing to do with > > > > overwriting active > > > > > > > > > > > > > groups. My > > > > > > > > > > > > > > > concern applies strictly to the normal > > > > synchronization of inactive/dead > > > > > > > > > > > > > > > groups, and is based directly on the race > > condition > > > > currently > > > > > > > > > > > > > documented in > > > > > > > > > > > > > > > the official KIP-1279 text: > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > *"During offset synchronization, the committed > > > > offset in the > > > > > > > > > > > > > destination > > > > > > > > > > > > > > > cluster may temporarily exceed the current log > > end > > > > offset (LEO) of the > > > > > > > > > > > > > > > mirror topic... consumers attempting to resume > > from > > > > offset 100 will > > > > > > > > > > > > > receive > > > > > > > > > > > > > > > an OffsetOutOfRangeException. To handle this > > > > gracefully, consumers > > > > > > > > > > > > > should > > > > > > > > > > > > > > > configure auto.offset.reset=latest..."* > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > If a failover happens during this documented > > > > divergence window, a > > > > > > > > > > > > > > > reconnecting consumer will hit the > > > > OffsetOutOfRangeException. If the > > > > > > > > > > > > > > > downstream consumer follows the KIP's official > > > > advice and relies on > > > > > > > > > > > > > > > auto.offset.reset=latest, the consumer will > jump > > to > > > > the absolute newest > > > > > > > > > > > > > > > offset on the partition. *This completely skips > > any > > > > newly produced > > > > > > > > > > > > > records > > > > > > > > > > > > > > > that arrived between the failover and the > > consumer > > > > reconnecting, > > > > > > > > > > > > > resulting > > > > > > > > > > > > > > > in silent, unrecoverable data loss for the > > > > downstream application.* > > > > > > > > > > > > > > > *Proposed Resolution: Double-Clamped Offset > > Safety > > > > Invariant* Instead > > > > > > > > > > > > > of > > > > > > > > > > > > > > > requiring consumers to use > > auto.offset.reset=latest > > > > and endorsing a > > > > > > > > > > > > > known > > > > > > > > > > > > > > > data loss vector, I propose that the > > > > ClusterMirrorCoordinator enforce a > > > > > > > > > > > > > > > pre-persist offset validation invariant during > > > > offset synchronization. > > > > > > > > > > > > > > > Before any translated offset is committed to > the > > > > destination cluster, > > > > > > > > > > > > > the > > > > > > > > > > > > > > > coordinator must apply a double-clamp: > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > *syncedOffset = max(destinationLSO, > > > > min(sourceCommitted, > > > > > > > > > > > > > destinationLEO))* > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > Where destinationLSO is the Log Start Offset > > > > (earliest readable > > > > > > > > > > > > > position > > > > > > > > > > > > > > > post-retention) and destinationLEO is the Log > End > > > > Offset (latest > > > > > > > > > > > > > replicated > > > > > > > > > > > > > > > position) of the destination partition. > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > This guarantees that every persisted offset > falls > > > > within the physically > > > > > > > > > > > > > > > valid range eliminating both the > > > > OffsetOutOfRangeException caused by > > > > > > > > > > > > > > > exceeding the LEO during replication lag, and > the > > > > expired-offset rewind > > > > > > > > > > > > > > > caused by falling below the LSO due to > > destination > > > > retention policies. > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > The worst-case trade-off under this invariant > is > > > > bounded re-processing > > > > > > > > > > > > > > > proportional to the replication lag at failover > > > time > > > > not total > > > > > > > > > > > > > partition > > > > > > > > > > > > > > > depth which is perfectly consistent with > Kafka's > > > > documented > > > > > > > > > > > > > at-least-once > > > > > > > > > > > > > > > delivery guarantees. Broad enterprise > production > > > > evidence from > > > > > > > > > > > > > large-scale > > > > > > > > > > > > > > > cross-cluster failovers confirms that state > > checks > > > > alone (preventing > > > > > > > > > > > > > active > > > > > > > > > > > > > > > group overwrites) are insufficient; strict > offset > > > > bounds clamping is > > > > > > > > > > > > > > > required to achieve enterprise-grade data > > > integrity. > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > I look forward to your thoughts on implementing > > > this > > > > final offset > > > > > > > > > > > > > capping > > > > > > > > > > > > > > > logic. Once VK12 is patched, I believe this > > > > architecture will be > > > > > > > > > > > > > > > exceptionally robust and ready for enterprise > > > > deployment. > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > Best Regards, > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > Viquar Khan > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > On Mon, 18 May 2026 at 14:23, Rajini Sivaram < > > > > [email protected]> > > > > > > > > > > > > > > > wrote: > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > Hi Federico, > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > Thanks for the updates! The KIP is looking > > good. > > > A > > > > few more small > > > > > > > > > > > > > > > comments. > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > RS13: A couple of places still refer to > > > > `kafka-mirror.sh` like under > > > > > > > > > > > > > > > > `Failover Process`. Can we change them to > > > > > > > > > > > > > `*kafka-cluster-mirrors.sh*`? > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > RS14: Should we change `--entity-type > mirrors` > > > for > > > > kafka-configs to > > > > > > > > > > > > > be ` > > > > > > > > > > > > > > > > --entity-type *cluster-mirrors*` to be > > > consistent? > > > > Also, > > > > > > > > > > > > > > > > CLUSTER_MIRROR((byte) > > > > > > > > > > > > > > > > 64, "mirror"); could be `*cluster-mirror*`? > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > RS15: It may be useful to rename > > > > `mirror.topic.num.partitions` and ` > > > > > > > > > > > > > > > > mirror.topic.replication.factor` since they > are > > > > very similar to ` > > > > > > > > > > > > > > > > mirror.topic.properties.exclude`, but the > > > > `mirror.topic` prefix > > > > > > > > > > > > > refers to > > > > > > > > > > > > > > > > different topics (the internal topic for the > > > first > > > > two and actual > > > > > > > > > > > > > mirror > > > > > > > > > > > > > > > > topics for the other one). > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > RS16: ACL Sync: KIP says "Deletes ACLs that > > exist > > > > in destination but > > > > > > > > > > > > > not > > > > > > > > > > > > > > > in > > > > > > > > > > > > > > > > source using DeleteAcls request." > > > > > > > > > > > > > > > > What happens if someone creates an ACL on the > > > > destination to deny > > > > > > > > > > > > > > > > User:Alice access to all topics? > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > 1. If that ACL also existed on the source > > > > cluster and then it was > > > > > > > > > > > > > > > > removed, will it get removed from the > > > > destination? > > > > > > > > > > > > > > > > 2. If that ACL never existed on the source > > > > cluster, will it get > > > > > > > > > > > > > > > removed > > > > > > > > > > > > > > > > from the destination? > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > RS17: The table in the Source ACLs section > > says: > > > > > > > > > > > > > "DescribeClusterMirrors > > > > > > > > > > > > > > > > MC Read Cluster Log > truncation" > > > > > > > > > > > > > > > > Should that be `ClusterMirror:Read` instead > of > > > > `Cluster:Read`? > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > Regards, > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > Rajini > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > On Fri, May 15, 2026 at 8:41 AM Federico > > Valeri < > > > > > > > > > > > > > [email protected]> > > > > > > > > > > > > > > > > wrote: > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > Hi Rajini, we finally addressed the API and > > > tool > > > > naming > > > > > > > > > > > > > refactoring as > > > > > > > > > > > > > > > > > you suggested in RS6. Please take a look > when > > > > you have time. > > > > > > > > > > > > > Thanks. > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > On Mon, May 11, 2026 at 6:17 PM Federico > > > Valeri < > > > > > > > > > > > > > [email protected]> > > > > > > > > > > > > > > > > > wrote: > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > Hello all, I want to highlight a couple > of > > > new > > > > paragraphs: > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > 1. Leader Epoch Invariant: Cluster > > mirroring > > > > enforces the > > > > > > > > > > > > > invariant > > > > > > > > > > > > > > > > > > that the destination leader epoch must > > always > > > > be greater than or > > > > > > > > > > > > > > > equal > > > > > > > > > > > > > > > > > > to the source leader epoch (DLE>=SLE). > > > Without > > > > this, consumers > > > > > > > > > > > > > on the > > > > > > > > > > > > > > > > > > destination cluster can get stuck in an > > > > infinite metadata refresh > > > > > > > > > > > > > > > loop > > > > > > > > > > > > > > > > > > when they encounter committed offsets > > > carrying > > > > source epochs > > > > > > > > > > > > > higher > > > > > > > > > > > > > > > > > > than the local epoch. The invariant is > > > > maintained through three > > > > > > > > > > > > > > > > > > mechanisms: reactive bumping (epoch > fencing > > > > triggered when SLE > > > > > > > > > > > > > > DLE > > > > > > > > > > > > > > > > > > during fetch), proactive bumping > (scheduled > > > > when SLE approaches > > > > > > > > > > > > > DLE > > > > > > > > > > > > > > > > > > within a threshold), and periodic bumping > > > > (checked during > > > > > > > > > > > > > coordinator > > > > > > > > > > > > > > > > > > metadata sync). > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > https://urldefense.com/v3/__https://cwiki.apache.org/confluence/pages/viewpage.action?pageId=406620973*KIP1279:ClusterMirroring-LeaderEpochInvariant__;Iw!!Ayb5sqE7!tihx6ElDOI77AtiHliFlPPSZqZJwpnkf9ZzYWGySbnmOir-2Y0uKuum9iDrYrvt-HCTR3A8oaxqUcl_uWkU$ > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > 2. Group Offsets: The coordinator > > > periodically > > > > syncs consumer and > > > > > > > > > > > > > > > > > > share group offsets from the source > cluster > > > to > > > > the destination > > > > > > > > > > > > > for > > > > > > > > > > > > > > > all > > > > > > > > > > > > > > > > > > mirrored topics. Groups are filtered by > > > > configurable > > > > > > > > > > > > > include/exclude > > > > > > > > > > > > > > > > > > patterns, and offsets are only synced for > > > > groups that are not > > > > > > > > > > > > > > > > > > currently active on the destination > > cluster, > > > > preventing > > > > > > > > > > > > > overwrites of > > > > > > > > > > > > > > > > > > local consumer progress. Because source > and > > > > destination share the > > > > > > > > > > > > > > > same > > > > > > > > > > > > > > > > > > topic offsets (no offset translation), > > synced > > > > offsets can be used > > > > > > > > > > > > > > > > > > directly without mapping. > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > https://urldefense.com/v3/__https://cwiki.apache.org/confluence/pages/viewpage.action?pageId=406620973*KIP1279:ClusterMirroring-GroupOffsets__;Iw!!Ayb5sqE7!tihx6ElDOI77AtiHliFlPPSZqZJwpnkf9ZzYWGySbnmOir-2Y0uKuum9iDrYrvt-HCTR3A8oaxqU-JubnOA$ > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > These new paragraphs directly address > some > > of > > > > your questions, > > > > > > > > > > > > > but let > > > > > > > > > > > > > > > > > > me list them here: > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > JR2: Yes, we removed the incorrect phrase > > and > > > > added more details > > > > > > > > > > > > > to > > > > > > > > > > > > > > > > > > the paragraph. > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > JR4: When source cluster topic has tiered > > > > storage enabled, CM > > > > > > > > > > > > > works > > > > > > > > > > > > > > > by > > > > > > > > > > > > > > > > > > mirroring remote and local log into > > > > destination cluster. When > > > > > > > > > > > > > > > > > > destination cluster topic has tiered > > storage > > > > enabled, CM fails in > > > > > > > > > > > > > > > > > > PREPARING state because the LME may be in > > > > remote storage, but > > > > > > > > > > > > > works > > > > > > > > > > > > > > > > > > fine if already MIRRORING because no > > > > truncation is needed. > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > JR11: See "Leader Epoch Invariant" > > paragraph > > > > mentioned above. > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > JR13: We can't support stateful Streams > > > > application because > > > > > > > > > > > > > > > > > > asynchronous replication cannot preserve > > the > > > > transactional > > > > > > > > > > > > > boundaries > > > > > > > > > > > > > > > > > > between input offset commits, state store > > > > mutations written to > > > > > > > > > > > > > > > > > > changelog topics, and intermediate > records > > > > written to repartition > > > > > > > > > > > > > > > > > > topics. The synchronous extension of this > > > > design will be able to > > > > > > > > > > > > > > > > > > support them. Existing Features > Integration > > > > paragraph updated. > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > JR18: See "Group Offsets" paragraph > > mentioned > > > > above. > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > IY1: See "Group Offsets" paragraph > > mentioned > > > > above. > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > Thanks > > > > > > > > > > > > > > > > > > Fede > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > On Mon, May 11, 2026 at 6:08 PM Federico > > > > Valeri < > > > > > > > > > > > > > > > [email protected]> > > > > > > > > > > > > > > > > > wrote: > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > Hi Vaquar, > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > VK4/VK8: We don't do PID mapping > anymore. > > > > The KIP was updated > > > > > > > > > > > > > some > > > > > > > > > > > > > > > > > > > time ago with the new approach based on > > the > > > > new PID reset > > > > > > > > > > > > > control > > > > > > > > > > > > > > > > > > > record. > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > VK11: The transaction index is always > > built > > > > locally during log > > > > > > > > > > > > > > > > append, > > > > > > > > > > > > > > > > > > > never copied. > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > VK1: The 50,000 * 1MB = 50GB > calculation > > > > misunderstands the > > > > > > > > > > > > > fetch > > > > > > > > > > > > > > > > > > > model. Fetcher threads don't allocate > one > > > > buffer per partition. > > > > > > > > > > > > > > > > Actual > > > > > > > > > > > > > > > > > > > peak memory is roughly > > num_fetcher_threads > > > * > > > > > > > > > > > > > response_max_bytes, > > > > > > > > > > > > > > > not > > > > > > > > > > > > > > > > > > > num_partitions * partition_max_bytes. > > With > > > 1 > > > > fetcher thread > > > > > > > > > > > > > and the > > > > > > > > > > > > > > > > > > > default response max, the memory > > footprint > > > > is modest > > > > > > > > > > > > > regardless of > > > > > > > > > > > > > > > > > > > partition count. We are leveraging the > > same > > > > proven pattern > > > > > > > > > > > > > used by > > > > > > > > > > > > > > > > the > > > > > > > > > > > > > > > > > > > internal replication. > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > VK3: The __mirror_state topic uses > > > > hash-based partitioning > > > > > > > > > > > > > based on > > > > > > > > > > > > > > > > > > > mirrorName, topicId and partition > number. > > > > With the production > > > > > > > > > > > > > > > default > > > > > > > > > > > > > > > > > > > of 50 partitions, 50,000 partition > > > > transitions distribute > > > > > > > > > > > > > across > > > > > > > > > > > > > > > ~50 > > > > > > > > > > > > > > > > > > > partition leaders on different brokers, > > not > > > > a single broker. > > > > > > > > > > > > > This > > > > > > > > > > > > > > > is > > > > > > > > > > > > > > > > > > > the same proven pattern as > > > > __consumer_offsets, which handles > > > > > > > > > > > > > > > millions > > > > > > > > > > > > > > > > > > > of commits. > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > VK12: The scenario described can only > > occur > > > > if offsets are > > > > > > > > > > > > > > > > > > > force-written to an active group, which > > the > > > > design prevents. > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > Cheers > > > > > > > > > > > > > > > > > > > Fede > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > > >
