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
> >> > > >
> >> >
> >>
> >
>
