Hi, Luke, Thanks for the reply. The KIP LGTM now.
Jun On Tue, Aug 18, 2026 at 1:08 AM Luke Chen <[email protected]> wrote: > Hi Jun, > > Thanks for the comments. > Answering them below. > > > JR80. AdminClient.startMirrorTopics(): Is includePatterns() optional? It > seems that mirroring can't start without it. > > Ah, right. I’ve moved the includePatterns from the > StartMirrorTopicsOptions into the required parameter in > AdminClient.startMirrorTopics(). > > > > 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. > Indeed, we added the “mirror” prefix because we were afraid of confusing > users like “mirror.fetch.min.bytes” and “replica.fetch.min.bytes”. But like > you said, these configurations are for CLUSTER_MIRROR resource type, not > for brokers. Users should be able to distinguish between them easily. > Removed the mirror prefix from the mirror-level configs. > > Thank you, > Luke > > > On Tue, Aug 18, 2026 at 2:04 AM Jun Rao via dev <[email protected]> > wrote: > >> 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!rUm8b3EcsLLsbBwkWG6hAiL-_Kard9D9TP_fPHp1g519R0HO7gRzlR4X5v2m4r58FdqsJdyCTqKHu14$> >> >> < >> 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 >> >> >> >> >> >> >> >> >> > > > >> >> >> >> >> >> >> >> >> > >> >> >> >> >> >> >> >> >> >> >> >> >> >> >> >> >> > >> >> >> >> >> >> >> >> >> >> >> >> >> >> >> > >> >> >> >> >> >> >> >> >> >> >> >> >> > >> >> >> >> >> >> >> >> >> >> >> > >> >> >> >> >> >> >> >> >> > >> >> >> >> >> >> >> > >> >> >> >> >> > >> >> >> > >> >
