Hi Jun, Thanks for the great comments again. Answering your question below.
> JR31.3 "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." Hmm, a given broker only caches the mirror partition state for partitions hashed into the __mirror_state partition hosted on this broker, right? Why does it need to fetch the state from other remote coordinators? > a given broker only caches the mirror partition state for partitions hashed into the __mirror_state partition hosted on this broker, right? Right. > Why does it need to fetch the state from other remote coordinators? Because the data partition leader needs to know the current mirror state of the partitions it leads. On a leadership change, the metadata delta is published to the MirrorMetadataManager, which queries the current mirror state for its led partitions from the coordinator (local or remote), and processes tasks based on that state. This is the reason why it fetches the state from other remote coordinator. This part of cache will be cleared after the leadership change. > JR36. I am still not quite sure why we need MirrorLeaderEpoch in the Fetch request. In the current Fetch request, in addition to CurrentLeaderEpoch, it has another field LastFetchedEpoch, which captures the last leader epoch in the data and is used to resolve the inconsistency between the follower and the leader. Does it achieve the same purpose as MirrorLeaderEpoch? Unfortunately, the LastFetchedEpoch cannot be used to do the validation to avoid the issue in KAFKA-18723. If we reject any fetched batches with epoch higher than the LastFetchedEpoch, it means the follower can never accept new appended batches with higher epoch in the leader node. Currently, the CurrentLeaderEpoch field serves 2 purposes in the fetch API: 1. The receiver (leader node) uses it to validate it matches the leader epoch in the leader node 2. The sender (follower node) uses it to validate the epoch in the fetched batches are all <= this value. For purpose (1), the CurrentLeaderEpoch is still serving for followers in the destination cluster. For purpose (2), my understanding of the validation for KAFKA-18723 is like: the follower knows the CurrentLeaderEpoch X is the highest epoch the leader node should contain in the log batches when fetch request is sent. Any log batch beyond this epoch in the fetch response means the follower’s metadata is stale. The batches need to be rejected and follower needs to update metadata for next fetch. In the Cluster Mirroring’s world, the MirrorLeaderEpoch serves this purpose to let the followers know the highest epoch in the leader node log in each fetch response. Once the info is stale in the follower side, it updates it from fetch response, and send next fetch request with correct MirrorLeaderEpoch to make sure no inconsistent data is possible like KAFKA-18723. I tried not to add new field in Fetch API, but this is the best solution I can think of. > JR48. "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." Hmm, this seems inconsistent. When a mirror is started, the source topic may not exist yet. When the source topic is created later, we can start mirroring without requiring the user to provide the number of partitions. > When the source topic is created later, we can start mirroring without requiring the user to provide the number of partitions. This is because this happens in periodical sync up in the destination cluster, and the numPartitions and other topic info in the source cluster is already retrieved during the sync up process. Also, if users are creating mirror topics via admin API, this info can also be retrieved inside the admin client if not provided. > JR51. Start Mirror: I am wondering if we truly need the last mirror epoch for truncation when starting a mirror. When a mirror is stopped at the source, we already bump up the leader epoch and write some additional records for cleanup. It seems that we can just use the DivergingEpoch returned in the initial mirror Fetch request to detect the diverging epoch and use that for truncation at the mirror destination. That’s a good question. I thought about this before but unfortunately it doesn’t work. The reason we need the last mirror epoch record is because the destination cluster might not completely sync with the source cluster before STOPPED. And this un-mirrored data might confuse the replication protocol when reverse mirror. For example, Cluster A (source) —> Cluster B (dest) foo-0 leader log foo-0 leader log Offset 0: from A (LE: 0) Offset 0: from A (LE: 0) Offset 1: from A (LE: 0) Offset 1: from A (LE: 0) Offset 2: from A (LE: 10) While cluster B mirrors until offset 1, the cluster A crashes and failover to cluster B. B bumps leader epoch to 10, appends a MIRROR_PID_RESET with LE: 10. Then cluster A starts to reverse mirror from cluster B, the log will be like this: Cluster A (dest) <-- Cluster B (source) foo-0 leader log: foo-0 leader log: Offset 0: from A (LE: 0) Offset 0: from A (LE: 0) Offset 1: from A (LE: 0) Offset 1: from A (LE: 0) Offset 2: from A (LE: 10) Offset 2: PID_RESET (LE: 10) Offset 3: from B (LE: 10) Offset 3: from B (LE: 10) As you can see, cluster A (new destination cluster) doesn’t know the offset 2 is not belonging to the cluster B, which causes inconsistent logs. So like what we described in the KIP, the last mirror epoch is the synchronization point between source and destination. Any data beyond the point is treated as unknown to the source cluster and should be truncated. > JR52. Stop Mirror: > JR52.1 "3. If patterns are provided, the controller removes matching entries from mirror.topics.include or adds them to mirror.topics.exclude on the CLUSTER_MIRROR resource." Why is it 'or' and not 'and'? Yes, that should be AND. KIP updated. Thanks. > JR52.2 "7. On subsequent metadata refresh cycles, MirrorMetadataManager discovers new source topics matching the persisted include/exclude patterns and repeats steps 3 through 7 for each." The broker will send a create topic request to the controller. How does the controller associate it with the mirror name to create a `MirrorTopicStateChangeRecord`? No, we don’t send a create topic request to the controller here. Actually we send a StartMirrorTopics request to the controller, which contains the mirror name and topic metadata like NumPartitions, topicId… etc. So the controller can create the non-existed topic as explained in JR48, and associate it to MirrorTopicStateChangeRecord. > JR53. Describe Mirrors: > JR53.1 "1. The user sends a DescribeClusterMirrors request with optional mirror names (empty means all mirrors)." In metadata request, we use a null arrary to represent all topics. It would be useful to be consistent here. OK, KIP updated. > JR53.2 "2. The admin client fans out the request to all brokers and merges the responses client-side." This is not ideal. Not every broker hosts a mirror partition. So sending the request to every broker is wasteful. It's probably better to send the request to a single broker, which can then contact the right MirrorFetcherManager and ClusterMirrorCoordinator. Agree, KIP updated. > JR54. MirrorTopicStateChangeRecord.DesiredState: Are MIRRORING, PAUSED, STOPPED the only validate states? The state table has other states too. Ditto for MirrorPartitionState. The DesiredState field in MirrorTopicStateChangeRecord is to define the target state the user wants. And the desired state users can assign is MIRRORING/PAUSED/STOPPED. Users cannot assign a desired state as STOPPING or LOG_TRUNCATION… etc because they are more like intermediate states. About MirrorPartitionState record, it’ll record any state in the state table that the current partition is. This is the real state for each partition. It’s like in the kubernetes world, when we change the replica field of the deployment yaml from 3 -> 5 (desired state), it won’t increase to 5 immediately. The current replica (MirrorPartitionState record) is still 3, then 4 (maybe), and finally 5. Thank you, Luke On Tue, Aug 4, 2026 at 3:21 AM Jun Rao via dev <[email protected]> wrote: > Hi, Fede, > > Thanks for the reply. A few more comments. > > JR31.3 "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." > Hmm, a given broker only caches the mirror partition state for partitions > hashed into the __mirror_state partition hosted on this broker, right? Why > does it need to fetch the state from other remote coordinators? > > JR36. I am still not quite sure why we need MirrorLeaderEpoch in the Fetch > request. In the current Fetch request, in addition to CurrentLeaderEpoch, > it has another field LastFetchedEpoch, which captures the last leader epoch > in the data and is used to resolve the inconsistency between the follower > and the leader. Does it achieve the same purpose as MirrorLeaderEpoch? > > JR48. "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." > Hmm, this seems inconsistent. When a mirror is started, the source topic > may not exist yet. When the source topic is created later, we can start > mirroring without requiring the user to provide the number of partitions. > > JR51. Start Mirror: I am wondering if we truly need the last mirror epoch > for truncation when starting a mirror. When a mirror is stopped at the > source, we already bump up the leader epoch and write some additional > records for cleanup. It seems that we can just use the DivergingEpoch > returned in the initial mirror Fetch request to detect the diverging epoch > and use that for truncation at the mirror destination. > > JR52. Stop Mirror: > JR52.1 "3. If patterns are provided, the controller removes matching > entries from mirror.topics.include or adds them to mirror.topics.exclude on > the CLUSTER_MIRROR resource." > Why is it 'or' and not 'and'? > JR52.2 "7. On subsequent metadata refresh cycles, MirrorMetadataManager > discovers new source topics matching the persisted include/exclude patterns > and repeats steps 3 through 7 for each." > The broker will send a create topic request to the controller. How does the > controller associate it with the mirror name to create a > `MirrorTopicStateChangeRecord`? > > JR53. Describe Mirrors: > JR53.1 "1. The user sends a DescribeClusterMirrors request with optional > mirror names (empty means all mirrors)." > In metadata request, we use a null arrary to represent all topics. It would > be useful to be consistent here. > JR53.2 "2. The admin client fans out the request to all brokers and merges > the responses client-side." > This is not ideal. Not every broker hosts a mirror partition. So sending > the request to every broker is wasteful. It's probably better to send the > request to a single broker, which can then contact the right > MirrorFetcherManager and ClusterMirrorCoordinator. > > JR54. MirrorTopicStateChangeRecord.DesiredState: Are MIRRORING, PAUSED, > STOPPED the only validate states? The state table has other states too. > Ditto for MirrorPartitionState. > > Jun > > On Fri, Jul 31, 2026 at 7:52 AM Federico Valeri <[email protected]> > wrote: > > > Hi Andrew, yes, validation is called at the start of createTopic. If > > the topic ID is invalid, the request fails with INVALID_REQUEST. If > > the topic ID is already used by a different topic name, it fails with > > TOPIC_ALREADY_EXISTS. In both cases, only the individual topic in the > > batch fails; other topics in the same CreateTopics request continue > > processing normally. Added this information to the KIP. > > > > > > > > On Fri, Jul 31, 2026 at 10:53 AM Andrew Schofield <[email protected] > > > > wrote: > > > > > > Hi Fede, > > > One more small question. > > > > > > AS37: If the controller fails validation for the MirrorInfo.TopicId > > added to the CreateTopic request, what does it do? I suspect that it > fails > > the request with a particular error code. > > > > > > Thanks, > > > Andrew > > > > > > On 2026/07/30 17:13:46 Federico Valeri wrote: > > > > Hi all, > > > > > > > > We updated the KIP to include Coordinator Runtime configurations and > > > > the coordinator state transition validation paragraph. The latter > > > > covers the per-partition epoch fencing mechanism (leader epoch and > > > > state epoch) that complements the existing per-topic StateOffset > > > > fencing for lifecycle operations. > > > > > > > > Thanks, > > > > Fede > > > > > > >
