Hi, Luke, Thanks for the reply.
JR80. AdminClient.startMirrorTopics(): Is includePatterns() optional? It seems that mirroring can't start without it. JR91. Mirror configs: Some of the configs are prefixed with "mirror," while others are not. Could we make them consistent? Since all those configs are for Mirror, the mirror prefix seems redundant. Jun On Mon, Aug 17, 2026 at 1:11 AM Luke Chen <[email protected]> wrote: > Hi Jun, > > Thanks for the good comments. > Answering them below. > > > JR80. In that case, it seems that there is no need for > AdminClient.startMirrorTopics() to take topics. The adminClient only uses > it to register the included topics, which can be passed in as > includePatterns (specific names or regex). The broker will use > StartMirrorTopicsRequest to set topic and topicId, but it doesn't go > through the AdminClient. > > Make sense. So when users invoke admin.startMirrorTopics, they only need > to set the topic patterns. And the receiving broker will resolve the > patterns into the actual topics using the listTopic response from the > source cluster, and get the topic id, numPartitions, … via the > describeTopics request from the source cluster if the topic is non-existent > in the destination cluster. KIP is updated. > > > > JR82. If “IncludeMirrorOffset” is disabled, > DescribeClusterMirrorsResponse.SourceOffset is still returned in the > response and what is its value in this case? > > It will be -1 because when the `IncludeMirrorOffset` is disabled, we won’t > query the mirror partition leader. I’ve updated the description to make it > clear. Thanks. > > > > JR88. Why is partition-load-time a Meter instead of a Gauge? > > You’re right. It is not the Meter type. It’s the SampledStat state like > “batch-size-avg” and “batch-size-max” in producer. I’ve updated the KIP. > > > > JR89. Why is thread-idle-ratio a Histogram? Typically, it's a Rate. > > You’re right. It’s the rate type. KIP is updated > > > > JR90. MetadataRefreshError : Is that cumulative? If so, it should be a > Meter. > Yes, it makes more sense to make it as Meter type. I’ve updated these > error-related metrics to Meter type: > 1. MetadataRefreshError > 2. TopicConfigMetadataSyncError > 3. ConsumerGroupOffsetSyncError > 4. ShareGroupOffsetSyncError > 5. AclSyncError > > > Thank you, > Luke > > > On Sat, Aug 15, 2026 at 8:10 AM Jun Rao via dev <[email protected]> > wrote: > >> Hi, Luke, >> >> Thanks for the reply. >> >> JR80. In that case, it seems that there is no need for >> AdminClient.startMirrorTopics() to take topics. The adminClient only uses >> it to register the included topics, which can be passed in as >> includePatterns (specific names or regex). The broker will use >> StartMirrorTopicsRequest to set topic and topicId, but it doesn't go >> through the AdminClient. >> >> JR82. If “IncludeMirrorOffset” is disabled, >> DescribeClusterMirrorsResponse.SourceOffset is still returned in the >> response and what is its value in this case? >> >> JR88. Why is partition-load-time a Meter instead of a Gauge? >> >> JR89. Why is thread-idle-ratio a Histogram? Typically, it's a Rate. >> >> JR90. MetadataRefreshError : Is that cumulative? If so, it should be a >> Meter. >> >> Jun >> >> On Fri, Aug 14, 2026 at 4:32 AM Luke Chen <[email protected]> wrote: >> >> > Hi Jun, >> > >> > Thanks for the good comments, >> > Answering them below. >> > >> > > JR80. PauseMirrorTopicsOptions: "PauseMirrorTopics/ResumeMirrorTopics >> > supports regex in the ClusterMirrorComamnd, but we don’t need to add the >> > “patterns”." >> > So, in this case, we use topics to support both specific topics and >> regex. >> > However, startMirrorTopicsOptions uses topics for specific topics and >> > includePatterns for regex. Could we make them consistent? >> > >> > >> > The purpose of the “includePatterns” field is for the >> > “mirror.topics.include” update, so that if there is any newly created >> topic >> > in the source cluster that matches the regex, we can create it. And the >> > “Topics” field in the StartMirrorTopicsRequest is to specify the topicId >> > and numPartitions in the source cluster. So I think “includePatterns” >> and >> > “Topics” both are necessary in the StartMirrorTopicsRequest. On the >> other >> > hand, in PauseMirrorTopics, because it won’t change the >> > “mirror.topics.include”/”mirror.topics.exclude” configs, we don’t need >> to >> > provide the regex in the request. >> > >> > >> > >> > > JR82. DescribeClusterMirrorsResponse.SourceOffset: Is the value also >> -1 >> > if >> > it's not included in the request? >> > >> > If “IncludeMirrorOffset” is disabled, we won’t query the mirror >> partition >> > leader, so we can’t get the sourceOffset. >> > >> > I assume you want to ask `if it’s still -1 if not actively mirroring, >> > while “IncludeMirrorOffset” is enabled`. That’s a good question. I think >> > our purpose of this command is to show the lag to users. So if the >> state is >> > not MIRRORING (ex: STOPPED/PAUSED), we think it’s not helpful to users >> to >> > show only the sourceOffset. Of course users can use >> `kafka-get-offsets.sh` >> > to get the offset if they really want to know. >> > >> > >> > > JR83. Start Mirror Topics: >> > > JR83.1 Should the controller verify the expected state for the topic? >> > >> > As replied in JR84, the desired state validation should happen in the >> > broker. And the controller validates the StateOffset. I’ve updated the >> > workflow to make it clear. >> > >> > > JR83.2 "d. truncates the log accordingly," >> > Could you add a description of the truncation offset? >> > >> > OK, added one more sentence: >> > d. based on the last mirror epoch response in (b), the broker translates >> > the epoch into the offset by querying the leaderEpochCache >> > (endOffsetForEpoch), and truncates the log accordingly, and waits for >> all >> > ISR members… >> > >> > >> > > JR84. Stop Mirror Topics: "The controller verifies that the desired >> state >> > of all topics are in MIRRORING or PAUSED state." >> > Should the broker verfy the topic level state? I thought the controller >> > will use StateOffset in the request to make sure the validation on the >> > broker is stil valid. If the controller performs the verification, is >> > StateOffset still useful? >> > >> > Oh, you’re right. We’re doing duplicated work here. Yes, the broker >> > already verifies the “desired state” of all topics are in MIRRORING or >> > PAUSED state, as well as the “actual state” is in MIRRORING or PAUSED >> > state. If all validation passes, we’ll forward it to the controller. So >> the >> > controller only needs to validate the StateOffset to make sure the >> > validation on the broker is still valid. So, yes, we don’t need the >> desired >> > state validation on the controller side. I’ve updated the KIP to make it >> > clear. >> > >> > One more thing to add is that, I’ve thought about not adding >> StateOffset, >> > and relying on the desired state validation to make sure we’re safe to >> do >> > the deletion or other operations. The problem is that, even if the >> broker >> > sees the desired state for topic A is STOPPED, and then the controller >> also >> > sees the desired state is STOPPED, it still cannot make sure the >> > controller’s STOPPED state equals the broker’s STOPPED state. For >> example: >> > 1. Broker sees STOPPED, then forward to controller >> > 2. User changes the desired state to MIRRORING >> > 3. User changes the desired state to STOPPED >> > 4. Controller sees the desired state is STOPPED >> > So I think StateOffset is required. >> > >> > Thanks for catching it BTW. >> > >> > >> > > JR85. There are still a few references to includeMirrorLags. They >> should >> > be >> > IncludeMirrorOffset. >> > >> > Sorry, just updated. >> > >> > >> > > JR86. Metrics: >> > > JR86.1 Could you describe the type (gauge, meter, etc) of each metric? >> > >> > KIP is updated. I also group the metrics into new added metrics and >> other >> > inherited metrics. >> > >> > > JR86.2 Could you describe a bit more what MinFetchRate represents? >> > >> > It means the lowest fetch request rate among all of this mirror fetcher >> > manager's fetcher threads. It inherits from the AbstractFetcherManager’s >> > metric. >> > >> > >> > > JR87. Version Compatibility: "For migration or DR use cases where >> > failback >> > is not required, this proposal supports source brokers up to version 2.1 >> > included" >> > This KIP needs topicId and KIP-516 was introduced later than 2.1. >> > Yes, when mirroring from a cluster without topic id support, we will >> > assign a random topic ID in the destination cluster and then fetch >> (without >> > topic id in the fetch request using the old version of Fetch API) as >> usual. >> > Of course this is for the one-way mirroring use case. It can’t start the >> > reverse mirror(failback) under this setup. You can check the >> “Compatibility >> > Matrix” section for more detail. >> > >> > >> > Thank you, >> > Luke >> > >> > On Fri, Aug 14, 2026 at 7:09 AM Jun Rao via dev <[email protected]> >> > wrote: >> > >> >> Hi, Luke, >> >> >> >> Thanks for the reply. >> >> >> >> JR80. PauseMirrorTopicsOptions: "PauseMirrorTopics/ResumeMirrorTopics >> >> supports regex in the ClusterMirrorComamnd, but we don’t need to add >> the >> >> “patterns”." >> >> So, in this case, we use topics to support both specific topics and >> regex. >> >> However, startMirrorTopicsOptions uses topics for specific topics and >> >> includePatterns for regex. Could we make them consistent? >> >> >> >> JR82. DescribeClusterMirrorsResponse.SourceOffset: Is the value also >> -1 if >> >> it's not included in the request? >> >> >> >> JR83. Start Mirror Topics: >> >> JR83.1 Should the controller verify the expected state for the topic? >> >> JR83.2 "d. truncates the log accordingly," >> >> Could you add a description of the truncation offset? >> >> >> >> JR84. Stop Mirror Topics: "The controller verifies that the desired >> state >> >> of all topics are in MIRRORING or PAUSED state." >> >> Should the broker verfy the topic level state? I thought the controller >> >> will use StateOffset in the request to make sure the validation on the >> >> broker is stil valid. If the controller performs the verification, is >> >> StateOffset still useful? >> >> >> >> JR85. There are still a few references to includeMirrorLags. They >> should >> >> be >> >> IncludeMirrorOffset. >> >> >> >> JR86. Metrics: >> >> JR86.1 Could you describe the type (gauge, meter, etc) of each metric? >> >> JR86.2 Could you describe a bit more what MinFetchRate represents? >> >> >> >> JR87. Version Compatibility: "For migration or DR use cases where >> failback >> >> is not required, this proposal supports source brokers up to version >> 2.1 >> >> included" >> >> This KIP needs topicId and KIP-516 was introduced later than 2.1. >> >> >> >> Jun >> >> >> >> >> >> On Thu, Aug 13, 2026 at 5:14 AM Luke Chen <[email protected]> wrote: >> >> >> >> > 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!t3M02ge9e-mRqsshxasG_K0qLi9sX21suoBd69nLUl_gWUvn9ZLLekJiQspa_q3gTVSzWyUQBQfQLHk$> >> >> < >> https://urldefense.com/v3/__https://cwiki.apache.org/confluence/spaces/KAFKA/pages/255070434/KIP-932*Queues*for*Kafka*KIP932*3AQueuesforKafka-WriteShareGroupStateAPI__;KysrIyU!!Ayb5sqE7!rX3-NwDU3s0U7POQYH1R-DvASnCPTTtFRWxZTVaS-PQHg-C1TAcu9xMe3MQF4fIXGO67WKOqrlz-lc8$ >> > >> >> >> < >> >> >> https://urldefense.com/v3/__https://cwiki.apache.org/confluence/spaces/KAFKA/pages/255070434/KIP-932*Queues*for*Kafka*KIP932*3AQueuesforKafka-WriteShareGroupStateAPI__;KysrIyU!!Ayb5sqE7!u8EkPGXm8_VKdltXy8lrIW14mR2kgYIoaBPD7LO2uAN6rjhrfCX37Pfa6r85QPpSwSPh4V6eW_bKlns$ >> >> > >> >> >> > < >> >> >> >> >> >> 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 >> >> >> >> >> >> >> >> > > > >> >> >> >> >> >> >> >> > >> >> >> >> >> >> >> >> >> >> >> >> >> >> >> > >> >> >> >> >> >> >> >> >> >> >> >> >> > >> >> >> >> >> >> >> >> >> >> >> > >> >> >> >> >> >> >> >> >> > >> >> >> >> >> >> >> > >> >> >> >> >> > >> >> >> > >> >
