Hi, Luke, Thanks for the reply.
JR36. "So I think we can say, the “MirrorLeaderEpoch” is specifically designed for the unclean leader election support in Cluster Mirroring because the unclean leader election will break the assumption that the committed data won’t be truncated." If the source cluster has an unclean leader election, the destination cluster will detect the inconsistency and truncate the log at the destination leader. We can just extend mirror.support.unclean.leader.election to cover this case, i.e., we wait until all replicas' log end offsets are reduced to the target offset before resuming mirroring. This avoids the inconsistency. If the destination cluster has an unclean leader election, it will re-fetch the same data from the source. There won't be any inconsistency either. So, I think we can avoid adding MirrorLeaderEpoch. JR51. Make sense. To truncate based on LME, we need to issue another RPC to translate the LME to an offset. It would be useful to include that in the workflow description. JR53.2 The current DescribeClusterMirrorsRequest is a bit strange because it mixes querying the mirror partition state with querying the mirror partition leader. Another option is to use ReadMirrorStatesRequest to query the mirror partition state and a new request to query the mirror partition leader. If we want the adminClient to query the right ClusterMirrorCoordinator directly, we can extend the FindCoordinatorRequest to support that. JR55. For __mirror_state topic, could we describe whether it is compacted? Jun On Wed, Aug 5, 2026 at 4:51 AM Luke Chen <[email protected]> wrote: > Hi Jun, > > Thanks for the comments. > Answering them below: > > > JR31.3 It makes sense for the new partition leader to query the current > mirror state from the local or remote MirrorMetadataManager that owns the > partition's state. However, the wording in the KIP is confusing. > MirrorMetadataManager > mirror partition state cache should be populated only by one way, which is > replaying __mirror_state partitions this broker leads. By saying "and by > fetching state from remote coordinators via ReadMirrorStates RPCs.", it's > as if the mirror partition state cache on each broker stores the state for > all mirror partitions. > > You’re right. KIP is updated to make it clear. > > > > JR36. I understand the intention now, but I still don't think we need > MirrorLeaderEpoch. Let's > assume the common case: no unclean leader election in the source cluster > during mirroring. Mirroring only fetches the committed data from the > source, which never changes if obtained a second time. In your example > image, follower broker 1 in the destination cluster can never obtain the > record B(0) because it's never committed in the source and thus is never > exposed to the destination leader. It's possible for broker 1 to first get > D(1) which is truncated later when broker 0 follows a new leader. But the > new leader will refetch the same D(1) from the source cluster. So, it's ok > for broker 1 to keep D(1). > > > Yes, you are correct. Because mirroring only fetches the committed data > from the source, the MirrorLeaderEpoch is unnecessary for normal mirroring. > Thanks for the clarification. So I think we can say, the > “MirrorLeaderEpoch” is specifically designed for the unclean leader > election support in Cluster Mirroring because the unclean leader election > will break the assumption that the committed data won’t be truncated. > > Given that, we could gate the MirrorLeaderEpoch field by only setting it > in the Fetch request when "mirror.support.unclean.leader.election=true" > (false by default), and skip it entirely for normal mirroring. What do you > think? > > > > JR51. Sounds good. Why do we choose to store the last mirror epoch > instead > of a last mirror offset? > > Yes, originally we stored the “last mirror offset” instead of “last mirror > epoch”. But later, we found it doesn’t work for unclean leader election > cases. (yes, `unclean leader election` again :) ) For example: > > Cluster A (source) --> Cluster B (dest) > foo-0 leader log foo-0 leader log > Offset 0: A (LE: 0) Offset 0: A (LE: 0) > Offset 1: A (LE: 0) Offset 1: A (LE: 0) > > Cluster B is mirroring from A. And A crashes and failover to B. B stores > the “last mirror offset” as 1. > When cluster A recovers, somehow the unclean leader election happens, log > is empty in the new leader, and then some more leadership changes to make > the leader epoch become 10. Then it starts the reverse mirror from B. The > log looks like this: > > Cluster A (dest) <-- Cluster B (source) > foo-0 leader log foo-0 leader log > Offset 0: B (LE: 10) Offset 0: A (LE: 0) > Offset 1: B (LE: 10) Offset 1: A (LE: 0) > Offset 2: pid_res (LE: 10) > > When A starts mirroring from B, log is truncated to last mirror offset: 1 > (no-op), and then fetch from cluster B. The 1st fetch request is like this: > { currentLeaderEpoch: 10, lastFetchedEpoch: 10, fetchOffset: 2, … }. The > divergingEpoch check in the cluster B queries the endOffset for leader > epoch 10 (in Partition#readRecords), and found the end offset for LE 10 is > 2 in B, which is >= fetchOffset 2, so divergingEpoch validation passes and > returns offset 2 to cluster A. In short, relying on “last mirror offset” > cannot resolve the log diverge issue in unclean leader election case. > > If instead, we truncate until “last mirror epoch”: 0, the offset 0 and 1 > in cluster A will be truncated as expected and then log converge with > cluster B correctly in the end. > > > Also, about JR53.2: > > 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. > > While working on this, I found we need to add one more tagged field > "Forwarded" to the DescribeClusterMirrorsRequest schema because otherwise > the receiving broker doesn’t know if this is the request from client and it > needs to contact other brokers, or this is just a forwarded request. The > workflow is like this: > 1. the adminClient sends the DescribeClusterMirrorsRequest to a > leastLoaded broker, which doesn’t contain the "Forwarded" field. > 2. The broker receives the request and sends DescribeClusterMirrorsRequest > with "Forwarded=true" to the responsible brokers and waits for the response > async. > 3. When all responses are received, merge the results and respond to the > client. > > Note: we cannot use ReadMirrorStatesRequest here because it doesn’t > contain the lag info in the coordinator. > > The alternative solution is the adminClient sends > DescribeClusterMirrorsRequest to the “responsible brokers only” (via > describeTopics), instead of ALL brokers, and merge the results in the > client side. This solution doesn’t need to add a new field in the > DescribeClusterMirrorsRequest and still can avoid unnecessary requests > sent. > > My preference is to keep the merge logic on the admin client side, which > saves the overhead in the broker side. > What do you think? > > > Thank you, > Luke > > On Wed, Aug 5, 2026 at 3:19 AM Jun Rao via dev <[email protected]> > wrote: > >> Hi, Luke, >> >> Thanks for the reply. >> >> JR31.3 It makes sense for the new partition leader to query the current >> mirror state from the local or remote MirrorMetadataManager that owns the >> partition's state. However, the wording in the KIP is confusing. >> MirrorMetadataManager >> mirror partition state cache should be populated only by one way, which is >> replaying __mirror_state partitions this broker leads. By saying "and by >> fetching state from remote coordinators via ReadMirrorStates RPCs.", it's >> as if the mirror partition state cache on each broker stores the state for >> all mirror partitions. >> >> JR36. I understand the intention now, but I still don't think we need >> MirrorLeaderEpoch. Let's >> assume the common case: no unclean leader election in the source cluster >> during mirroring. Mirroring only fetches the committed data from the >> source, which never changes if obtained a second time. In your example >> image, follower broker 1 in the destination cluster can never obtain the >> record B(0) because it's never committed in the source and thus is never >> exposed to the destination leader. It's possible for broker 1 to first get >> D(1) which is truncated later when broker 0 follows a new leader. But the >> new leader will refetch the same D(1) from the source cluster. So, it's ok >> for broker 1 to keep D(1). >> >> JR51. Sounds good. Why do we choose to store the last mirror epoch instead >> of a last mirror offset? >> >> Jun >> >> On Tue, Aug 4, 2026 at 5:26 AM Luke Chen <[email protected]> wrote: >> >> > 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 >> >> > > > >> >> > >> >> >> > >> >
