Hello everyone, Thank you all for the thoughtful questions and suggestions, and thanks to Michael Maison for proposing a better approach to the chained mirroring problem. Some of these required careful consideration and led us to refine the design.
For reviewers, I recommend focusing on the paragraphs with significant changes: 1. MirrorFetcherThread (updated) 2. Security Control (updated) 3. Idempotent Producer (updated) 4. Command Workflows (new) 5. Error Names (new) Below are answers to the outstanding questions. AS6, AS7: I added a table to "Security Control" illustrating authorization requirements for each component. Let me know if it covers what you had in mind. AS8: I removed that field for now. It is on my backlog and I will post an update soon. AS9, AS17: There is a new "Errors" paragraph with a table listing new and reused error codes, along with the RPCs that use each one. AS18: Could you clarify which LeaderEpoch checks you are referring to? Here is our understanding: 1. WriteMirrorStatesRequest: The coordinator validates that the current leader epoch matches the request leader epoch, rejecting outdated requests. 2. ReadMirrorStatesResponse: The coordinator includes the current leader epoch in the response, and the leader node validates it against its own current leader epoch. 3. LastMirroredOffsetsResponse: We replaced this with LastMirroredEpochsResponse, so we believe this no longer applies. AS22: Description updated to: "This property is filtered out from DescribeConfigs responses to avoid exposing internal state to users." --- RS1: I added the DeleteMirror RPC and updated all relevant sections. RS5: As you suggested, I introduced a new ResourceType called ClusterMirror. See "Security Control." RS6: The refactoring suggestions make sense, but we need more time to think them through. RS9: Until we support tiered storage, we will mirror all records including data in remote storage into the destination cluster. RS10: There is a future improvement briefly described in "Bandwidth Control." Let me know if you need more details. --- JR3: I created a new "Command Workflows" section that covers the workflow for each mirror operation, including failover and failback. Let me know if you spot any issues. I hope I have addressed everything, but please let me know if I missed any question. Cheers Fede On Fri, Mar 20, 2026 at 5:12 AM Luke Chen <[email protected]> wrote: > > Hi Jun, > > Thanks for the review comments. > Answer them below. > > > JR1. "Epoch rewriting: When records are appended to the destination log, > the batch epochs are rewritten to match the destination cluster's leader > epochs, maintaining consistency within the destination cluster." This has a > couple of impacts. > > We have an updated design to support unclean leader election in this doc, > where we will NOT rewrite the leader epoch anymore. > https://cwiki.apache.org/confluence/display/KAFKA/Unclean+Leader+Election+in+Cluster+Mirroring > > > > JR2. "Tiered Storage is not initially supported": Ideally, we should > support tiered storage. Same as RS9, the destination cluster issues > consumer requests, which support tiering. > The OffsetMovedToTieredStorageException is used by replication > inside AbstractFetcherThread. This suggests that it's probably not a good > fit for cluster mirroring to use AbstractFetcherThread. > > Yes, thanks for pointing this out. I was wrong about this. Before we > support tiered storage, the mirroring will mirror all records including > data in the remote storage into the destination cluster. > > > JR3. For each new cluster mirroring command, it would be useful to > document > the underlying workflow (e.g, which RPC requests are issued, to which node; > what records are written to metadata topic, or internal topic, which > actions are triggered on the broker, etc). > > Will do. > > > JR4. Truncating a log to LMO. Currently, there is no internal API that > truncates a partition from the end. Could you describe how this will be > implemented to ensure all replicas are consistent after the truncation? > > The truncation flow is like this: > 1. When the MirrorMetadataManager in the node gets notified about the > partition leader assignment when onMetadataUpdate (via TopicsDelta), it'll > query the mirror coordinator about mirror partition state. > 2. When it's the PREPARING state, the MirrorMetadataManager in the leader > node will get the last mirrored offset (or epoch) from the source cluster > (new API) and then do the log truncate. > 3. In (2), we'll also register a callback in Partition instance, and wait > until all ISRs complete the truncation by checking the follower replica's > LEO. > 4. In (3), this check will be invoked every time the leader node update > follower fetch state, like how we check if high watermark should be > incremented. > 5. After all ISRs complete the truncation, we'll invoke the callback and > move the mirror partition state to MIRRORING, and then start fetching data > from the source cluster. > > Note: > (1) In PREPARING state, the partition is READ-ONLY, so there will no any > data written in the leader node > (2) During step (1) ~ (4), if any leadership change happens, the new leader > will start from step (1) to complete the log truncation process. > (3) If unclean leader election is supported > (i.e. mirror.support.unclean.leader.election=true), then we'll wait until > ALL registered replicas complete the truncation before moving on to > MIRRORING state. > > We'll update the KIP in the following days to address the community > feedback. Some questions need more thinking. > Please give us some time. :) > > Thank you, > Luke > > > On Fri, Mar 20, 2026 at 9:10 AM Jun Rao via dev <[email protected]> > wrote: > > > Hi, Federico, > > > > Thanks for the KIP. A few comments. > > > > JR1. "Epoch rewriting: When records are appended to the destination log, > > the batch epochs are rewritten to match the destination cluster's leader > > epochs, maintaining consistency within the destination cluster." This has a > > couple of impacts. > > JR1.1 How do we ensure that the leader epoch in committed offsets is > > consistent with the leader epoch in the batch? This consistency is > > important when the consumer fails over to a different cluster. It seems the > > KIP doesn't translate the leader epoch when mirroring the comitted offsets. > > JR1.2 Typically, leader epochs increase monotonically in the log. Do we > > ensure this remains the case after failover and failback? > > > > JR2. "Tiered Storage is not initially supported": Ideally, we should > > support tiered storage. Same as RS9, the destination cluster issues > > consumer requests, which support tiering. > > The OffsetMovedToTieredStorageException is used by replication > > inside AbstractFetcherThread. This suggests that it's probably not a good > > fit for cluster mirroring to use AbstractFetcherThread. > > > > JR3. For each new cluster mirroring command, it would be useful to document > > the underlying workflow (e.g, which RPC requests are issued, to which node; > > what records are written to metadata topic, or internal topic, which > > actions are triggered on the broker, etc). > > > > JR4. Truncating a log to LMO. Currently, there is no internal API that > > truncates a partition from the end. Could you describe how this will be > > implemented to ensure all replicas are consistent after the truncation? > > > > Jun > > > > On Mon, Mar 16, 2026 at 2:44 AM Federico Valeri <[email protected]> > > wrote: > > > > > I mentioned a corner case in the chained mirroring use case. Let me > > > clarify what I mean with a simple example: > > > > > > 1. B is fetching from A, and C is fetching from B (A --> B --> C) > > > 2. A producer with PID 5 sends records to A > > > 3. Failover happens and B becomes writable (A -x-> B --> C) > > > 4. A different producer with PID 5 sends records to B > > > 5. Collision on cluster C (two different producers mapped to PID -7 in C) > > > > > > (arrows represent data flowing, not fetch direction) > > > > > > > > > On Sun, Mar 15, 2026 at 7:14 PM Federico Valeri <[email protected]> > > > wrote: > > > > > > > > Hi Rajini, thanks for your thoughtful review and for catching a few > > > > bugs. I'll skip some questions that we will address later. > > > > > > > > RS2: The metadata records are described in "Mirror Metadata Records" > > > > paragraph. Currently there are only two records: "LastMirroredOffsets" > > > > record tracks the latest successfully mirrored offset for each > > > > partition, while "MirrorPartitionState" record represents the > > > > lifecycle states of a mirrored partition. > > > > > > > > RS3: That's a good point that was also raised by Andrew. It was an > > > > easy solution that we used for our prototype, but we need to think > > > > about a better solution. > > > > > > > > RS4: Current design is that mirror fetcher threads behaves like a > > > > read_committed consumer fetching up to "source LSO". On failover we > > > > truncate destination log to "local LSO". > > > > > > > > The approach of fetching up to HW that you propose is still safe as we > > > > keep truncating to local LSO on failover, but it trades lower > > > > steady-state lag (especially when long-running transactions exist on > > > > the source) for more data loss on failover (the net data loss relative > > > > to the source is the same in both approaches). In other words, with > > > > your approach we fetch more data that we may then need to truncate. > > > > Also, read_uncommited consumers on the destination cluster would be > > > > able to read records that may be truncated on failover. These are just > > > > my consideration, but we are open to discussion on which is the best > > > > approach here. > > > > > > > > When a failover is triggered (RemoveTopicsFromMirror), the sequence is: > > > > > > > > 1. Partitions transition to STOPPING state > > > > 2. Fetchers are removed > > > > 3. For each partition, truncate to local LSO is called > > > > 3.1. Reads LSO from each partition's local log > > > > 3.2. Calls log.truncateTo(offset) on the UnifiedLog > > > > 3.3. Ensures ISR members complete truncation before the partition > > > > becomes writable > > > > 4. For each partition, the LSO is recorded as the last mirrored offset > > > > (LMO) in __mirror_state > > > > 5. Partitions transition to STOPPED and become writable > > > > > > > > When a failback is triggered (AddTopicsToMirror), the sequence is: > > > > > > > > 1. Partitions transition to PREPARING state > > > > 2. For each partition, truncation to LMO is called > > > > 2.1. This sends a LastMirroredOffsetsRequest to the source cluster > > > > to fetch the offsets that were recorded during the previous failover > > > > 2.2.a. The response offsets are used to truncate local logs > > > > 2.2.b. If the source cluster doesn't support the LastMirroredOffsets > > > > API or first-time mirror, it truncates to offset 0 > > > > 3. Partitions transition to MIRRORING > > > > > > > > RS7: Can you point me to the section that says configs are stored in > > > > __mirror_state? Mirror connection configs (bootstrap servers, > > > > credentials, etc.) are stored in KRaft metadata via > > > > ConfigResource.Type.MIRROR, not in __mirror_state. The internal topic > > > > only stores partition states and last mirrored offsets. Sensitive > > > > credentials follow the existing KRaft config handling, which is > > > > already protected by controller/broker access controls and sensitive > > > > config redaction in DescribeConfigs responses. > > > > > > > > RS8: Not sure what's the recommended approach here. Adding a new error > > > > code does not change the response schema and older clients that don't > > > > recognize the new error code will surface it as an > > > > UnknownServerException (non-retriable). > > > > > > > > RS11: Good catch. This is a prototype simplification that we need to > > > > address. To properly sync consumer group offsets, the implementation > > > > would need to send ListGroups to all source brokers (or use the > > > > AdminClient which does this internally), send FindCoordinator to > > > > discover the group coordinator for each group, send OffsetFetch to the > > > > correct group coordinator. > > > > > > > > RS12: You are absolutely right, the transformation is not idempotent, > > > > so it is not safe for chained mirroring (A -> B -> C). Instead, > > > > round-trip mirroring (A -> B, then B -> A) works because, when doing a > > > > failback, the log is truncated before mirroring resumes, so previously > > > > mirrored records with negative pids are removed and the transformation > > > > is only applied to new records produced natively on that cluster > > > > (double-transformation never occurs). Non-transactional batches stay > > > > at -1 [ -(-1 + 2) = -(1) = -1], which is correct. > > > > > > > > The chained mirroring would work if we skip the transformation when > > > > pid is negative, but there is still an edge case: A -> B -> C with > > > > local B producer. If cluster A has local pid 5 and cluster B also has > > > > local pid 5, both end up as -7 on cluster C. Collision: two different > > > > producers with the same pid on the destination. No pid-only > > > > transformation can solve that. We would need to incorporate cluster > > > > identity. > > > > > > > > Possible solution that would handle any topology: The producer IDs are > > > > 64-bit signed longs used to identify a producer. The clusterId (UUID) > > > > is a globally 128-bit unique identifier for each source cluster. We > > > > could use the clusterId hash to partition the entire negative PID > > > > space into regions, one per source cluster. Basically we divide the 64 > > > > bits into three fields: bit 63 (sign bit), bits 62-31 (region > > > > selector), bits 30-0 (producer identity). Once a non-negative PID is > > > > mapped to a region, it passes through unchanged no matter how many > > > > hops follow (i.e. we apply the transformation only for PIDs >= 0). > > > > > > > > Example with two clusters: > > > > > > > > - Bit 63: This is the sign bit that makes the value negative and > > > > distinguishes mirrored pids from local ones (which are non-negative). > > > > - Bits 62-31 cluster A: clusterId = "abc-123", clusterHash = 42 > > > > - Bits 62-31 cluster B: clusterId = "xyz-789", clusterHash = 99 > > > > - Bits 30-0: Local producer ID 5 that is the same on both clusters. > > > > > > > > A's pid 5 --> > > > > 1|00000000000000000000000000101010|0000000000000000000000000000101 > > > > B's pid 5 --> > > > > 1|00000000000000000000000001100011|0000000000000000000000000000101 > > > > > > > > On Wed, Mar 11, 2026 at 1:23 PM Rajini Sivaram < > > [email protected]> > > > wrote: > > > > > > > > > > A few more questions about the KIP for clarification: > > > > > > > > > > RS8: The KIP says produce requests to mirror topics will throw > > > > > ReadOnlyTopicException. For Produce Requests returning a new error to > > > > > clients, don’t we need to bump Produce request version? > > > > > > > > > > RS9: The KIP says we use OffsetMovedToTieredStorageException to > > prevent > > > > > mirroring of data in tiered storage. But doesn’t the mirror client > > look > > > > > like a regular consumer to the source cluster and return records > > > fetched > > > > > from tiered storage? > > > > > > > > > > RS10: Client-id based quotas for the source cluster look hard to > > manage > > > > > since there is no hierarchy or grouping possible. Seems better to > > rely > > > on > > > > > secure user-principal based quotas on the source-side. > > > > > > > > > > RS11: The KIP says `The manager maintains a connection pool with one > > > > > blocking sender per source cluster`. If this is the connection used > > for > > > > > periodic sync of offsets, topic configs etc. the coordinator is > > likely > > > to > > > > > need connections to all source brokers (i.e. all group coordinators). > > > > > > > > > > RS12: The KIP proposes to transform producer ids for mirror records > > to > > > > > avoid conflicts. This comes at a cost because CRC checksum needs to > > be > > > > > recomputed. To justify this cost, we need to ensure that this > > > > > transformation works in all cases. What happens if you are mirroring > > a > > > > > mirror topic? Is that a supported scenario? Or mirroring back > > mirrored > > > data > > > > > during failback because the source was truncated? > > > > > > > > > > > > > > > > > > > > Thanks, > > > > > > > > > > Rajini > > > > > > > > > > > > > > > On Tue, Mar 10, 2026 at 8:19 PM Rajini Sivaram < > > > [email protected]> > > > > > wrote: > > > > > > > > > > > Hi team, > > > > > > > > > > > > Thanks for the KIP! I have a few questions, mostly clarification at > > > this > > > > > > point. > > > > > > > > > > > > RS1: There is a `CreateMirror` request but no corresponding > > > `DeleteMirror` > > > > > > request. Is that intentional? > > > > > > > > > > > > > > > > > > > > > > > > RS2: It will be good to define the format of data going into the > > > internal > > > > > > mirror state topic. There is an example under kafka-dump-logs, > > which > > > > > > shows partition-level state in the payload and the mirror name as > > > key. I > > > > > > guess that is not what we expect it to be. Do we delete this > > > information > > > > > > when a topic is deleted or a mirror is deleted? > > > > > > > > > > > > > > > > > > > > > > > > RS3: KIP currently says mirror name cannot end with .removed. I > > > guess it > > > > > > cannot also end with .paused. Have we considered storing state and > > > > > > mirror name separately, but updated together for a topic? Since new > > > > > > states may be added in future, name restrictions may become hard to > > > > > > implement. > > > > > > > > > > > > > > > > > > > > > > > > RS4: The KIP says *“mirroring must fetch only up to the LSO to > > > maintain > > > > > > transactional consistency”* and it also says *“During the mirror > > > stopping > > > > > > transition, the MirrorCoordinator performs a log truncation > > > operation that > > > > > > resets each mirror partition to its LSO.”* I guess the plan is to > > > fetch > > > > > > up to high watermark and truncate to locally computed LSO on > > > failover? > > > > > > Details of the sequence here will be useful. How does > > > MirrorCoordinator > > > > > > perform truncation? > > > > > > > > > > > > > > > > > > > > > > > > RS5: The KIP says “*On the destination cluster, mirror-related > > > operations > > > > > > (creating mirrors, adding/removing topics from mirrors, managing > > > mirror > > > > > > configurations) require the CLUSTER_ACTION permission on the > > cluster > > > > > > resource.*” The `Cluster:ClusterAction` ACL is currently used for > > > broker > > > > > > service account, e.g. local replication is authorized using this. > > It > > > seems > > > > > > odd to grant this permission to users managing a resource on the > > > cluster. > > > > > > Have we considered adding a new resource type `ClusterMirror` and > > > define > > > > > > ACLs like `ClusterMirror:Create`, `ClusterMirror:Alter` and ` > > > > > > ClusterMirror:AlterConfigs`? > > > > > > > > > > > > RS6: The KIP talks about three entities: Cluster Mirror, Mirror > > > Topic and Mirror > > > > > > Partition, with Cluster Mirroring as the feature name. Since we > > > already > > > > > > have MirrorMaker that also refers to mirrors, it will be nice if we > > > can > > > > > > refer to the entities using their full name in the CLI and public > > > APIs. > > > > > > That will enable us to add more mirror topic and mirror partition > > > APIs in > > > > > > the future if needed. For example: > > > > > > > > > > > > > > > > > > - `kafka-cluster-mirrors.sh` to manage cluster mirrors > > > > > > - createClusterMirrors(), listClusterMirrors(), > > > > > > describeClusterMirrors() etc on the Admin API and Kafka > > Protocol. > > > > > > - KIP proposes pauseMirrorTopics(), resumeMirrorTopics() which > > > are > > > > > > good. > > > > > > > > > > > > > > > > > > > > > > > > RS7: The KIP proposes to store mirror configs in the internal > > mirror > > > state > > > > > > topic. This includes sensitive credentials of another cluster. Have > > > we > > > > > > considered other options? Can a user with read access read the data > > > from > > > > > > the state topic using a consumer? > > > > > > > > > > > > > > > > > > > > > > > > Thanks, > > > > > > > > > > > > > > > > > > Rajini > > > > > > > > > > > > > > > > > > > > > > > > On Sun, Mar 8, 2026 at 8:58 PM Andrew Schofield < > > > [email protected]> > > > > > > wrote: > > > > > > > > > > > >> Hi Fede and friends, > > > > > >> I've re-read in detail and have quite a lot of comments, mostly > > > minor > > > > > >> clarifications, but as it approaches a vote, it's good to get the > > > details > > > > > >> nailed down. > > > > > >> > > > > > >> AS6: Could we have a diagram which shows which RPCs are served by > > > which > > > > > >> components? This will help illustrate the authorisation > > > requirements for > > > > > >> the various components, which is an aspect of the KIP that I don't > > > think is > > > > > >> completely specified yet. > > > > > >> > > > > > >> AS7: Please could you include a table of the operations and > > > resources > > > > > >> which will be checked for authorisation of each of the RPCs > > > introduced. > > > > > >> Also, please could you document the permissions which the > > > destination > > > > > >> cluster will require to mirror data and ACLs (for example, I think > > > it will > > > > > >> need ALTER on the CLUSTER resource to manipulate ACLs)? It's going > > > to need > > > > > >> Metadata, DescribeConfigs, DescribeAcls, ListGroups, OffsetFetch, > > > > > >> LastMirrorOffset and Fetch RPCs I think, possibly others too. The > > > user is > > > > > >> probably going to want to give as little permission as possible to > > > the > > > > > >> destination cluster to get its job done. > > > > > >> > > > > > >> AS8: You include AuthorizedOperations in DescribeMirrorsResponse, > > > but I > > > > > >> don't know what the operations are. I think the implies MIRROR is > > a > > > new > > > > > >> resource type in the Kafka security model and DescribeMirrors can > > > be used > > > > > >> to enquire the authorised operations for the client making the > > > Admin API > > > > > >> request. > > > > > >> > > > > > >> AS9: I think you're going to need some new error codes in the > > Kafka > > > > > >> protocol, as least: > > > > > >> > > > > > >> * INVALID_MIRROR_NAME or similar if the mirror name doesn't meet > > the > > > > > >> rules for a topic name > > > > > >> * UNKNOWN_MIRROR if the mirror doesn't exist > > > > > >> > > > > > >> And probably some more for logical inconsistencies such as this > > > topic > > > > > >> isn't in that mirror, that topic is already in another mirror, and > > > so on. > > > > > >> > > > > > >> AS10: Could you add the usage information for kafka-mirrors.sh > > (the > > > > > >> intended output from kafka-mirrors.sh --help) so all of the > > options > > > are > > > > > >> documented together? For example, I see that --replication-factor > > is > > > > > >> included in one of the examples, which seems a bit surprising and > > > I'm not > > > > > >> sure whether it's a mistake or a feature. I can probably use > > > --describe > > > > > >> with a specific --mirror but it's not specified. > > > > > >> > > > > > >> AS11: I would expect the signature for Admin.addTopicsToMirror to > > be > > > > > >> Admin.addTopicsToMirror(String mirrorName, Set<String> topics, > > > > > >> AddTopicsToMirrorOptions options) because it's for adding topics > > to > > > a > > > > > >> mirror, as the counterpart to Admin.removeTopicsFromMirror(String > > > > > >> mirrorName, Set<String> topics, RemoveTopicsFromMirrorOptions > > > options). > > > > > >> > > > > > >> AS12: I don't think ignorable RPC fields in version 0 RPCs make > > > sense > > > > > >> because they're not trying to be compatible with a previous > > version. > > > > > >> > > > > > >> AS13: I would have expected AddTopicsToMirrorRequest to have > > mirror > > > name > > > > > >> above the list of topics because the same mirror name applies to > > > all of the > > > > > >> topics being added. As specified, you repeat the mirror name for > > > all of the > > > > > >> topics. > > > > > >> > > > > > >> AS14: I suggest adding ErrorMessage to the responses in all cases > > > to make > > > > > >> it easier to give more descriptive exception messages than just > > the > > > default > > > > > >> for the error codes. > > > > > >> > > > > > >> AS15: I may have the wrong end of the stick here, but I expected > > > > > >> RemoveTopicsFromMirrorRequest to remove the topics from a specific > > > named > > > > > >> mirror as implied by the example of the kafka-mirrors.sh command. > > > In fact, > > > > > >> I was expecting the mirror to contain the topics in the admin RPC > > > requests > > > > > >> and responses, and that's only true for about half of them. > > > > > >> > > > > > >> AS16: Can I change the mirror.name config using > > > IncrementalAlterConfigs? > > > > > >> If I attempt it, what's the error? > > > > > >> > > > > > >> AS17: If I attempt mirror RPCs when the mirror is in the wrong > > > state, the > > > > > >> error is specified as INVALID_REQUEST. That's usually kept for > > > badly formed > > > > > >> requests, as opposed to logically invalid ones. Maybe > > > MIRROR_NOT_STOPPED or > > > > > >> MIRRORING_ACTIVE or similar would be more expressive. > > > > > >> > > > > > >> AS18: Should the LastMirroredOffsetsResponse, > > > ReadMirrorStatesResponse > > > > > >> and WriteMirrorStatesRequest include LeaderEpoch? I suspect so. > > > > > >> > > > > > >> AS19: In DescribeMirrorsResponse, I suspect you will want "null" > > > values > > > > > >> for some fields which don't have values during initialisation and > > > so on, > > > > > >> such as lag. > > > > > >> > > > > > >> AS20: Do you need to add new versions of the DescribeConfigs and > > > > > >> IncrementalAlterConfigs RPCs to support mirror resources? > > > > > >> > > > > > >> AS21: The topic configuration > > mirror.replication.throttled.replicas > > > is > > > > > >> described as a list, but the default is MAX_LONG. > > > > > >> > > > > > >> AS22: By including mirror.name as a topic config, a client which > > > has > > > > > >> permission to describe configs for the topic is able to discover > > > the name > > > > > >> of the mirror, whether they are permitted to list the mirrors or > > > describe > > > > > >> that particular mirror. Generally, the Kafka authorisation model > > > does not > > > > > >> permit this kind of unauthorised information disclosure. For > > > example, when > > > > > >> a client describes the committed offsets for a consumer group, the > > > list of > > > > > >> topics returned is filtered to only those topics which the client > > is > > > > > >> permitted to describe, even though that may results in an > > > incomplete set of > > > > > >> topic partitions being returned. Is there an alternative way in > > > which this > > > > > >> information could be stored so Kafka only reveals mirror > > > information to > > > > > >> principals authorised to see it? > > > > > >> > > > > > >> AS23: I observe that there are situations in which a `.removed` > > > suffix is > > > > > >> added to the mirror name. Is it permitted for the user to define a > > > mirror > > > > > >> called "my.nasty.mirror.removed" and does it break anything? > > > > > >> > > > > > >> > > > > > >> Thanks, > > > > > >> Andrew > > > > > >> > > > > > >> On 2026/03/06 13:41:52 Paolo Patierno wrote: > > > > > >> > Hi Fede, > > > > > >> > something more ... > > > > > >> > > > > > > >> > Is there any migration path for users who want to migrate from > > > using > > > > > >> Mirror > > > > > >> > Maker 2 to the cluster mirroring? > > > > > >> > I mean, something like a tool useful to create a corresponding > > > cluster > > > > > >> > mirroring configuration starting from a MM2 one. Nothing that > > > runs the > > > > > >> > migration automatically but something that can be provided to > > the > > > users > > > > > >> as > > > > > >> > output to be validated and put in place by them. > > > > > >> > > > > > > >> > The Admin Client is missing methods to pause and stop mirroring > > > (but we > > > > > >> > have corresponding protocol messages). Is it on purpose? Any > > > specific > > > > > >> > reasons? They would be important from an automatic operator > > > perspective > > > > > >> use > > > > > >> > case. > > > > > >> > Also a method to provide the LastMirroredOffset from the source > > > cluster > > > > > >> > could be useful for progress and tracking purposes. > > > > > >> > Finally, what about a method to get the mirror states? I don't > > > think the > > > > > >> > describe method provides such information. > > > > > >> > In general, I think that the Admin Client section needs to cover > > > in more > > > > > >> > details the new classes definition like CreateMirrorOptions, > > > > > >> > CreateMirrorResult, ... and so on for all the defined new > > methods. > > > > > >> > > > > > > >> > > AddTopicsToMirrorResult addTopicsToMirror(Map<String, String> > > > > > >> > topicToMirrorName, AddTopicsToMirrorOptions options); > > > > > >> > > > > > > >> > Isn't it missing the mirrorName (as you have in the > > > > > >> removeTopicsFromMirror > > > > > >> > counterpart)? > > > > > >> > What's the topicToMirrorName parameter if it's defined as a Map? > > > The > > > > > >> method > > > > > >> > is also plural using "topics" so comparing to the > > > removeTopicsFromMirror > > > > > >> > method, I would assume the parameter really is Set<String> > > topics? > > > > > >> > Comparing to the corresponding protocol message > > > > > >> AddTopicsToMirrorRequest, I > > > > > >> > see a list of topics but each of them has id, name and > > > corresponding > > > > > >> > mirror. So it's unclear how the addTopicsToMirror is defined. > > > > > >> > > > > > > >> > > RemoveTopicsFromMirrorResult removeTopicsFromMirror(String > > > mirrorName, > > > > > >> > Set<String> topics, RemoveTopicsFromMirrorOptions options); > > > > > >> > > > > > > >> > This method gets a mirrorName but if I look at the corresponding > > > > > >> protocol > > > > > >> > message RemoveTopicsFromMirrorRequest, it says "Allows users to > > > detach > > > > > >> > topics from their associated mirror" so the mirror is actually > > not > > > > > >> provided > > > > > >> > and it's exactly what I see in the JSON definition (only topics > > > list > > > > > >> with > > > > > >> > id and name). > > > > > >> > > > > > > >> > Finally, regarding the protocol change: > > > > > >> > > > > > > >> > * ListMirrorsResponse I would add the clusterId in the JSON > > > definition > > > > > >> > (it's related to my comments in the previous email when using > > the > > > tool). > > > > > >> > * WriteMirrorStatesRequest has the following in the JSON which > > > should > > > > > >> not > > > > > >> > be part of it "{ "name": "RemovedTopics", "type": "[]string", > > > > > >> "versions": > > > > > >> > "0+", "about": "The topic names to be removed." }" > > > > > >> > > > > > > >> > Thanks, > > > > > >> > Paolo. > > > > > >> > > > > > > >> > On Fri, 6 Mar 2026 at 13:08, Paolo Patierno < > > > [email protected]> > > > > > >> > wrote: > > > > > >> > > > > > > >> > > Hi Fede, > > > > > >> > > thank you for the proposal. I had a first pass with following > > > > > >> thoughts and > > > > > >> > > questions. > > > > > >> > > > > > > > >> > > > When the unclean.leader.election.enable is set to true, the > > > broker > > > > > >> will > > > > > >> > > log a warning at every configuration synchronization period. > > > > > >> > > Be more explicit about what the warning says. > > > > > >> > > > > > > > >> > > > This topic ID is not used by other topics in the current > > > cluster > > > > > >> > > In such a case, which should be very unlikely, what's going to > > > happen? > > > > > >> > > Isn't it possible to mirror the topic? > > > > > >> > > > > > > > >> > > > To enable it, all cluster nodes (controllers and brokers) > > must > > > > > >> > > explicitly enable unstable API versions and unstable feature > > > versions > > > > > >> in > > > > > >> > > all configuration files. After starting the cluster with a > > > minimum > > > > > >> metadata > > > > > >> > > version, operators can dynamically enable the mirror version > > > feature > > > > > >> to > > > > > >> > > activate Cluster Mirroring. > > > > > >> > > AFAIU there is going to be a dedicated feature flag for it, > > > right? If > > > > > >> yes > > > > > >> > > can we state it clearly also specifying the exact name (i.e. > > > > > >> mirror.version > > > > > >> > > or something similar)? > > > > > >> > > > > > > > >> > > When running the kafka-mirrors.sh tool to list the mirrors, > > > other than > > > > > >> > > showing the SOURCE-BOOTSTRAP, it could be useful to have also > > > the > > > > > >> clusterId > > > > > >> > > which, as a unique identifier, could be helpful in automated > > > systems > > > > > >> using > > > > > >> > > the cluster mirroring. Of course, it would be important to > > have > > > in the > > > > > >> > > ListMirrorsResponse as well as an additional field. > > > > > >> > > > > > > > >> > > What happens in case of Kafka downgrade from a version > > > supporting > > > > > >> > > mirroring to an older one not supporting it. > > > > > >> > > The mirror won't be running but the topic configuration will > > > still > > > > > >> have > > > > > >> > > config parameters like mirror.name and so on, right? Are they > > > just > > > > > >> > > ignored by the older Kafka version and the cluster will work > > > without > > > > > >> issues? > > > > > >> > > > > > > > >> > > Thanks, > > > > > >> > > Paolo > > > > > >> > > > > > > > >> > > On Fri, 6 Mar 2026 at 10:43, Luke Chen <[email protected]> > > > wrote: > > > > > >> > > > > > > > >> > >> Hi Andrew and all, > > > > > >> > >> > > > > > >> > >> About AS5, yes, I've created a sub-document > > > > > >> > >> < > > > > > >> > >> > > > > > >> > > > > > https://cwiki.apache.org/confluence/display/KAFKA/Unclean+Leader+Election+in+Cluster+Mirroring > > > > > >> > >> >to > > > > > >> > >> explain the algorithm to support unclean leader election in > > > cluster > > > > > >> > >> mirroring. > > > > > >> > >> Thanks for your comments, I'm inspired by that! :) > > > > > >> > >> > > > > > >> > >> About your idea, to store the owner of the leader epoch when > > > > > >> leadership > > > > > >> > >> change, I think it might not be needed because the most > > > important > > > > > >> thing > > > > > >> > >> should be this: > > > > > >> > >> > you might find that both ends have declared a local epoch > > N, > > > but > > > > > >> someone > > > > > >> > >> has to win. > > > > > >> > >> > > > > > >> > >> That is, as long as we have a way to declare who is the owner > > > of > > > > > >> leader > > > > > >> > >> epoch N, then the 2 clusters can sync up successfully. > > > > > >> > >> And that's why I proposed to the "last mirrored leader epoch" > > > > > >> semantic in > > > > > >> > >> the sub-proposal > > > > > >> > >> < > > > > > >> > >> > > > > > >> > > > > > https://cwiki.apache.org/confluence/display/KAFKA/Unclean+Leader+Election+in+Cluster+Mirroring > > > > > >> > >> >, > > > > > >> > >> which is a solution to draw a line between these 2 clusters > > to > > > > > >> declare > > > > > >> > >> records beyond the "last mirrored leader epoch" N, it belongs > > > to > > > > > >> who. I > > > > > >> > >> think this should work well, as long as all replicas in the > > > cluster > > > > > >> can > > > > > >> > >> truncate the log correctly. > > > > > >> > >> > > > > > >> > >> What do you think? > > > > > >> > >> > > > > > >> > >> Any feedback is appreciated. > > > > > >> > >> > > > > > >> > >> Thank you, > > > > > >> > >> Luke > > > > > >> > >> > > > > > >> > >> On Fri, Mar 6, 2026 at 6:02 PM Andrew Schofield < > > > > > >> [email protected]> > > > > > >> > >> wrote: > > > > > >> > >> > > > > > >> > >> > Hi Fede, > > > > > >> > >> > Thanks for your response. > > > > > >> > >> > > > > > > >> > >> > AS1: Thanks for the clarification. > > > > > >> > >> > > > > > > >> > >> > AS2: I expect you'll include a version bump of > > > > > >> AlterShareGroupOffsets in > > > > > >> > >> > this KIP, but that's a small matter compared with the rest > > > of the > > > > > >> > >> protocol > > > > > >> > >> > changes. > > > > > >> > >> > > > > > > >> > >> > AS3: OK. > > > > > >> > >> > > > > > > >> > >> > AS4: Thanks for the details. My only comment is that it > > > might be a > > > > > >> bit > > > > > >> > >> > laborious when you want to failover all topics. I suggest > > > adding > > > > > >> > >> > `--all-topics` so you could do: > > > > > >> > >> > > > > > > >> > >> > $ bin/kafka-mirror.sh --bootstrap-server :9094 --remove > > > > > >> --all-topics > > > > > >> > >> > --mirror my-mirror > > > > > >> > >> > > > > > > >> > >> > AS5: Thanks for the response. I understand there are good > > > reasons > > > > > >> for > > > > > >> > >> the > > > > > >> > >> > way epochs are handled in the KIP. I see that there is a > > > > > >> sub-document > > > > > >> > >> for > > > > > >> > >> > the KIP about unclean leader election. I'll spend some time > > > > > >> reviewing > > > > > >> > >> that. > > > > > >> > >> > > > > > > >> > >> > Thanks, > > > > > >> > >> > Andrew > > > > > >> > >> > > > > > > >> > >> > On 2026/02/18 13:27:07 Federico Valeri wrote: > > > > > >> > >> > > Hi Andrew, thanks for the review. > > > > > >> > >> > > > > > > > >> > >> > > Let me try to answer your questions and then other > > authors > > > can > > > > > >> join > > > > > >> > >> > > the discussion. > > > > > >> > >> > > > > > > > >> > >> > > AS1 > > > > > >> > >> > > ------ > > > > > >> > >> > > > > > > > >> > >> > > Destination topics are created with the same topic IDs > > > using the > > > > > >> > >> > > extended CreateTopics API. Then, data is replicated > > > starting from > > > > > >> > >> > > offset 0 with byte-for-byte batch copying, so destination > > > offsets > > > > > >> > >> > > always match source offsets. When failing over, we record > > > the > > > > > >> last > > > > > >> > >> > > mirrored offset (LMO) in the destination cluster. When > > > failing > > > > > >> back, > > > > > >> > >> > > the LMO is used for truncating and then start mirroring > > the > > > > > >> delta, > > > > > >> > >> > > otherwise we start mirroring from scratch by truncating > > to > > > zero. > > > > > >> > >> > > > > > > > >> > >> > > Retention: If the mirror leader attempts to fetch an > > > offset that > > > > > >> is > > > > > >> > >> > > below the current log start offset of the source leader > > > (e.g. > > > > > >> fetching > > > > > >> > >> > > offset 50 when log start offset is 100), the source > > broker > > > > > >> returns an > > > > > >> > >> > > OffsetOutOfRangeException that the mirror leader handles > > by > > > > > >> truncating > > > > > >> > >> > > to the source's current log start offset and resuming > > > fetching > > > > > >> from > > > > > >> > >> > > that point. Compaction: The mirror leader replicates > > these > > > > > >> compacted > > > > > >> > >> > > log segments exactly as they exist in the source cluster, > > > > > >> maintaining > > > > > >> > >> > > the same offset assignments and gaps. > > > > > >> > >> > > > > > > > >> > >> > > Do you have any specific corner case in mind? > > > > > >> > >> > > > > > > > >> > >> > > AS2 > > > > > >> > >> > > ------ > > > > > >> > >> > > > > > > > >> > >> > > Agreed. The current AlterShareGroupOffsetsRequest (v0) > > only > > > > > >> includes > > > > > >> > >> > > PartitionIndex and StartOffset with no epoch field. When > > > > > >> mirroring > > > > > >> > >> > > share group offsets across clusters, the epoch is needed > > to > > > > > >> ensure the > > > > > >> > >> > > offset alteration targets the correct leader generation. > > > > > >> > >> > > > > > > > >> > >> > > AS3 > > > > > >> > >> > > ------ > > > > > >> > >> > > > > > > > >> > >> > > Right, the enum is now fixed. Yes, we will parse from the > > > right > > > > > >> and > > > > > >> > >> > > apply the same naming rules used for topic name ;) > > > > > >> > >> > > > > > > > >> > >> > > AS4 > > > > > >> > >> > > ------- > > > > > >> > >> > > > > > > > >> > >> > > Agreed. I'll try to improve those paragraphs because they > > > are > > > > > >> crucial > > > > > >> > >> > > from an operational point of view. > > > > > >> > >> > > > > > > > >> > >> > > Let me shortly explain how it is supposed to work: > > > > > >> > >> > > > > > > > >> > >> > > 9091 (source) -----> 9094 (destination) > > > > > >> > >> > > > > > > > >> > >> > > The single operation that allows an operator to switch > > all > > > > > >> topics at > > > > > >> > >> > > once in case of disaster is the following: > > > > > >> > >> > > > > > > > >> > >> > > bin/kafka-mirror.sh --bootstrap-server :9094 --remove > > > --topic .* > > > > > >> > >> > > --mirror my-mirror > > > > > >> > >> > > > > > > > >> > >> > > 9091 (source) --x--> 9094 (destination) > > > > > >> > >> > > > > > > > >> > >> > > After that, all mirror topics become detached from the > > > source > > > > > >> cluster > > > > > >> > >> > > and start accepting writes (the two cluster are allowed > > to > > > > > >> diverge). > > > > > >> > >> > > > > > > > >> > >> > > When the source cluster is back, the operator can > > failback > > > by > > > > > >> creating > > > > > >> > >> > > a mirror with the same name on the source cluster (new > > > > > >> destination): > > > > > >> > >> > > > > > > > >> > >> > > echo "bootstrap.servers=localhost:9094" > > > > > > >> /tmp/my-mirror.properties > > > > > >> > >> > > bin/kafka-mirrors.sh --bootstrap-server :9091 --create > > > --mirror > > > > > >> > >> > > my-mirror --mirror-config /tmp/my-mirror.properties > > > > > >> > >> > > bin/kafka-mirrors.sh --bootstrap-server :"9091 --add > > > --topic .* > > > > > >> > >> > > --mirror my-mirror > > > > > >> > >> > > > > > > > >> > >> > > 9091 (destination) <----- 9094 (source) > > > > > >> > >> > > > > > > > >> > >> > > AS5 > > > > > >> > >> > > ------- > > > > > >> > >> > > > > > > > >> > >> > > This is the core of our design and we reached that > > > empirically by > > > > > >> > >> > > trying out different options. We didn't want to change > > > local > > > > > >> > >> > > replication, and this is something you need to do when > > > > > >> preserving the > > > > > >> > >> > > source leader epoch. The current design is simple and > > > keeps the > > > > > >> epoch > > > > > >> > >> > > domains entirely separate. Destination cluster is in > > > charge of > > > > > >> the > > > > > >> > >> > > leader epoch for its own log. The source epoch is only > > used > > > > > >> during the > > > > > >> > >> > > fetch protocol to validate responses and detect > > divergence. > > > > > >> > >> > > > > > > > >> > >> > > The polarity idea of tracking whether an epoch bump > > > originated > > > > > >> from > > > > > >> > >> > > replication vs. local leadership change is interesting, > > > but adds > > > > > >> > >> > > significant complexity and coupling between source and > > > > > >> destination > > > > > >> > >> > > epochs. Could you clarify what specific scenario polarity > > > > > >> tracking > > > > > >> > >> > > would address that the current separation doesn't handle? > > > One > > > > > >> case we > > > > > >> > >> > > don't support is unclean leader election reconciliation > > > across > > > > > >> > >> > > clusters, is that the gap you're aiming at? > > > > > >> > >> > > > > > > > >> > >> > > I tried to rewrite the unclean leader election paragraph > > > in the > > > > > >> > >> > > rejected alternatives to be easier to digest. Let me know > > > if it > > > > > >> works. > > > > > >> > >> > > > > > > > >> > >> > > On Tue, Feb 17, 2026 at 2:57 PM Andrew Schofield > > > > > >> > >> > > <[email protected]> wrote: > > > > > >> > >> > > > > > > > > >> > >> > > > Hi Fede and friends, > > > > > >> > >> > > > Thanks for the KIP. > > > > > >> > >> > > > > > > > > >> > >> > > > It’s a comprehensive design, easy to read and has > > clearly > > > > > >> taken a > > > > > >> > >> lot > > > > > >> > >> > of work. > > > > > >> > >> > > > The principle of integrating mirroring into the brokers > > > makes > > > > > >> total > > > > > >> > >> > sense to me. > > > > > >> > >> > > > > > > > > >> > >> > > > The main comment I have is that mirroring like this > > > cannot > > > > > >> handle > > > > > >> > >> > situations > > > > > >> > >> > > > in which multiple topic-partitions are logically > > > related, such > > > > > >> as > > > > > >> > >> > transactions, > > > > > >> > >> > > > with total fidelity. Each topic-partition is being > > > replicated > > > > > >> as a > > > > > >> > >> > separate entity. > > > > > >> > >> > > > The KIP calls this out and describes the behaviour > > > thoroughly. > > > > > >> > >> > > > > > > > > >> > >> > > > A few initial comments. > > > > > >> > >> > > > > > > > > >> > >> > > > AS1) Is it true that offsets are always preserved by > > > this KIP? > > > > > >> I > > > > > >> > >> > *think* so but > > > > > >> > >> > > > not totally sure that it’s true in all cases. It would > > > > > >> certainly be > > > > > >> > >> > nice. > > > > > >> > >> > > > > > > > > >> > >> > > > AS2) I think you need to add epoch information to > > > > > >> > >> > AlterShareGroupOffsetsRequest. > > > > > >> > >> > > > It really should already be there in hindsight, but I > > > think > > > > > >> this KIP > > > > > >> > >> > requires it. > > > > > >> > >> > > > > > > > > >> > >> > > > AS3) The CoordinatorType enum for MIRROR will need to > > be > > > 3 > > > > > >> because 2 > > > > > >> > >> > is SHARE. > > > > > >> > >> > > > I’m sure you’ll parse the keys from the right ;) > > > > > >> > >> > > > > > > > > >> > >> > > > AS4) The procedure for achieving a failover could be > > > clearer. > > > > > >> Let’s > > > > > >> > >> > say that I am > > > > > >> > >> > > > using cluster mirroring to achieve DR replication. My > > > source > > > > > >> cluster > > > > > >> > >> > is utterly lost > > > > > >> > >> > > > due to a disaster. What’s the single operation that I > > > perform > > > > > >> to > > > > > >> > >> > switch all of the > > > > > >> > >> > > > topics mirrored from the lost source cluster to become > > > the > > > > > >> active > > > > > >> > >> > topics? > > > > > >> > >> > > > Similarly for failback. > > > > > >> > >> > > > > > > > > >> > >> > > > AS5) The only piece that I’m really unsure of is the > > > epoch > > > > > >> > >> management. > > > > > >> > >> > I would > > > > > >> > >> > > > have thought that the cluster which currently has the > > > writable > > > > > >> > >> > topic-partition > > > > > >> > >> > > > would be in charge of the leader epoch and it would not > > > be > > > > > >> > >> necessary to > > > > > >> > >> > > > perform all of the gymnastics described in the section > > > on epoch > > > > > >> > >> > rewriting. > > > > > >> > >> > > > I have read the Rejected Alternatives section too, but > > I > > > don’t > > > > > >> fully > > > > > >> > >> > grasp > > > > > >> > >> > > > why it was necessary to reject it. > > > > > >> > >> > > > > > > > > >> > >> > > > I wonder if we could store the “polarity” of an epoch, > > > > > >> essentially > > > > > >> > >> > whether the > > > > > >> > >> > > > epoch bump was observed by replication from a source > > > cluster, > > > > > >> or > > > > > >> > >> > whether > > > > > >> > >> > > > it was bumped by a local leadership change when the > > > topic is > > > > > >> locally > > > > > >> > >> > writable. > > > > > >> > >> > > > When a topic-partition switches from read-only to > > > writable, we > > > > > >> > >> should > > > > > >> > >> > definitely > > > > > >> > >> > > > bump the epoch, and we could record the fact that it > > was > > > a > > > > > >> local > > > > > >> > >> epoch. > > > > > >> > >> > > > When connectivity is re-established, you might find > > that > > > both > > > > > >> ends > > > > > >> > >> have > > > > > >> > >> > > > declared a local epoch N, but someone has to win. > > > > > >> > >> > > > > > > > > >> > >> > > > Thanks, > > > > > >> > >> > > > Andrew > > > > > >> > >> > > > > > > > > >> > >> > > > > On 14 Feb 2026, at 07:17, Federico Valeri < > > > > > >> [email protected]> > > > > > >> > >> > wrote: > > > > > >> > >> > > > > > > > > > >> > >> > > > > Hi, we would like to start a discussion thread about > > > > > >> KIP-1279: > > > > > >> > >> > Cluster > > > > > >> > >> > > > > Mirroring. > > > > > >> > >> > > > > > > > > > >> > >> > > > > Cluster Mirroring is a new Kafka feature that enables > > > native, > > > > > >> > >> > > > > broker-level topic replication across clusters. > > Unlike > > > > > >> > >> MirrorMaker 2 > > > > > >> > >> > > > > (which runs as an external Connect-based tool), > > Cluster > > > > > >> Mirroring > > > > > >> > >> is > > > > > >> > >> > > > > built into the broker itself, allowing tighter > > > integration > > > > > >> with > > > > > >> > >> the > > > > > >> > >> > > > > controller, coordinator, and partition lifecycle. > > > > > >> > >> > > > > > > > > > >> > >> > > > > > > > > > >> > >> > > > > > > >> > >> > > > > > >> > > > > > https://cwiki.apache.org/confluence/display/KAFKA/KIP-1279%3A+Cluster+Mirroring > > > > > >> > >> > > > > > > > > > >> > >> > > > > There are a few missing bits, but most of the design > > is > > > > > >> there, so > > > > > >> > >> we > > > > > >> > >> > > > > think it is the right time to involve the community > > > and get > > > > > >> > >> feedback. > > > > > >> > >> > > > > Please help validating our approach. > > > > > >> > >> > > > > > > > > > >> > >> > > > > Thanks > > > > > >> > >> > > > > Fede > > > > > >> > >> > > > > > > > > >> > >> > > > > > > > >> > >> > > > > > > >> > >> > > > > > >> > > > > > > > >> > > > > > > > >> > > -- > > > > > >> > > Paolo Patierno > > > > > >> > > > > > > > >> > > *Senior Principal Software Engineer @ IBM**CNCF Ambassador* > > > > > >> > > > > > > > >> > > Twitter : @ppatierno <http://twitter.com/ppatierno> > > > > > >> > > Linkedin : paolopatierno < > > > http://it.linkedin.com/in/paolopatierno> > > > > > >> > > GitHub : ppatierno <https://github.com/ppatierno> > > > > > >> > > > > > > > >> > > > > > > >> > > > > > > >> > -- > > > > > >> > Paolo Patierno > > > > > >> > > > > > > >> > *Senior Principal Software Engineer @ IBM**CNCF Ambassador* > > > > > >> > > > > > > >> > Twitter : @ppatierno <http://twitter.com/ppatierno> > > > > > >> > Linkedin : paolopatierno < > > http://it.linkedin.com/in/paolopatierno > > > > > > > > > >> > GitHub : ppatierno <https://github.com/ppatierno> > > > > > >> > > > > > > >> > > > > > > > > > > > > > >
