Hi Rajini, thanks for checking KIP updates.
>RS20: Does `mirror.metadata.refresh.interval.ms` control how often consumer
>offsets, ACLs etc. are sync'ed as well, in addition to topic metadata?
Yes, mirror.metadata.refresh.interval.ms controls how often ALL
metadata is synced, including topic configs, consumer offsets and
ACLs. To be precise, every broker syncs topic metadata, but only the
one hashing to the particular mirror name syncs topic configs, group
offsets, acls, and new new/deleted topics matching patterns. This
periodic metadata refresh process runs in a dedicated scheduler
thread.
>RS21: If we are supporting source versions without topic ids, how are we
>detecting if a source topic was deleted and recreated? Will we end up with
>mirror topics that contain records from both the old and new incarnation of
>the topic?
For pre-2.8 source clusters there is a fallback to name-based lookup
so the problem you mention can actually happen. I think there is no
solution to that gap if we want to support pre-2.8 releases. In that
case the operator needs to take extra care and manually stop the
mirror topic and delete the destination topic before re-creating it,
so mirroring starts fresh against the new incarnation. KIP updated.
>RS23: ConfigType currently has `CLUSTER_MIRRORS("cluster-mirrors")`. To
>match existing entries, should we change to `
>CLUSTER_MIRROR("cluster-mirrors")`?
Sure, KIP updated.
>RS24: "*Group ids with active members on the destination cluster are
>retrieved with a ListGroups RPC and skipped with a warning log*"
>Is this round trip request to group coordinator necessary? It isn't
>watertight anyway since a group could add a member after the RPC before the
>commit. Since group coordinators reject OffsetCommitRequest anyway if the
>group has members, do we need the extra RPC, potentially to every broker
>per commit cycle, to check state?
I agree, the pre-check could be removed. The alterConsumerGroupOffsets
would fail per-group with an appropriate error. That approach is both
simpler and more correct since the rejection happens atomically at the
coordinator. KIP updated.
>RS25: KIP says: "*During offset synchronization, the committed offset is
>clamped to the valid range of the destination partition using the formula
>max(dest log start offset, min(dest LEO, source committed offset))"*
>Does this mean that every offset commit operation needs to get log offsets
>of partitions from leader brokers?
No. The clamping reads LSO and LEO from the local ReplicaManager.
These are in-memory lookups against the local UnifiedLog instance, not
RPCs to leader brokers. This works because offset sync only runs on
the coordinator broker, and mirror topics are destination-local topics
whose log state is available on any broker that hosts a replica. If
the local broker doesn't host the partition (log is absent), that
partition is skipped.
>RS26: Is "mirror.admin.listener.name" only used to send requests to other
>brokers or is it used for broker->controller as well? From the discussion
>on this thread, it sounded like it was both. And that doesn't seem right
>since listeners can be different on brokers and controllers. For controller
>connections, we could perhaps use the same configs as existing
>broker->controller connections since we also rely on ClusterAction ACLs,
>e.g. for BumpLeaderEpochs.
We have dedicated channels with different listener configurations:
1. Controller requests (channelManager.sendRequest) uses the shared
broker-to-controller forwarding channel. This uses
control.plane.listener.name / the controller's listener, not
mirror.admin.listener.name.
2. Destination Admin client (dstAdmin) uses mirror.admin.listener.name
to find the broker's own advertised endpoint and security config for
broker-to-broker RPCs (offset commits, start/stop mirror topics).
>RS27: The KIP says that the minimum supported source version is 2.1. Which
>means we need to support source clusters without topic ids, which seems ok
>and the KIP describes this. It also means we need to support source
>clusters which don't support truncation-on-fetch. The KIP says: "*Mirror
>fetcher threads use the explicit OffsetsForLeaderEpochRequest path for log
>truncation rather than truncation-on-fetch (diverging epoch in fetch
>responses) for older version source clusters (Fetch API v11 or older). *"
>I don't think Fetch API v11 is sufficient to support truncation-on-fetch.
>If I remember correctly, we had some issues that were fixed later and the
>minimum version is 2.7. The KIP describes the truncation-on-fetch mode in
>detail, e.g. for handling unclean leader election, but not the older style
>*OffsetsForLeaderEpochRequest *path. How does the fetcher thread determine
>the mode
It doesn't dynamically detect the source version. Mirror threads
always start with the explicit OffsetsForLeaderEpochRequest path. But
diverging epochs in fetch responses are still honored when present.
We have two paths depending on the source cluster version:
Proactive (diverging epoch, source >= 2.7): The fetcher includes
lastFetchedEpoch in every fetch request. The source compares this
against its own epoch lineage. If the epochs diverge (ULE changed the
leader lineage), the source returns a divergingEpoch in the response
immediately, even if the fetcher's offset is still valid. Detection
happens on the very next fetch, regardless of offset position.
Reactive (OFFSET_OUT_OF_RANGE, source < 2.7): The source ignores
lastFetchedEpoch (doesn't understand the field). The only signal is
when the fetcher requests an offset that no longer exists on the
source because the ULE truncated the log below the fetcher's position.
If the fetcher is at offset 100 and the ULE truncated the source to
offset 50, the next fetch returns OFFSET_OUT_OF_RANGE and the fetcher
corrects. But if the ULE truncated to offset 90 and new records have
already been produced past 100, the fetcher sees no error and keeps
appending from the diverged log. The divergence is never detected.
In short, we can only support ULE with source cluster on 2.7+. KIP updated.
> - can it handle upgrade/downgrade of the source cluster?
Not explicitly. The initial truncation path is fixed. If a source
cluster upgrades mid-mirroring, the fetcher won't switch to
truncation-on-fetch for initial truncation.
>How will
>unclean leader election etc. differ? It will be good to understand the
>additional complexity for supporting pre-2.7 source clusters.
Supporting pre-2.7 means the ULE recovery path described in the KIP
(which relies on diverging epochs in fetch responses) simply won't
work for those sources. I think it's fine to explicitly document that.
>RS28: The semantics of the config
>`mirror.support.unclean.leader.election` seems
>odd. We perform unclean truncation below high watermark even when the
>config is true. For deployments where unclean leader election is disabled
>and likely to trigger alerts, it will be useful to have a config that
>prevents unclean truncation and moves the mirror topic to failed state.
>Have we considered just using a config `
>mirror.unclean.leader.election.enable=false` to prevent unclean truncation?
>Then `mirror.unclean.leader.election.enable=true` would transition the
>topic to ULE_RECOVERY. Not sure if the in-between state which truncates,
>but leaves the partition in MIRRORING state is that useful.
Good point. With your proposal, the operator gets a clear choice:
accept data loss from ULE (mirror.unclean.leader.election.enable=true)
or halt mirroring and move to FAILED (=false). The tradeoff is that
both paths may require manual recovery. When ULE is enabled,
ULE_RECOVERY waits for all assigned replicas to converge with no
timeout. A single stuck or decommissioned replica that never rejoins
ISR blocks mirroring for that partition indefinitely.
Let me discuss it with the team and we will update.
