Hi, Luke, Thanks for the reply.
JR59.1 Could you add the RPC to the workflow? The broker first needs to issue ListClusterMirrorsRequest with SourceClusterIdFilter to get the mirror name and then issue DescribeClusterMirrors to get the LME. JR71. mirror.support.unclean.leader.election : The name is confusing. It's also used in LOG_TRUNCATION, which is not related to unclean leader election. Could we use a better name? JR72. DescribeClusterMirrors JR72.1 A user may specify just the topic name when describing a mirror. Should we include the topic name in the request? JR72.2 LastMirrorEpochLookups is a bit weird since the response returns either mirror state or mirror lag and LastMirrorEpochLookups doesn't match either. JR72.3 Should it return UNKNOWN_CLUSTER_MIRROR too? JR72.4 "includeMirrorLags: source offset, destination offset, lag" DescribeClusterMirrorsRequest doesn't include those fields. Also, lag should be omitted. JR73. MirrorPartitionState record, LastMirrorEpochs record and MirrorPartitionStateRequest: The values for type and name are incorrect. JR74. ListClusterMirrorsRequest JR74.1 When is DesiredMirrorStateFilter used? There is no option to specify the desired state in the mirror-topic command. JR74.2 Should it return CLUSTER_MIRROR_AUTHORIZATION_FAILED too? JR75. ReadMirrorLags JR75.1 Should it be named ReadMirrorOffsets since the response doesn't include lag? JR75.2 Should it return UNKNOWN_CLUSTER_MIRROR too? JR75.3 Some places refer to this as ReadClusterMirrorLags. We need to be consistent. JR76. WriteMirrorStates: Should it return UNKNOWN_CLUSTER_MIRROR too? JR77. Delete Mirror: "3. The controller verifies that no topics are in MIRRORING or PAUSED state." Is this necessary since step 2 has verified all partitions are in STOPPED state, which can only happen if the topics are in STOPPED state. JR78. STOPPING: "The mirror fetcher is removed, the leader epoch is bumped, ABORT markers are appended for all ongoing transactions, last mirror epochs are persisted, and a MIRROR_PID_RESET record appended." This describes the state for STOPPED, not STOPPING. JR79. RPCs like DescribeClusterMirrors can take time. Should we add a timeout field? JR80. PauseMirrorTopicsOptions: Should it support patterns? The command line seems to suppor it. JR81. AdminClient JR81.1 ListClusterMirrors: We need to add the option to specify different filters supported in the request. JR81.2 DescribeClusterMirrorsOptions.includeAuthorizedOperations: The request doesn't support this option. Could we add other options supported in the request? JR81.3 LeaderStateDescription: Should it be named MirrorPartitionStateDescription? Also, could we make the fields consistent with those in the response? Jun On Wed, Aug 12, 2026 at 5:36 AM Luke Chen <[email protected]> wrote: > Hi Jun, > > Thanks for the good comments again. > Answering them below. > > > JR51. "ULE_LOG_TRUNCATION When mirror.support.unclean.leader.election is > enabled and the destination leader has log truncation, the state will move > from MIRRORING to this state." > This still suggests that divergeEpoch is only handled if > mirror.support.unclean.leader.election > is enabled. > > > OK, KIP is updated to make it clear. > > > JR55. Should the mirror state and the mirror lags always be exposed > together? If so, we can remove IncludeMirrorState and IncludeMirrorLags > from DescribeClusterMirrorsRequest. If not, it's better to use two separate > RPCs. > > > No, we sometimes only need MirrorState, and sometimes need both. So I’ll > keep “IncludeMirrorState” and “IncludeMirrorLags” fields in > DescribeClusterMirrorsRequest. And we will use ReadMirrorStates and > ReadMirrorLags to get each info. > > > > DescribeClusterMirrorsRequest should be used in the Describe Mirror flow, > right? Currently it's missing. Also, I assume the admin client refers to > the one in the kafka-cluster-mirrors tool. But it seems that sometimes it > also refers to the admin client on the broker? Perhaps we can use 'the > tool' instead to make it clear. > > Yes, I missed the DescribeClusterMirrorsRequest. Added. > And sorry for the confusion. I’d like to keep the admin client as the one > invoking from the tool (or user’s direct admin client API invocation). On > the broker side, we actually use InterBrokerSendThread to invoke these RPCs > or get directly from MetadataImage. I’ve updated the KIP to make it clear. > > > > JR58. BumpLeaderEpochs: Why does it need to be issued to the broker, > instead of just to the controller? > > Yes, we only need to issue to the controller. Updated to the > BumpLeaderEpochs API. > > > > JR59. Start Mirror Topics: > > JR59.1 "During this state, the broker queries the source cluster for last > mirror epochs". > Could you describe which RPC is issued to find the LME? > > We’re using describeClusterMirror API via admin client here. > > > JR59.2 When does it receive INVALID_REPLICATION_FACTOR? The request > doesn't > contain a replication factor. > > We use “default.replication.factor” when topic creation, so the > INVALID_REPLICATION_FACTOR will be thrown like normal topic creation. Also, > because we require the replicas for the partition assignment to be all > active and not in fenced or controlled shutdown(JR35), we’ll throw > INVALID_REPLICATION_FACTOR if some replicas are fenced. > > > > JR59.3 TOPIC_ALREADY_EXISTS: "Request topic ID is already used by a > different topic name" > Should we also return this error when the topic name is used as a regular > topic? > > There are 2 errors we will throw when topic id <-> topic name mismatch. > Suppose provided topic id: ID_A, topic name: NAME_A > 1. Destination cluster exists topic id: ID_A, but topic name is: NAME_B > TOPIC_ALREADY_EXISTS will be thrown > 2. Destination cluster exists topic id: ID_B, but topic name is: NAME_A > INCONSISTENT_TOPIC_ID will be thrown > > Hmm.. indeed, it looks like using the same TOPIC_ALREADY_EXISTS error for > these 2 cases makes more sense. KIP updated. > > > > JR60. CreateClusterMirrorRequest: When will it receive an > INVALID_CLUSTER_MIRROR_STATES error not covered by > CLUSTER_MIRROR_ALREADY_EXISTS? > > > Sorry, createClusterMirrorRequest doesn’t throw > INVALID_CLUSTER_MIRROR_STATES because we don’t need to validate the mirror > states when creating a cluster mirror. We only care if the “mirror name” > exists or not. KIP updated. > > > > JR61. Since we have INVALID_CLUSTER_MIRROR_STATES, should we just remove > MIRROR_TOPIC_NOT_STOPPED, MIRROR_TOPIC_NOT_PAUSED and > CLUSTER_MIRROR_NOT_EMPTY? We can use the error message to describe the > specific state causing the error. > > I agree. INVALID_CLUSTER_MIRROR_STATES can cover the following errors: > a. MIRROR_TOPIC_ALREADY_PAUSED > b. MIRROR_TOPIC_NOT_PAUSED, > c. MIRROR_TOPIC_ALREADY_STOPPED, > d. CLUSTER_MIRROR_NOT_EMPTY > e. MIRROR_TOPIC_NOT_STOPPED > > KIP is updated. > > > > JR62. NOT_CONTROLLER/UNKNOWN_CLUSTER_MIRROR are included in some requests > (e.g. StopMirrorTopics), but not others (e.g. StartMirrorTopics). Could we > make them consistent? > > > Good catch. The updated errors for the new added RPC are this: > > CreateClusterMirror: NOT_CONTROLLER > StartMirrorTopics: NOT_CONTROLLER and UNKNOWN_CLUSTER_MIRROR > StopMirrorTopics: NOT_CONTROLLER and UNKNOWN_CLUSTER_MIRROR > PauseMirrorTopics: NOT_CONTROLLER and UNKNOWN_CLUSTER_MIRROR > ResumeMirrorTopics: NOT_CONTROLLER and UNKNOWN_CLUSTER_MIRROR > DeleteMirrorTopics: NOT_CONTROLLER and UNKNOWN_CLUSTER_MIRROR > > > > JR63. StopMirrorTopics: > > JR63.1 "The broker validates that all target topic partitions are in > either > LOG_TRUNCATION or MIRRORING or FAILED state." > Is this accurate? LOG_TRUNCATION can't transition to STOP. PAUSED can. > > Yes, nice catch. LOG_TRUNCATION should change to PAUSED. KIP is updated. > > > JR63.2 Why does it receive MIRROR_TOPIC_ALREADY_PAUSED error since PAUSED > -> STOPPED is valid? > > Yes, it is not correct. KIP updated to remove this error. > > > JR63.3 MIRROR_TOPIC_NOT_STOPPED: "The mirror topic is not in stopped > state" > This is weird. The reason to issue StopMirrorTopics is that it's not in > stopped state. > > Indeed. KIP is updated to only include INVALID_CLUSTER_MIRROR_STATES > error, and will be thrown when states are not in MIRRORING or PAUSED state. > > > JR64. ReadClusterMirrorLagsResponse.lag is redundant since it's derived > from SourceOffset and DestinationOffset. > > True. Removed. > > > JR65. Why do we need WriteMirrorStates.StoppedTopics? It seems that we > could just set the partition level state to STOPPED. > > Nice catch. We don’t use it at all. That is some old leftover field we > forgot to remove. KIP updated. > > > JR66. ReadMirrorStates/WriteMirrorStates: Why do they need to be sent to > the controller? > > No, these 2 only send to the broker. KIP updated. > > > > JR67. WriteMirrorStates: Do we need both LeaderEpoch and StateEpoch? It > seems both are used for fencing and fencing based on the latter is > stronger. > > Yes, basically the StateEpoch can fence for most cases. The LeaderEpoch > fencing is used for cases like: > > 1. b1 is the current leader for foo-0, preparing write state from > MIRRORING to EPOCH_FENCING in { LE: 1, stateEpoch: 1 } > 2. leadership change to b2 for foo-0, LE changes to 2. > 3. b2 reads state from coordinator with { LE: 2, stateEpoch: 1 }. > 4. The coordinator receives the read from b2 in step (3), the highest > lastSeen LE for foo-0 updates to 2 in coordinator. > 5. The writes state from b1 in step (1) arrives the coordinator, but it > rejects b1 writes because the highest lastSeen LE is 2, which is > 1 in the > request. > 6. b2 writes state from MIRRORING to PAUSING in LE: 2, stateEpoch: 1 and > succeeds. > > The leaderEpoch fence is to avoid any stale broker writes after the new > leader epoch read. This cannot be protected by the stateEpoch. The > stateEpoch can prevent the write with older stateEpoch. So in the situation > above, the stateEpoch cannot determine if we should accept b1 writes or > not, but leaderEpoch can. By the way, we borrowed this concept from > KIP-932: > https://cwiki.apache.org/confluence/spaces/KAFKA/pages/255070434/KIP-932+Queues+for+Kafka#KIP932%3AQueuesforKafka-WriteShareGroupStateAPI > <https://urldefense.com/v3/__https://cwiki.apache.org/confluence/spaces/KAFKA/pages/255070434/KIP-932*Queues*for*Kafka*KIP932*3AQueuesforKafka-WriteShareGroupStateAPI__;KysrIyU!!Ayb5sqE7!rM45DU3iceu-9W4moXR1xuj6XCrozML7X6eIkpxep5RCI9NA3jAcpJzCnEx76Q83EhQko0_Ye9r9bPQ$> > > > > JR68. Could we make the usage of topicId and topicName in the RPCs > consistent? Also, BumpLeaderEpochsRequest uses topicId, but > BumpLeaderEpochsReponse uses TopicName. > > Yes, we have made: > 1. BumpLeaderEpochsRequest/BumpLeaderEpochsResponse use TopicName. > 2. DescribeClusterMirrorRequest/DescribeClusterMirrorResponse use > TopicName. > 3. MirrorTopicStateChangeRecord uses TopicName because we will do > validation before writing the metadata log. > > > > JR69. MIRROR_PID_RESET: > > JR69.1 Could you specify the value of the control record's type? > > It’s MIRROR_PID_RESET(7). KIP updated. > > > JR69.2 How is the SourceClusterId field used? > > I’m also surprised to see this field in the MirrorPidResetRecord. I traced > back to the internal PR we added for this field, and confirmed it is not > used since it was added. Sorry for the confusion and nice catch! KIP > updated. > > > JR70. MirrorPartitionState: type should be "data" not "coordinator-key", > right? Ditto for LastMirrorEpochs. > We store both MirrorPartitionState and LastMirrorEpochs records into > internal __mirror_state topic, and rely on Coordinator Runtime to > write/read the records, like how `OffsetCommit` and `GroupMetadata` work. > So I think we should keep them as "coordinator-key"/"coordinator-value" > type. > > > Thank you, > Luke > > On Tue, Aug 11, 2026 at 7:02 AM Jun Rao via dev <[email protected]> > wrote: > >> Hi, Luke, >> >> Thanks for the reply. >> >> JR51. "ULE_LOG_TRUNCATION When mirror.support.unclean.leader.election is >> enabled and the destination leader has log truncation, the state will move >> from MIRRORING to this state." >> This still suggests that divergeEpoch is only handled if >> mirror.support.unclean.leader.election >> is enabled. >> >> JR55. Should the mirror state and the mirror lags always be exposed >> together? If so, we can remove IncludeMirrorState and IncludeMirrorLags >> from DescribeClusterMirrorsRequest. If not, it's better to use two >> separate >> RPCs. >> >> DescribeClusterMirrorsRequest should be used in the Describe Mirror flow, >> right? Currently it's missing. Also, I assume the admin client refers to >> the one in the kafka-cluster-mirrors tool. But it seems that sometimes it >> also refers to the admin client on the broker? Perhaps we can use 'the >> tool' instead to make it clear. >> >> JR58. BumpLeaderEpochs: Why does it need to be issued to the broker, >> instead of just to the controller? >> >> JR59. Start Mirror Topics: >> JR59.1 "During this state, the broker queries the source cluster for last >> mirror epochs". >> Could you describe which RPC is issued to find the LME? >> JR59.2 When does it receive INVALID_REPLICATION_FACTOR? The request >> doesn't >> contain a replication factor. >> JR59.3 TOPIC_ALREADY_EXISTS: "Request topic ID is already used by a >> different topic name" >> Should we also return this error when the topic name is used as a regular >> topic? >> >> JR60. CreateClusterMirrorRequest: When will it receive an >> INVALID_CLUSTER_MIRROR_STATES error not covered by >> CLUSTER_MIRROR_ALREADY_EXISTS? >> >> JR61. Since we have INVALID_CLUSTER_MIRROR_STATES, should we just remove >> MIRROR_TOPIC_NOT_STOPPED, MIRROR_TOPIC_NOT_PAUSED and >> CLUSTER_MIRROR_NOT_EMPTY? We can use the error message to describe the >> specific state causing the error. >> >> JR62. NOT_CONTROLLER/UNKNOWN_CLUSTER_MIRROR are included in some requests >> (e.g. StopMirrorTopics), but not others (e.g. StartMirrorTopics). Could we >> make them consistent? >> >> JR63. StopMirrorTopics: >> JR63.1 "The broker validates that all target topic partitions are in >> either >> LOG_TRUNCATION or MIRRORING or FAILED state." >> Is this accurate? LOG_TRUNCATION can't transition to STOP. PAUSED can. >> JR63.2 Why does it receive MIRROR_TOPIC_ALREADY_PAUSED error since PAUSED >> -> STOPPED is valid? >> JR63.3 MIRROR_TOPIC_NOT_STOPPED: "The mirror topic is not in stopped >> state" >> This is weird. The reason to issue StopMirrorTopics is that it's not in >> stopped state. >> >> JR64. ReadClusterMirrorLagsResponse.lag is redundant since it's derived >> from SourceOffset and DestinationOffset. >> >> JR65. Why do we need WriteMirrorStates.StoppedTopics? It seems that we >> could just set the partition level state to STOPPED. >> >> JR66. ReadMirrorStates/WriteMirrorStates: Why do they need to be sent to >> the controller? >> >> JR67. WriteMirrorStates: Do we need both LeaderEpoch and StateEpoch? It >> seems both are used for fencing and fencing based on the latter is >> stronger. >> >> JR68. Could we make the usage of topicId and topicName in the RPCs >> consistent? Also, BumpLeaderEpochsRequest uses topicId, but >> BumpLeaderEpochsReponse uses TopicName. >> >> JR69. MIRROR_PID_RESET: >> JR69.1 Could you specify the value of the control record's type? >> JR69.2 How is the SourceClusterId field used? >> >> JR70. MirrorPartitionState: type should be "data" not "coordinator-key", >> right? Ditto for LastMirrorEpochs. >> >> Jun >> >> On Mon, Aug 10, 2026 at 5:52 AM Luke Chen <[email protected]> wrote: >> >> > Hi Jun, >> > >> > Thanks for the comments. >> > Answering them below. >> > >> > > JR36. Sounds good. Should we remove >> MirrorLeaderEpochExceededException? >> > >> > Sorry, it’s a confusing name. The MirrorLeaderEpochExceededException is >> > used and thrown when the fetched mirror batch leader epoch > local >> leader >> > epoch in destination cluster. And then we’ll stop mirroring and move the >> > partition to EPOCH_FENSING state to wait for local leader epoch bump. I >> > think since we don’t have the concept of `MirrorLeaderEpoch` anymore, >> the >> > `MirrorLeaderEpochExceededException` should not cause confusion now. >> I’ll >> > keep the name for the exception. >> > >> > >> > > JR51. Sounds good. Do we plan to always set lastFetchedEpoch in the >> > mirror >> > fetch request issued to the source cluster, independent of the value of >> > mirror.support.unclean.leader.election? It's needed to detect divergence >> > during failover. However, if it's also set when >> > mirror.support.unclean.leader.election=false, >> > an unclean leader election in the source cluster can halt the mirroring >> > since we don’t do anything with the returned divergeEpoch in that mode >> > and continue >> > the next fetch. >> > >> > > Do we plan to always set lastFetchedEpoch in the mirror >> > fetch request issued to the source cluster, independent of the value of >> > Mirror.support.unclean.leader.election? >> > >> > Yes, we always set lastFetchedEpoch in the mirror fetch request. >> > >> > > However, if it's also set when >> > mirror.support.unclean.leader.election=false, >> > an unclean leader election in the source cluster can halt the mirroring >> > since we don’t do anything with the returned divergeEpoch in that mode >> > and continue the next fetch. >> > >> > Sorry I didn’t make it clear. When >> > “mirror.support.unclean.leader.election=false” and a “divergeEpoch” is >> set >> > in fetch response, the leader node will handle the divergeEpoch and then >> > continue the next fetch as usual. The difference with >> > “mirror.support.unclean.leader.election=true” is that “should we wait >> for >> > ALL replicas caught up (complete truncation)”. Without the waiting, it’s >> > possible the inconsistent log issue in KAFKA-18723 could happen between >> > leader and follower nodes in the destination cluster because we don’t >> have >> > “MirrorLeaderEpoch” concept now to reject higher leader epoch batches. >> That >> > is, we won't halt the mirroring if "divergeEpoch" is set but >> > “mirror.support.unclean.leader.election=false”. >> > >> > >> > > JR55. The DescribeClusterMirrorsRequest still returns both the mirror >> > partition state and the mirror partition leader state. The problem is >> that >> > if a caller only needs one of the state, the other state becomes >> redundant. >> > For example, when starting the mirror, to obtain LME, we only need the >> > mirror partition state. Here is an alternative design: We expose the >> mirror >> > partition state through ReadMirrorState. The client will use the >> > FindCoordinatorRequest >> > to find the right mirror partition coordinator to issue that request. We >> > expose the mirror partition leader state through the >> ReadClusterMirrorLags >> > request. The client can use the metadata request to find the partition >> > leader to issue that request. >> > >> > > In the describing mirror flow, the admin client first issues the >> > ListMirrorRequest to find all the mirrors and the topics being >> mirrored. It >> > then issues the metadata request to find the partitions for those topics >> > and issues a FindCoordinatorRequest to find the mirror partition state >> > coordinators. Finally, it issues ReadMirrorState to the mirror partition >> > state coordinator and ReadClusterMirrorLags to the partition leader. >> > >> > > In the starting mirror flow, the partition leader in the destination >> > cluster issues a ListMirrorRequest to find the mirror name based on its >> > cluster id. It then issues a FindCoordinatorRequest to find the mirror >> > partition state coordinators. Finally, it issues ReadMirrorState to the >> > mirror partition state coordinator to get the LME. >> > >> > > We can also improve the ListMirror request to accept a filter based on >> > mirror name, source cluster ID and topic name. >> > >> > >> > Answers: >> > >> > > The DescribeClusterMirrorsRequest still returns both the mirror >> > partition state and the mirror partition leader state. The problem is >> that >> > if a caller only needs one of the state, the other state becomes >> redundant. >> > For example, when starting the mirror, to obtain LME, we only need the >> > mirror partition state. >> > >> > Good point! >> > >> > >> > > Here is an alternative design: We expose the mirror >> > partition state through ReadMirrorState. The client will use the >> > FindCoordinatorRequest >> > to find the right mirror partition coordinator to issue that request. We >> > expose the mirror partition leader state through the >> ReadClusterMirrorLags >> > request. The client can use the metadata request to find the partition >> > leader to issue that request. >> > >> > >> > Sounds good. But we try not to expose >> > ReadMirrorState/ReadClusterMirrorLags directly to the admin client >> because >> > users might get confused about these 2 requests, and they are also both >> > related to describeClusterMirror. I think we can add 2 additional >> fields to >> > describeClusterMirrorRequest: “includeMirrorStates” and >> > “includeMirrorLags”. So the admin client can include either one or both >> of >> > them in the DescrbieClusterMirrorRequest based on the provided options. >> KIP >> > is updated. >> > >> > >> > > In the describing mirror flow, the admin client first issues the >> > ListMirrorRequest to find all the mirrors and the topics being >> mirrored. It >> > then issues the metadata request to find the partitions for those topics >> > and issues a FindCoordinatorRequest to find the mirror partition state >> > coordinators. Finally, it issues ReadMirrorState to the mirror partition >> > state coordinator and ReadClusterMirrorLags to the partition leader. >> > >> > Makes sense. KIP is updated. >> > >> > >> > > In the starting mirror flow, the partition leader in the destination >> > cluster issues a ListMirrorRequest to find the mirror name based on its >> > cluster id. It then issues a FindCoordinatorRequest to find the mirror >> > partition state coordinators. Finally, it issues ReadMirrorState to the >> > mirror partition state coordinator to get the LME. >> > >> > Make sense. So we can send DescribeClusterMirrorRequest with >> > “includeMirrorStates” only. >> > >> > > We can also improve the ListMirror request to accept a filter based on >> > mirror name, source cluster ID and topic name. >> > >> > Good suggestion, just like what ListGroupsRequest does. >> > Added "SourceClusterIdFilter", "MirrorNameFilter", >> > "DesiredMirrorStateFilter" in the ListClusterMirrorsRequest. KIP >> updated. >> > >> > >> > > JR56. Could you list the ACL needed for each new RPC? >> > >> > >> > Yes, it’s already listed in “Security Control” section. >> > >> > >> > > JR57. In the command workflow, it sometimes says "The user sends" and >> > some >> > other times says "The admin client sends". It's better to use "The admin >> > client" consistently in all those places. >> > >> > Agree. KIP updated. >> > >> > >> > Thank you, >> > Luke >> > >> > On Sat, Aug 8, 2026 at 9:34 AM Jun Rao via dev <[email protected]> >> > wrote: >> > >> >> Hi, Luke, >> >> >> >> Thanks for the reply. >> >> >> >> JR36. Sounds good. Should we remove MirrorLeaderEpochExceededException? >> >> >> >> JR51. Sounds good. Do we plan to always set lastFetchedEpoch in the >> mirror >> >> fetch request issued to the source cluster, independent of the value of >> >> mirror.support.unclean.leader.election? It's needed to detect >> divergence >> >> during failover. However, if it's also set when >> >> mirror.support.unclean.leader.election=false, >> >> an unclean leader election in the source cluster can halt the mirroring >> >> since we don’t do anything with the returned divergeEpoch in that mode >> >> and continue >> >> the next fetch. >> >> >> >> JR55. The DescribeClusterMirrorsRequest still returns both the mirror >> >> partition state and the mirror partition leader state. The problem is >> that >> >> if a caller only needs one of the state, the other state becomes >> >> redundant. >> >> For example, when starting the mirror, to obtain LME, we only need the >> >> mirror partition state. Here is an alternative design: We expose the >> >> mirror >> >> partition state through ReadMirrorState. The client will use the >> >> FindCoordinatorRequest >> >> to find the right mirror partition coordinator to issue that request. >> We >> >> expose the mirror partition leader state through the >> ReadClusterMirrorLags >> >> request. The client can use the metadata request to find the partition >> >> leader to issue that request. >> >> >> >> In the describing mirror flow, the admin client first issues the >> >> ListMirrorRequest to find all the mirrors and the topics being >> mirrored. >> >> It >> >> then issues the metadata request to find the partitions for those >> topics >> >> and issues a FindCoordinatorRequest to find the mirror partition state >> >> coordinators. Finally, it issues ReadMirrorState to the mirror >> partition >> >> state coordinator and ReadClusterMirrorLags to the partition leader. >> >> >> >> In the starting mirror flow, the partition leader in the destination >> >> cluster issues a ListMirrorRequest to find the mirror name based on its >> >> cluster id. It then issues a FindCoordinatorRequest to find the mirror >> >> partition state coordinators. Finally, it issues ReadMirrorState to the >> >> mirror partition state coordinator to get the LME. >> >> >> >> We can also improve the ListMirror request to accept a filter based on >> >> mirror name, source cluster ID and topic name. >> >> >> >> JR56. Could you list the ACL needed for each new RPC? >> >> >> >> JR57. In the command workflow, it sometimes says "The user sends" and >> some >> >> other times says "The admin client sends". It's better to use "The >> admin >> >> client" consistently in all those places. >> >> >> >> Jun >> >> >> >> >> >> >> >> On Thu, Aug 6, 2026 at 5:13 AM Luke Chen <[email protected]> wrote: >> >> >> >> > Hi Jun, >> >> > >> >> > Thanks for your comments. >> >> > Answering them below. >> >> > >> >> > >> >> > > 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. >> >> > >> >> > >> >> > Yes, that works, too. I agree this is better because the unclean >> leader >> >> > election is rare, we don't want to occupy a field just for it. >> >> > >> >> > So to achieve this, we have to: >> >> > 1. add one more state: “ULE_LOG_TRUNCATION”, >> >> > 2. When the destination cluster leader completes a log truncation >> after >> >> a >> >> > fetch response due to divergeEpoch, we know it must be caused by >> unclean >> >> > leader election in the source cluster because we only fetch committed >> >> > records. If “mirror.support.unclean.leader.election=false”, we don’t >> do >> >> > anything and continue the next fetch. If >> >> > “mirror.support.unclean.leader.election=true”, we remove the mirror >> >> fetcher >> >> > and move the partition to “ULE_LOG_TRUNCATION” state. >> >> > 3. In “ULE_LOG_TRUNCATION” state, we wait for ALL replicas to be >> caught >> >> up >> >> > into ISR and reduced to the expected LEO. >> >> > 4. Move to “MIRRORING” state and create the mirror fetcher >> >> > >> >> > So that we don’t need the “MirrorLeaderEpoch” field in Fetch API. >> >> > KIP is updated. >> >> > >> >> > >> >> > > 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. >> >> > >> >> > About the additional RPC to translate the LME to the offset (i.e. >> >> > OffsetForLeaderEpochRequest), actually we don’t rely on this to do >> the >> >> > offset translation. We use the local leader epoch cache in the >> >> destination >> >> > cluster to translate the epoch to save the RPC call. This works >> because >> >> of >> >> > an invariant the LME truncation protocol maintains: for any leader >> >> epoch ≤ >> >> > LME, that epoch's data belongs to exactly one cluster, either it >> >> originated >> >> > there, or it was mirrored. This holds even across repeated mirror -> >> >> > reverse-mirror -> mirror cycles, since each direction's >> LOG_TRUNCATION >> >> > phase re-establishes the invariant before that cluster starts >> mirroring. >> >> > With this invariant, the replication protocol can find out the >> diverge >> >> > point and truncate the log correctly when mirroring. 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) >> >> > Offset 2: A (LE: 0) >> >> > Offset 3: A (LE: 1) >> >> > >> >> > Cluster A crashes while Cluster B mirrors to offset 1. Failover to B >> and >> >> > store the LME=0 in cluster B. When cluster A recovers and starts the >> >> > reverse mirroring, it’ll query the LME from B and get back LME: 0. >> Then >> >> > cluster A gets the end offset for epoch 0 from leaderEpochCache, and >> >> gets >> >> > offset 2. So offset 3 is truncated. The log is like this now: >> >> > >> >> > Cluster A (dest) <-- Cluster B (source) >> >> > 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) >> >> > Offset 2: A (LE: 0) Offset 2: pid_res (LE: 10) >> >> > Offset 3: B (LE: 10) >> >> > >> >> > When mirror starts, cluster A sends the fetch request with { >> >> > lastFetchedEpoch: 0, fetchOffset: 3, … } to B, the divergingEpoch >> check >> >> in >> >> > the cluster B queries the endOffset for leader epoch 0, and found the >> >> end >> >> > offset for LE 0 is 1 in B, which is < fetchOffset 3, so returns fetch >> >> > response with divergingEpoch info to cluster A for log truncation. In >> >> the >> >> > end, offset 2 in cluster A will be truncated and then log converges >> with >> >> > cluster B. >> >> > >> >> > >> >> > >> >> > > 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. >> >> > >> >> > >> >> > Good suggestion. Yes, we need to fan out to all brokers because we >> have >> >> to >> >> > collect info from mirror partition leaders and mirror coordinators. >> If >> >> we >> >> > create a new API to contact mirror partition leaders, it will make >> the >> >> API >> >> > cleaner and clearer. So we will do this: >> >> > >> >> > a. Remove SourceOffset, DestinationOffset, and Lag fields from >> >> > DescribeClusterMirror request/response >> >> > >> >> > b. Add a new API: ReadClusterMirrorLags API to request for >> SourceOffset, >> >> > DestinationOffset, and Lag info. >> >> > >> >> > The workflow for admin client describeClusterMirror will be like >> this: >> >> > 1. adminClient sends describeClusterMirror to the leastLoadedNode >> broker >> >> > 2. The received broker collects the current mirrored partitions, and >> >> > groups them by coordinator node and by partition leader node. >> >> > 3. The broker sends ReadMirrorState request to the coordinator nodes >> to >> >> > get the mirror state related info >> >> > 4. The broker sends ReadClusterMirrorLags request to the mirror >> >> partition >> >> > leader nodes to get the mirror lag, source offset, destination offset >> >> > (which updates on every fetch response). >> >> > 5. The broker merge all the response into describeClusterMirror >> response >> >> > and send to admin client >> >> > >> >> > KIP is updated. Thanks for the suggestion. >> >> > >> >> > >> >> > > JR55. For __mirror_state topic, could we describe whether it is >> >> > compacted? >> >> > >> >> > Yes, it is a compacted topic. I made it much clearer in the >> >> > “ClusterMirrorCoordinator” section. >> >> > >> >> > >> >> > Thank you, >> >> > Luke >> >> > >> >> > >> >> > On Thu, Aug 6, 2026 at 6:47 AM Jun Rao via dev <[email protected] >> > >> >> > wrote: >> >> > >> >> >> 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 >> >> >> >> >> > > > >> >> >> >> >> > >> >> >> >> >> >> >> >> >> > >> >> >> >> >> >> >> > >> >> >> >> >> > >> >> >> > >> >
