Hi Jun, we updated the KIP to reflect the following answers and hopefully clear most doubts. We also added a "Scalability Considerations" paragraph at the bottom.
On Thu, Jul 16, 2026 at 2:01 AM Jun Rao via dev <[email protected]> wrote: > > Hi, Fede, > > Thanks for the reply. A few more comments. > > JR26. The mirror partition states table link is not clickable. Ok. This should be fixed now. > 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? One correction first: start/stop/pause/resume changes the state at the mirror level, not at the topic level. The partition state machine is managed separately by the ClusterMirrorCoordinator. This is similar to an operator reconciling the desired state with the actual cluster state. The validation is done by MirrorMetadataManager, not by the coordinator directly. The request is sent to any broker, and that broker's MirrorMetadataManager validates that all affected partitions are in a valid state for the requested operation before forwarding to the controller. It does this in two steps: first it checks partitions whose __mirror_state records are coordinated locally (from its in-memory cache), then for partitions coordinated by other brokers, it sends ReadMirrorStates RPCs to the responsible coordinators. If all partitions pass, the broker stamps the current metadata offset on the request and forwards it to the controller. On the controller side, two more things happen. First, the controller checks that no mirror state changed between the broker validation and the controller processing, using the stamped metadata offset as an optimistic lock. If any topic's last state change offset exceeds that value, the entire request is rejected. Second, the controller validates each topic individually (ID consistency, whether the topic is already active in another mirror, whether it is in the expected state, etc.). At this level, individual topics can fail while others succeed, so the controller operates as a partial-success model. > 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? The topic names are actually persisted. When the controller processes a StartMirrorTopics request, it adds every topic (whether it’s literal or regex) to the "mirror.topics.include” configuration of that mirror before doing anything else. So even if the topic does not exist on the source yet and no MirrorTopicStateChangeRecord can be written for it, the topic name is recorded in the include pattern. >From that point on, MirrorMetadataManager picks it up during periodic sync. On every metadata refresh cycle, it sends a listTopics request to the source cluster, matches the results against mirror.topics.include (filtering out mirror.topics.exclude), and removes topics that are already being mirrored. Any remaining matches are new topics to start mirroring. It then issues a StartMirrorTopics request for those topics with the topic ID and partition count from the source, so the controller can create the destination topic and write the MirrorTopicStateChangeRecord. > 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? Yes, that is mostly correct, with one clarification. MirrorMetadataManager does run on every broker and it does serve those two purposes, but the scoping is slightly different from what you described. For purpose (1), metadata sync from the source, source topic state sync (leader caches, deletion detection, missed partition recovery) runs on every broker. Topic creation, partition scaling, and coordinator-level sync (topic configs, consumer group offsets, ACLs, topic discovery by pattern) run only on the broker that leads the __mirror_state partition determined by hashing the mirror name. For purpose (2), initiating state transitions (truncation, epoch fencing, creating fetchers), when a KRaft metadata update arrives, each broker collects only the partitions where it gained or holds leadership and triggers state transitions for those partitions. State writes are routed to the coordinator that leads the relevant __mirror_state partition for the composite key (mirror name, topic id, partition number). Regarding state caching: the mirror partition cache lives in MirrorMetadataManager. It is populated in two ways: by replaying __mirror_state partitions this broker leads, and by fetching state from remote coordinators via ReadMirrorStates RPCs. When this broker gains __mirror_state leadership, the ClusterMirrorCoordinator replays the log to rebuild the cache. When it loses leadership, the cache is cleared entirely. > 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? No, there can be multiple mirrors between the same pair of clusters. Each topic on the destination can only belong to one mirror at a time, but different topics can be mirrored through different mirror configurations even if they share the same source and destination. For example, topic "foo" on cluster B can be mirrored from cluster A via mirror "a-to-b", while topic "bar" on the same cluster B can be mirrored from the same cluster A via a separate mirror "a-to-b-2". Regarding the second part: when ClusterId is set in DescribeClusterMirrors, the MirrorNames field does not need to be empty. They serve independent purposes. MirrorNames controls which mirrors to describe (empty means all). ClusterId is a tagged field used specifically for last mirror epoch lookup during failback. When set, the receiving broker scans all its mirror configs to find mirrors whose source cluster ID matches the requesting cluster's ID, then looks up the stored last mirror epochs for the requested partitions. If multiple mirrors match (because there are multiple mirrors from the same source cluster), the broker aggregates the results by taking the maximum last mirror epoch per partition across all matching mirrors. So the two fields can be used together or independently. > Also, when ClusterId is set in > DescribeClusterMirrors, should the MirrorNames field be empty? In practice it would be natural to leave MirrorNames empty. But the protocol does not require it, and setting both fields would produce both outputs. We actually leave it empty on LME lookup, which means describe all mirrors, and we use the results to validate that the source partitions are in STOPPED state before proceeding with failback (mirror loop protection). This optimization spares one RPC request. > "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? No, the requesting cluster does not use FindCoordinator. It sends a DescribeClusterMirrors request, and the admin client fans out that request to all brokers on the target cluster. Each broker reports the last mirror epochs for partitions it coordinates locally, and the admin client aggregates the results by taking the maximum last mirror epoch per partition across all broker responses. The reason it broadcasts instead of targeting a specific coordinator is that the __mirror_state topic uses a composite key (mirror name, topic ID, partition number) to distribute coordination across its partitions. Two partitions of the same topic can be coordinated by different brokers, so there is no single coordinator that holds all the LME data for a given mirror or topic. Broadcasting to all brokers and merging the results is the simplest way to collect the complete picture. Since FindCoordinator is not involved, there is no need to bump its version or add a new KeyType. > If so, should we bump up the version of FindCoordinator to support the new > KeyType? We actually don’t need this anymore, so we reverted the change. Internally, we find the coordinator using the local metadata cache in the destination cluster. > 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? The follower initializes MirrorLeaderEpoch to 0 when it starts fetching a mirror partition. On each fetch response, the leader includes the latest leader epoch present in its log as the MirrorLeaderEpoch. The follower then uses its current MirrorLeaderEpoch (from the previous fetch cycle, not the one just received) to validate the fetched batches before appending them. Here is a concrete example. Suppose the destination leader has these records in its log: Offset 0: record A (batch epoch = 2) Offset 1: record B (batch epoch = 2) The follower sends a fetch request with fetchOffset=0 and MirrorLeaderEpoch=0. The leader responds with the data and MirrorLeaderEpoch=2 (the latest epoch in its log). Before appending, the follower validates the batches using the MirrorLeaderEpoch from its current state (0, not the 2 it just received). Since the batch epoch (2) is greater than the MirrorLeaderEpoch used for validation (0), the records are rejected. After processing completes, the follower updates its MirrorLeaderEpoch to 2 (from the fetch response) and retries. On the next fetch, the batch epoch (2) matches the MirrorLeaderEpoch (2), so validation passes and the records are appended. Regarding the second question: in normal replication, the leader and follower share the same epoch lineage, so the follower can validate batches against the current leader epoch directly. In cluster mirroring, the destination log contains batches carrying the source cluster's leader epochs, which are unrelated to the destination's own leader epoch. The follower cannot use the destination leader epoch for validation because it would not match the batch epochs at all. MirrorLeaderEpoch solves this by letting the leader tell each follower what the latest epoch in the log currently is, so the follower can validate that it is not receiving batches from a future epoch it has not seen yet. The initial fetch always rejects because the follower starts from 0 and needs one round trip to learn the actual epoch from the leader. > 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. It is correct. But the issue in KAFKA-18723 is about the “log truncation” and new records appended before fetch response sent. In the normal replication, the log truncation can only happen on “uncommitted data” like you said. But in an unclean leader election, even the committed data will be truncated. So our solution is to protect the unclean leader election in the source cluster causing inconsistent data in the destination cluster. About the case of unclean leader election causing inconsistent data, you can refer to the image in the KIP here: https://cwiki.apache.org/confluence/download/attachments/406620973/image-2026-7-14_11-47-39.png?version=1&modificationDate=1784022460000&api=v2 . It shows that due to the unclean leader election in the source cluster and a late fetch response, the follower (broker 1) might append the {E (LE: 1)} into offset 2 if no MirrorLeaderEpoch validation. > 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? Transitioning to FAILED makes the partition's actual status visible to operators. If a partition stayed in MIRRORING while backing off due to a connection issue or a transient error, it would look healthy from the outside, which could mask real problems. By moving to FAILED, the CLI tool and describe output can surface the error message, the retry attempt count, and the previous state, giving operators clear visibility into what went wrong and whether the system is recovering on its own. There is also a distinction between retriable and terminal failures. For transient errors (network issues, temporary unavailability), the coordinator retries with exponential backoff, transitioning back to the previous state automatically. But if the retry attempts are exhausted or the error is non-retriable (e.g. the source cluster ID changed, indicating a misconfiguration), the partition stays in FAILED permanently and requires manual intervention. Having an explicit FAILED state makes this distinction actionable: operators can use the CLI with the --failed flag to list partitions that need attention, and metrics can track retry behavior over time. > JR44. Could you add the description to the flow? Ok. I added a “Leadership Changes” sub-paragraph to MirrorClusterCoordinator and MirrorMetadata manager paragraphs. > JR45. Is there any benefit of introducing this new config instead of just > reusing sasl.mechanism.inter.broker.protocol? The goal is to separate the security principal used for mirroring from the one used for inter-broker communication. If we reused sasl.mechanism.inter.broker.protocol, users would be forced to grant additional permissions (GROUP, TOPIC READ/DESCRIBE, and others required by mirror metadata sync) to the inter-broker principal, which currently does not need them. This widens the blast radius of that principal unnecessarily. By introducing sasl.mechanism.mirror.admin.protocol, we keep the inter-broker principal scoped to its original responsibilities. The default value falls back to sasl.mechanism.inter.broker.protocol, so there is no extra configuration burden for users who are fine with a shared principal. It simply gives users the option to use a dedicated principal with narrower permissions for mirroring when their security model requires it. > 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? All mirror operations (start, stop, pause, resume, delete) are sent to an arbitrary broker. None of them target a coordinator directly. The receiving broker validates the partition states before forwarding the request to the controller. The validation works as follows, using pause as an example. When a PauseMirrorTopics request arrives at a broker, MirrorMetadataManager checks that every partition under the requested topics is in a state that is valid for the operation. For pause, all partitions must be in MIRRORING state. It first checks partitions whose __mirror_state records are coordinated locally (from its in-memory cache). For partitions coordinated by other brokers, it sends ReadMirrorStates RPCs to those brokers and waits for the responses. If any partition is not in the expected state, the entire request is rejected with an error. If all partitions pass, the broker stamps the current metadata offset on the request and forwards it to the controller. The controller then performs an optimistic locking check using that offset to guard against concurrent state changes, and if it passes, writes the MirrorTopicStateChangeRecord to the metadata log. The same two-phase validation pattern (local check, then remote coordinator RPCs, then forward to controller) applies to all mirror operations. The only difference is which partition states are considered valid: start requires STOPPED or UNKNOWN, stop requires MIRRORING or PAUSED, pause requires MIRRORING, and resume requires PAUSED. I updated the workflows paragraph to include this information. > JR47.2 It seems that RPCs like WriteMirrorStates and ReadMirrorStates > should be used in the workflow, but they are not listed. WriteMirrorStates and ReadMirrorStates are inter-broker RPCs used internally by MirrorMetadataManager, not user-facing operations, so they don't have their own workflow sections. But they play a role in several of the listed workflows and should be called out where they appear. Specifically, ReadMirrorStates shows up in the broker-side validation step of start, stop, pause, resume, and delete (when the broker needs partition state from a remote coordinator). WriteMirrorStates shows up when partition state changes need to be persisted to a __mirror_state partition coordinated by another broker. I updated the workflows paragraph to include this information. > JR48. StartMirrorTopics: Why does it need the NumPartitions field since all > partitions in the topic will be mirrored? The NumPartitions field is needed so the controller can create the destination topic if it does not already exist. When a topic is being mirrored for the first time, the destination cluster has no knowledge of the source topic's partition count, so the request must carry it. If the topic already exists on the destination, the NumPartitions field is ignored. The CLI populates this field automatically by describing the source topic before sending the request. > 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? Agreed. An empty string is not intuitive for "no mirror". Using null is clearer and aligns with Kafka's convention for nullable fields in wire protocol records. KIP updated. > JR50. Configs: Could you add the type for each new config? Sure. KIP updated. > Jun Thanks Fede
