...

Solving this issue would require creating a shared leader epoch between source and destination clusters. Every time there is a source leader election we would need to notify the destination cluster and append data only after receiving a reply. This means that the overall latency would be cross-cluster replication latency plus intra-cluster replication latency. Read more in the Rejected Alternatives section.

Proposed Changes

Cluster Mirroring introduces a coordinator-based architecture integrated into Kafka brokers for managing cross-cluster replication. The design consists of three primary components that work together to provide automatic metadata synchronization and data replication. The following diagram illustrates how these components are wired together.

...

• Performance Benchmark: Measure replication throughput and latency across WAN.
• Scalability Test: Replicate 1000 topics with 100,000 partitions across clusters.
• Failover Test: Simulate source cluster failure, measure consumer recovery time.
• Long-Running Stability: Run continuous replication for 7 days, verify no memory leaks or performance degradation.
• Security Validation: Test all authentication mechanisms (SASL PLAIN, SCRAM, Kerberos, mTLS) via kafka-mirrors.sh config files.

Rejected Alternatives

Keep using MirrorMaker 2:

This KIP is to address the drawbacks existing MirrorMaker 2 as described in the motivation section. 

Support unclean leader election

As described in the non-goal section, since there's no shared leader epoch between source and

...

detination cluster, supporting unclean leader election becomes very tricky. For example:

source cluster
leader for foo-0 contains this data:
offset 0, epoch: 0, value: A
offset 1, epoch: 1, value: B

Suppose we mirror everything from the source into destination cluster, including the leader epoch in batches:
target cluster
leader for foo-0 contains this data:
offset 0, epoch: 0, value: A
offset 1, epoch: 1, value: B

===
This could happen:

1. leadership change in the source cluster, bumping the leader epoch to 2
2. New records appended to source cluster: offset 2, epoch: 2, value: C
3. Before target cluster fetch from the source to identify the leader epoch update, source cluster down, failover happened
4. Topics in destination cluster becomes writable, new records appended from producer:  offset 2, epoch: 2, value: D
5. Keep using MirrorMaker 2:
   This KIP is to address the drawbacks existing MirrorMaker 2 as described in the motivation section. 

Inconsistent result:
source cluster
leader for foo-0 contains this data:
offset 0, epoch: 0, value: A
offset 1, epoch: 1, value: B
offset 2, epoch: 2, value: C

target cluster
leader for foo-0 contains this data:
offset 0, epoch: 0, value: A
offset 1, epoch: 1, value: B
offset 2, epoch: 2, value: D

The issue above can be resolved by the LastMirroredOffset API we did in this KIP. The flow will be like this:

1. leadership change in the source cluster, bumping the leader epoch to 2
2. New records appended to source cluster: offset 2, epoch: 2, value: C
3. Before target cluster fetch from the source to identify the update, source cluster down, failover happened
4. When failover, the destination cluster will store the current last mirrored offset (1 in this case) into internal topic.
5. Topics in destination cluster becomes writable, new records appended from producer:  offset 2, epoch: 2, value: D
6. When the old source cluster wants to reverse mirroring to the new source cluster, it'll firstly ask for the last mirrored offset, which is 1 in this case. Then, truncate data to offset 1.
7. Then, start fetch from offset 1.

It works well, but when unclean leader election comes into the play, it'll become complicated:
 

1. unclean leader election happened and leadership change in the source cluster, bumping the leader epoch to 2
2. New leader has empty log in disk
3. New records appended to source cluster: offset 0, epoch: 2, value: C
4. Before target cluster fetch from the source to identify the update, source cluster down, failover happened
5. When failover, the destination cluster will store the current last mirrored offset (1 in this case) into internal topic.
6. Topics in destination cluster becomes writable, new records appended from producer:  offset 2, epoch: 2, value: D
7. When the old source cluster wants to reverse mirroring to the new source cluster, it'll firstly ask for the last mirrored offset, which is 1 in this case. Then, truncate data to offset 1.

After this truncation, the data diverge still exist:

source cluster
leader for foo-0 contains this data:
offset 0, epoch: 2, value: C

target cluster
leader for foo-0 contains this data:
offset 0, epoch: 0, value: A
offset 1, epoch: 1, value: B
offset 2, epoch: 2, value: D

In summary, no matter we store the source partition leader epoch in the target cluster or not, there will always be a gap in the target cluster given it's using async fetch request/response or metadata request/response to get the metadata update. When the target cluster misses some leadership change update and failover to the target clsuter, there is no way to sync up with the source cluster anymore. Thus, the inconsistent data will happen after the old source cluster starts to reverse mirror from the old target cluster (new source). To fix this issue, a shared leader epoch mechanism is required. But that's out of the scope of this KIP.