Hi, Fede, Thanks for the reply. A few more comments.
JR26. The mirror partition states table link is not clickable. JR28. Hmm, I am a bit confused. start/stop/pause etc changes the state at the mirror/topic level, but mirror coordinator manages the mirror partition state. Which broker does the validation? Is it the mirror coordinator or the MirrorMetadataManager? JR31.1 Hmm, if only topics are specified in StartMirrorTopicsRequest and those topics don't exist in the source yet, topics are not persisted anywhere, right? Then how can the MirrorMetadataManager picks up those topics when they are created in the source? JR31.3 OK. So MirrorMetadataManager serves two purposes: (1) mirrors topic/group/acl metadata from the source (2) initiates truncation/mirroring on mirror partitions. The instance responsible for (1) is the one based on mirror name hashed into the _mirror_state partition and the inststance responsible for (2) is the one on the broker that is the mirror partition leader. This also means that MirrorMetadataManager needs to run on every broker, consumes every _mirror_state partition and caches every mirror partition state. Is that correct? JR30. "During failback, the requesting cluster sends its own cluster ID, and the receiving cluster finds the matching mirror by scanning which of its mirror configs has that cluster ID as source." Does that mean that there could only be 1 mirror between a pair of source and destination cluster? Also, when ClusterId is set in DescribeClusterMirrors, should the MirrorNames field be empty? "Then it does a simple LME lookup in its metadata log using the original mirror name." Does the requesting cluster use the FindCoordinator RPC to identify the ClusterMirrorCoordinator storing the LME for the mirror? If so, should we bump up the version of FindCoordinator to support the new KeyType? JR36. Hmm, how does the follower set the MirrorLeaderEpoch field initially and subsequently? Also, when will the fetched batch epoch ever pass the expected value in the follower? Our replication flow uses a similar logic based on leader epoch because the batches obtained from the leader are not necessariy committed. In cluster mirroring, only committed records are mirrored. JR40. "When a state is in FAILED state, we’ll retry it in exponential backoff time to re-transition to the PreviousState to retry the state." Hmm, if we are trying to re-transition to the previous state, why allowing the previous state to transition to the error state in the first place? JR44. Could you add the description to the flow? JR45. Is there any benefit of introducing this new config instead of just reusing sasl.mechanism.inter.broker.protocol? JR47. Workflow: JR47.1 For each of the initial requests, could you describe whether it lands on an arbitrary broker or a coordinator? If it's the latter, how is the coordinator determined? JR47.2 It seems that RPCs like WriteMirrorStates and ReadMirrorStates should be used in the workflow, but they are not listed. JR48. StartMirrorTopics: Why does it need the NumPartitions field since all partitions in the topic will be mirrored? JR49. "Also it writes a MirrorTopicStateChangeRecord with empty mirror name. " An empty mirror name is not very intuitive. Should we use a null value instead? JR50. Configs: Could you add the type for each new config? Jun On Wed, Jul 15, 2026 at 2:00 AM Federico Valeri <[email protected]> wrote: > Hi Jun, thanks for the additional and useful feedback. > > JR20: Yes. MirrorMetadataManager sends a ListGroups RPC to the > destination cluster filtering by active states (STABLE, > PREPARING_REBALANCE, COMPLETING_REBALANCE, ASSIGNING, RECONCILING). > Groups in those active states are skipped with a warning log message > during offset sync. Groups in EMPTY or DEAD are implicitly allowed > (not returned by the filter). KIP updated. > > JR21: No, mirrored records skip timestamp validation like how a > follower does right now. Timestamp validation is skipped for mirrored > records to preserve source timestamps. We basically rely on the source > leader validation. KIP updated. > > JR22: Yes, they are correct, let me explain. The phrase "unchanged > during replication" means per-batch appends don't bump, while the > invariant is maintained by periodic/proactive bumps via a separate > BumpLeaderEpochs controller RPC (see Leader Epoch Invariant > paragraph). KIP updated. > > JR23: Destination. The mirror.support.unclean.leader.election config > controls truncation behavior on the destination cluster. This prevents > a destination ULE from electing a replica that has stale > pre-truncation data. The source cluster's ULE behavior is not affected > by this config. KIP updated. > > JR24: Agreed. KIP updated to remove the KIP-1327 paragraph. > > JR25: Exclude takes precedence. In the config section we have “Topics > matching the exclude pattern are not mirrored even if they match > mirror.topics.include. Internal topics are always excluded. Exclude > always wins over include.” and similar for groups exclude patterns. > KIP and protocol descriptions are updated to make it more visible. > > JR26: Ok. I created a mirror partition states table and linked where > it is needed. > > JR27: Good catch. KIP updated. > > JR28: It is used for an optimistic locking guard that prevents a > state-change operation (start, stop, pause, resume, delete) from going > through if a mirror's topic state changed between the moment the > broker did its validation (preconditions) and the moment the > controller processes the operation. > > Mirror partition state transitions involve two separate nodes: a > broker and the controller. The broker validates preconditions against > the mirror coordinator, then sends a request to the controller, which > writes the actual metadata records. There is a window between the > broker's check and the controller's write where another operation can > sneak in and change the state, making the original operation invalid. > > The StateValidationOffset field contains the cluster metadata offset > before any validation. It represents the metadata snapshot against > which the broker made its decision. If the controller detects that the > MirrorTopicStateChangeRecord offset is strictly greater than > StateValidationOffset, it rejects the operation with > InvalidClusterMirrorStatesException. > > I added a ClusterMirrorCoordinator sub-paragraph to explain this. > > JR29: No RPC, the KIP sentence is slightly misleading. KIP updated. > > Each broker only reports partitions it is responsible for (i.e. it is > the mirror leader for that partition). The ReplicaManager is just a > pass-through that delegates to the MirroFetcherManager. The actual > source of lag data is the MirrorFetcherThread, which continuously > updates lag as it replicates from the source cluster. The admin client > fans out the DescribeClusterMirrors request to all brokers to get the > full picture. > > JR30: Cluster B does not match by mirror name. It matches by source > cluster ID. The mirror config stores the source cluster ID at creation > time, which is then used for validation and incremental mirroring. > During failback, the requesting cluster sends its own cluster ID, and > the receiving cluster finds the matching mirror by scanning which of > its mirror configs has that cluster ID as source. Then it does a > simple LME lookup in its metadata log using the original mirror name. > > I added a ClusterMirrorCoordinator sub-paragraph to explain this. > > JR31.1: Topics are already persisted, but not as ConfigRecord entries. > When startMirrorTopics succeeds, a MirrorTopicStateChangeRecord is > written to the metadata log for each topic that exists at that point. > Topics that don't exist yet are picked up later by the periodic auto > discovery loop, which matches source topics against the persisted > include/exclude patterns and issues a StartMirrorTopics for any new > matches. In short, the patterns are the durable config records of > “what should be mirrored”. Individual topics are the resolved result. > > JR31.2: Yes, the include/exclude topic pattern configs are persisted > and then the topics will be created if not exist. The KIP is updated > to avoid any confusion. > > JR31.3: Every broker has its own MirrorMetadataManager instance that > receives the same metadata update, but each instance only processes > partitions for which that broker is the partition leader. Truncation > and state transitions are performed locally by the leader broker. The > same applies to stop: each broker's MMM handles the stopping process > only for its own leader partitions. > > JR32: Yes, a stopped topic's mirror association persists in snapshots > indefinitely until someone deletes the mirror. Once you delete the > mirror, you lose the ability to leverage delta failback, but you can > still do a failback from scratch. > > JR33.1/JR34: MirrorTopicStateChangeRecord (metadata log) is not > tombstoned. It is cleared by writing a new record with an empty mirror > name. The effect is that the topic's mirror association is removed > from the TopicImage. When the next metadata snapshot is written, the > snapshot writer skips topics with a blank mirror name. The KIP is > updated. > > The records in __mirror_state (partition state and LME) are the ones > that get tombstoned, and they do have keys (MirrorPartitionStateKey > and LastMirrorEpochsKey). > > JR34.1: Sure. KIP Updated. > > JR34.2: When cluster A does a failback and sends > DescribeClusterMirrors to its source (cluster B), it sets clusterId to > A's own cluster ID. Cluster B then uses that to find which of its > mirror configs has A as the source, as we discussed earlier. > > JR35: In a regular topic, if a topic is deleted and one of the replica > brokers is offline, the topic deletion completes. Later when the topic > re-creates with the same name, it’ll be assigned with a different > topic ID. So when the fenced broker becomes online, it can detect the > topic ID change and make the existing data folder as “stray”. On the > other hand, a mirror topic is deleted and re-created with “the same > source topic ID” while a broker is offline, that broker still has log > directories tagged with the old topic ID. When it comes back online, > it cannot distinguish between the stale log directory from the > previous incarnation and the active one from the new creation. It > would treat the stale directory as valid, causing data corruption or > inconsistency. > > JR36: Your understanding is correct. The MirrorLeaderEpoch is not for > fetching from the source cluster. It is for destination internal > replication, where follower brokers fetch from the mirror leader. > > The MirrorLeaderEpoch field in the Fetch request tells the follower > "this request was sent when the mirror epoch was N." The follower > uses that epoch (not the regular leader epoch) to validate the fetched > batches before appending them. If the fetched batch epoch has passed > what the follower expected, the batches are rejected. > > JR37: Good catch. They are not necessary. It was related to the old > design. KIP updated. > > JR38: Metadata synchronization operates at the mirror level rather > than the partition level, so it uses a separate coordinator assignment > based on the mirror name alone. That is: > > leader of partition [Utils.abs(mirrorName.hashCode()) % partitionNum] > in __mirror_state topic > > Only the broker assigned as the metadata coordinator for a given > mirror performs synchronization, and it applies changes only to the > mirror partitions it manages. This avoids both redundant > synchronization across brokers and unnecessary updates to partitions > managed by other coordinators. > > JR39: TopicId is not needed for StopMirrorTopics. The admin client > does not populate it, and the controller resolves the topic by name. > The field appears in the schema but is unused. We removed it now, as > well as the topicId field in PauseMirrorTopics and ResumeMirrorTopics > requests. > > JR40: PreviousState is used for the FAILED retry mechanism. When a > state is in FAILED state, we’ll retry it in exponential backoff time > to re-transition to the PreviousState to retry the state. So the data > partition leader will get this info from the coordinator and then > apply the retry mechanism. We updated the KIP to explain it. > > JR41: As you suggest, we added a dedicated LeaderEpoch field to > PartitionChangeRecord (version 3+). The controller computes the final > epoch value and writes it into the record. All nodes just read it from > PartitionRegistration.merge() when replay the PartitionChangeRecord, > with no local derivation needed. For backward compatibility with older > metadata versions (v0-v2), merge() falls back to the previous implicit > derivation (leader change = epoch + 1) when the field is not set and > leader is changed. KIP updated. > > JR42: Indeed, we should be consistent in all places to use topicId > instead of topicName. We updated to use topicID in > MirrorPartitionStateValue and LastMirrorEpochsValue. KIP is updated. > > JR43: Yes, we support all the sasl.* and ssl.* configurations for > authentication with the source cluster. We’ve added 2 placeholder > entries into the mirror config table. > > JR44: When a data partition leader changes, the onMetadataUpdate > callback in the new leader node will be invoked and then it’ll get the > current mirror partition state from the mirror coordinator, either > read from local log, or sending ReadMirrorStates API to the > coordinator node and wait for the response async. After the state is > retrieved, it’ll process the current partition state tasks. Ex: > MIRRORING state is to start a mirror fetcher thread to fetch from the > source cluster. > > When a coordinator partition leader changes, the leader will load all > the logs locally and start to do what a coordinator should do. > > JR45: This configuration is used for the admin client in the > destination cluster (MirrorMetadataManager) to talk to the brokers and > controllers in the same cluster. For example, when syncing up the > consumer group offsets, the admin client needs to talk to the group > coordinator in the destination cluster to commit/alter the offset. > > JR46: This information is in the “Early Access” sub-paragraph of the > “Release Phases” paragraph, but I created an introduction to define > them properly. > > Thanks > Fede >
