Hi Jun, Thanks for the good comments. Answering them below.
> 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. Yes, you’re right. The workflow is updated to make them clear. > 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? I personally think the name should be fine because even though we use in “LOG_TRUNCATION” state, the config is to allow us to know if users want to support unclean leader election, if so, we wait for all replicas to catch up. If we want to explicitly show the intention of the config, how about “mirror.full.replica.convergence.required”? I’m open to other suggestions. > 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? Hmm.. we didn’t think about the use case. But yes, if users only care about the state of a topic, they don’t need the info for other topics. I’ve combined with JR72.2, added “TopicPartitions” filed into DescribeClusterMirrorsRequest, and by default is null, which means all topic partitions under the requested mirrors will be returned. If set, we’ll query the mirrors and return the topic partitions that users expected. KIP is updated. > JR72.2 LastMirrorEpochLookups is a bit weird since the response returns either mirror state or mirror lag and LastMirrorEpochLookups doesn't match either. Yes, after the recent refactor, I think the LastMirrorEpochLookups can be merged into one level up, combining with the topic name filter in above comment, which will allow users to query down to the partition level. The LastMirrorEpochLookups will query along with normal mirror state by the readMirrorState from the coordinator. The difference is that after receiving the response, it will filter out the mirrors that don’t match the requested sourceClusterId, and then add the lastMirrorEpoch into DescribeClusterMirrorResponse. KIP is updated. > JR72.3 Should it return UNKNOWN_CLUSTER_MIRROR too? Right. Added. > JR72.4 "includeMirrorLags: source offset, destination offset, lag" DescribeClusterMirrorsRequest doesn't include those fields. Also, lag should be omitted. Sorry, I removed the “SourceOffset” and “DestinationOffset” fields from DescribeClusterMirrorsResponse because I added them into ReadMirrorOffsets (ex-ReadMirrorLags) API. But I forgot we still need them in the DescribeClusterMirrorsResponse so that the adminClient can receive the results. KIP is updated. > JR73. MirrorPartitionState record, LastMirrorEpochs record and MirrorPartitionStateRequest: The values for type and name are incorrect. Ah, that’s my bad. Fixed. > JR74. ListClusterMirrorsRequest > JR74.1 When is DesiredMirrorStateFilter used? There is no option to specify the desired state in the mirror-topic command. Yes, referred to kafka-consumer-groups.sh, I added the `--state` option to allow users to set the state when listMirror. > JR74.2 Should it return CLUSTER_MIRROR_AUTHORIZATION_FAILED too? Yes, KIP updated. > JR75. ReadMirrorLags > JR75.1 Should it be named ReadMirrorOffsets since the response doesn't include lag? Indeed. Updated. > JR75.2 Should it return UNKNOWN_CLUSTER_MIRROR too? Yes, added > JR75.3 Some places refer to this as ReadClusterMirrorLags. We need to be consistent. Thanks for pointing it out. KIP is updated. > JR76. WriteMirrorStates: Should it return UNKNOWN_CLUSTER_MIRROR too? Yes, we should validate the mirror name before writing the data. KIP is updated. > 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. It’s true, but there could be some delay between “actual state” and “desired state”. In step (2), we verify all partitions’ “actual state” are in STOPPED state. But the current desired state could be MIRRORING, it’s just that the mirror partition leaders haven’t got the metadata update yet. So, the controller still needs to verify the desired state is not in MIRRORING or PAUSED. I found we didn’t specify that the controller is actually verifying the “desired state”, not “actual state”. And this controller’s desired state validation also applies in startMirrorTopics/stopMirrorTopics/pauseMirrorTopics/resumeMirrorTopics. I’ve updated the KIP to make it clear. > 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. No, these are the tasks during STOPPING state, like what we described in “Stop Mirror Topics” workflow. If we do these in STOPPED state, we can’t safely move the readonly topic into writable because not all required tasks are done. I think I know why it confuses you, it’s because all the tasks are in the completed state. I’ve rephrased it to make it clear. > JR79. RPCs like DescribeClusterMirrors can take time. Should we add a timeout field? Make sense. Added into `DescribeClusterMirrors` request, as well as startMirrorTopics/stopMirrorTopics/pauseMirrorTopics/resumeMirrorTopics/deleteClusterMirrors because we will query the partition states from coordinator for state validation in these APIs. > JR80. PauseMirrorTopicsOptions: Should it support patterns? The command line seems to suppor it. Sorry, it’s a little confusing here. The “pattern” in startMirrorTopicsOptions or stopMirrorTopicsOptions is to update the “mirror.topics.include” and “mirror.topics.exclude” configuration for topic pattern discovery periodically. For the PauseMirrorTopics and ResumeMirrorTopics in the clusterMirrorCommand, users can provide regex in the “--topics” option, and the clusterMirrorCommand will listTopic in the destination cluster and match with the patterns users provided, and then feeds the matched topics into the PauseMirrorTopics/ResumeMirrorTopics requests. So in short, yes, PauseMirrorTopics/ResumeMirrorTopics supports regex in the ClusterMirrorComamnd, but we don’t need to add the “patterns”. > JR81. AdminClient > JR81.1 ListClusterMirrors: We need to add the option to specify different filters supported in the request. We already have the `--mirror` option. Added `--state` and `--sourceClusterId` options to support the other filters. > JR81.2 DescribeClusterMirrorsOptions.includeAuthorizedOperations: The request doesn't support this option. Could we add other options supported in the request? Currently, we have “includeAuthorizedOperations” in DescribeCluster, DescribeConsumerGroups, DescribeTopics, DescribeShareGroups, DescribeStreamsGroups requests. But none of them include the option in the tool to allow users to enable it. I think the reason is that we don’t want to show this to users on the tool output. But having this field, we allow the advanced users to invoke admin API with includeAuthorizedOperations enabled to get the results they want. > JR81.3 LeaderStateDescription: Should it be named MirrorPartitionStateDescription? Also, could we make the fields consistent with those in the response? OK. KIP is updated. Thank you, Luke On Thu, Aug 13, 2026 at 8:04 AM Jun Rao via dev <[email protected]> wrote: > 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 > >> >> >> >> >> > > > > >> >> >> >> >> > > >> >> >> >> >> > >> >> >> >> > > >> >> >> >> > >> >> >> > > >> >> >> > >> >> > > >> >> > >> > > >> > > >
