Last Mirrored Leader Epoch (LMLE): This is the greatest leader epoch that the cluster recognizes from the other cluster. Or we can say, below this leader epoch (inclusive), the source cluster and destination cluster is in sync in the perspective of leader epoch.
Bump Leader Epoch: This is to make sure the leader epoch in the partition is monotonically increasing. Because the leader epoch in records are syncing up with the source cluster, when the partition becomes writable after failover, we have to make sure the new records appended from producer having leader epoch > current highest leader epoch in the log.
After log truncation for LMLE, the destination cluster should not contain any record beyond the LMLE. And all the records after LMLE should only be able to mirror from the source cluster. In other words, the owner of leader epoch > LMLE belongs to the source cluster. This is to make sure each leader epoch only belongs to 1 cluster even if the leader epoch in these 2 clusters are not in sync.
1. Cluster B creates a mirror and starts mirroring from the source cluster A. At this point, because the source cluster stores nothing about the LMLE for this partition, -1 will be returned. The "-1" means the source cluster (A) doesn't recognize any leader epoch in destination cluster (B), so all the logs should be truncated.
2. Revere mirroring. Cluster B becomes the source cluster. And before cluster B stops the mirroring from A, it stores the LMLE in the internal topic, which is 1 in this case. When cluster A mirrors from cluster B, it will first query the LMLE from cluster B, and then truncate every record beyond this leader epoch. That is, for cluster A and B, leader epochs for [0, 1] are in sync.
Also, in the cluster B, when it stops the mirror from A, it also bumps the leader epoch to a leader epoch > 1. In this case, it is 2, to make sure the leader epoch is increasing.
3. Reverse mirroring again. Cluster A stores LMLE 3 in the internal topic, and cluster B queries this LMLE and truncate records beyond leader epoch 3. Besides, the leader epoch in cluster A bumps to 4.
Please note that after LMLE truncation, it doesn't mean the log between 2 clusters are in sync. It only means the leader epoch history is in sync. The leader epoch history (i.e. the leader-epoch-checkpoint file in the partition directory) contains the entry for each [leader epoch, start offset]. However, the log might still not converge, yet. In the example below, after cluster B truncates LMLE 3, the offset 4 (in leader epoch 3) still exists, which is inconsistent with the source cluster. In this case, the second log truncation will be triggered by the replication protocol by the last fetched leader epoch comparison. So the offset 4 will be truncated this time, and the logs in these 2 clusters are completely in sync.
When the log truncation is processing before mirroring, the leader node will wait until all ISRs (and ISR size >= "min.insync.replicas" value) complete the truncation. This is to make sure all ISRs logs are in sync.
The unclean leader election means a replica that is not part of ISR becomes the leader. It will cause data loss because some committed records haven't replicated to this new leader replica, yet. With the LMLE design, it can resolve unclean leader election happened before mirroring. For example:
1. cluster B is mirroring from cluster A.
2. broker 0 in cluster A is down. No ISR is available, and unclean leader election triggered to a replica with only offset 0 in log.
3. New records append to the new leader in offset 1,2,3.
4. If cluster B continues to mirror from the cluster A, it’ll identify the epoch diverge and do log truncation to offset 0. But before the fetch request is sent, the whole cluster A is down and failover to cluster B. The cluster A will query the LMLE, and cluster B responds with 1. So records with leader epoch > 1 will be truncated.
The example above resolves the unclean leader election before the reverse mirroring. However, the unclean leader election could happen anytime. And this might mess up the design above because the log truncation will wait only for all ISRs are completed. That means, once unclean leader election happens, the non-ISR might contain records in leader epoch beyond the LMLE. Thus the replication protocol cannot detect the log diverge and cause inconsistent data between these 2 clusters.
1. Continue the step 3 above, the current log status is like this, after the unclean leader election.
2. Unclean leader election again in the cluster A, the old leader becomes the new leader now.
3. Failover to cluster B, and cluster A reverse mirroring from cluster B. These 2 clusters are in sync now.
4. Unclean leader election happened in cluster A, but no log truncation this time. Thus it causes log inconsistent.
1. Continue the step 3 in case 1. It failover to cluster B, and cluster A reverse mirroring from cluster B. These 2 clusters are in sync now.
2. fail over to cluster A, and cluster B reverse mirror from cluster A
3. unclean leader election in cluster A, which causes inconsistent logs
These cases all have one common root cause: the non-ISRs, which have not completed the log truncation for LMLE, can become the leader (due to unclean leader election), and it breaks the assumption at the beginning of this page:
After log truncation for LMLE, the destination cluster should not contain any record beyond the LMLE.
So, to resolve these cases, we introduce a new "topic-level" config mirror.support.unclean.leader.election . By default it is false. When it is true, during the log truncation for LMLE, we will wait for "all replicas" becomes ISR and completes the log truncation. Again, this is to fulfill the assumption above. So after the log truncation for LMLE, even if there is unclean leader election triggered any time, these log in 2 clusters can still converge successfully using the existing replication protocol.
If there are some replicas cannot catch up with the leader due to that are slow network, disk issue, ... etc while log truncation for LMLE, it will cause the mirroring pending or in FAILED state. In this case, users should manually resolve the issue, or re-assign the replicas into other healthy brokers, and start the mirroring again.
Because the new added mirror.support.unclean.leader.election config is a dynamic configuration, it could be enabled in the middle of the mirroring, not at the beginning of the mirroring. If it is enabled at the beginning of the mirroring, all records will guarantee to be consistent between these 2 clusters even if unclean leader election happened. But if it is enabled in the middle of the mirroring, the guarantee becomes:
records after next log truncation for LMLE will be consistent.
For example, cluster A enables unclean leader election for this topic. And during the log truncation for LMLE 1 at (2), the mirror.support.unclean.leader.election is false, and it is enabled after LMLE 1 truncation, then, we will make sure all records after next LMLE truncation, i.e. (4) LMLE 5, will be consistent in the 2 clusters.
The reason of the guarantee is because the unclean leader truncation in cluster could happen after (4) LMLE 5 truncation like below image. The leader epoch 5 for offset 2 ~ 5 in the non-ISR log didn't get truncated in (2) LMLE: 1 process, so it causes inconsistent data before LMLE 5. But after LMLE 5, all records will be consistent with the cluster B because of log truncation are done in "all replicas".
