Page History

...

Support Unclean Leader Election

As described in the non-goal section, since there's Since there is no shared leader epoch between source and destination cluster, supporting unclean leader election becomes very tricky.

For example:

...

1. Leadership change in the source cluster, bumping the leader epoch to 2.
2. New records appended to the 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 become writable, new records appended from producer: offset=2, epoch=2, value=D.

...

The above issue can be resolved by the LastMirroredOffset API we propose in this KIP.

The flow will be like this:

...

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:

...

clusters, the destination cannot detect log divergence caused by unclean leader election on the source. With clean leader elections, the LastMirroredOffsets API is sufficient to reconcile the two clusters during failback. With unclean leader elections, it is not.

Clean leader election: (LMO works)

Consider a mirror where the destination has replicated offsets 0 and 1 from the source:

source (foo-0):               destination (foo-0):
offset 0, value: A            offset 0, value: A
offset 1, value: B            offset 1, value: B

A clean leadership change on the source bumps its epoch and appends a new record at offset 2. Before the destination fetches this record, the source goes down and failover occurs. The destination records LMO = 1 and becomes writable. A producer appends a different record at offset 2:

source (foo-0):               destination (foo-0):
offset 0, value: A            offset 0, value: A
offset 1, value: B            offset 1, value: B
offset 2, value: C            offset 2, value: D

When the old source comes back and wants to reverse-mirror from the destination, it queries the LastMirroredOffsets API and gets LMO = 1. It truncates its log to offset 1, then starts fetching from the destination. The logs converge.

Unclean leader election (LMO is not sufficient)

Now consider the same setup, but the leadership change on the source is an unclean election. The new source leader has lost data and starts with a divergent log:

source after unclean election (foo-0):           destination (foo-0):
offset 0, value: X                               offset 0, value: A
                                                 offset 1, value: B

The destination has not yet detected this divergence. The source appends more data. Before the destination can fetch and discover the epoch change, the source goes down and failover occurs. The destination records LMO = 1:

source (foo-0):               destination (foo-0):
offset 0, value: X            offset 0, value: A
offset 1, value: Y            offset 1, value: B
                              offset 2, value: D  (written after failover)

When the old source reverse-mirrors, it queries LMO = 1 and truncates to offset 1. But this only removes offset 1 onward, while offset 0 still contains X on the source vs A on the destination. The divergence at offsets below the LMO cannot be detected or resolved, because the destination has no record of the source's epoch history to compare against.

In summary, LMO-based truncation assumes the source log up to the LMO is a prefix of the destination log. Clean leader elections preserve this invariant. Unclean leader elections break it by allowing the source log to diverge at arbitrary offsets, including offsets already replicated to the destination. Resolving this would require a shared epoch mechanism or cross-cluster log reconciliation protocol

...

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.

...

The above issue can be resolved by the LastMirroredOffset API we propose 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 the 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 failback 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 fetching 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 the 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 failback 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.

...

In summary, because mirroring relies on asynchronous fetch and metadata requests, the destination cluster can always miss leadership changes on the source cluster. If a failover occurs while the destination cluster has missed such updates, it cannot reconcile its state with the source cluster. This leads to inconsistent data once the old source cluster begins reverse mirroring from the old destination (now the new source). Resolving this requires a shared leader epoch mechanism, which is out of scope for this KIP.

...

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. The new leader has an empty log in the 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: DIn summary, because mirroring relies on asynchronous fetch and metadata requests, the destination cluster can always miss leadership changes on the source cluster. If a failover occurs while the destination cluster has missed such updates, it cannot reconcile its state with the source cluster. This leads to inconsistent data once the old source cluster begins reverse mirroring from the old destination (now the new source). Resolving this requires a shared leader epoch mechanism, which is out of scope for this KIP.