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
