KAFKA-15168: Handle overlapping remote log segments in RemoteLogMetadata cache

[kamalcph]

KAFKA-15168: Handle overlapping remote log segments in RemoteLogMetadata cache
Added unit tests covering different scenarios of overlapping segments.

Committer Checklist (excluded from commit message)

• Verify design and implementation
• Verify test coverage and CI build status
• Verify documentation (including upgrade notes)

[satishd]

Thanks @kamalcph for the PR. Overall LGTM, left a minor comment to add doc with an example/

[satishd]

Please add an example of overlapping segments int he doc here for reader's clarity.

[abhijeetk88]

+1

[abhijeetk88]

Overall LGTM. Thanks for the PR.

[kamalcph]

@satishd @abhijeetk88

Addressed your review comments. PTAL

[kamalcph]

@jeqo @showuon

PTAL when you get chance. Thanks!

[satishd]

Thanks @kamalcph for addressing the review comments. LGTM.

[jeqo]

LGTM, thanks @kamalcph

[Nickstery]

Hi, I have two questions here:

1. In the while loop we are checking highestLogOffset <= leaderEpochEndOffset but we do not change them within loop. Do we really need it inside while, or we can wrap while in that if check?
2. Why do we care about highestLogOffset?
   Input: We have 1 topic with RF=2 and min.insync.replica=1. Two brokers per cluster.
   Imagine situation when Broker-1 and Broker-2 have segment each with 10 synced messages, so they both have segment [start offset = 1 and end offset = 11]. New segment rolled, Broker-1 got 100 messages, Broker-2 replicated 80 of then become offline [non tiered topic existed on cluster and occupied 100% of storage]. Broker-1 sends segment to remote storage. At that point Remote storage has 2 segments [startOffset=1, endOffset=11], [startOffset=12, endOffset=112]. Broker-1 dies with same data disk issue, and Broker-2 restored after some time before Broker-1 revived. Since it had 80 messages and retention by segment size was not reached, producer produces few bigger messages than usual, segment is rolled with 99 messages [startOffset=12, endOffset=101]. By the logic there it won't remove last segment, since its highestLogOffset > leaderEpochEndOffset

What kafka does in that case? when Broker-1 is up, it checks that new leader epoch change took place, and it removes messages up to latest synced message and replicate missing messages from Broker-2. That means previous non replicated messages on Broker-1 is gone, but in tiered storage it will be duplicate segment. Are we ok with that?

If no, we should remove check for highestLogOffset <= leaderEpochEndOffset and extend logic with checking leaderEpoch number. Or combine this checks

[kamalcph]

1. highestLogOffset is the highest end offset of all the uploaded segments so far, so while removing the last entry inside the while loop, updating the highestLogOffset value is not required. A last entry is eligible to be removed only when the highestLogOffset seen so far is lesser than the current segment end offset.
2. The case that you mentioned is about unclean leader election. A passive segment is eligible for upload to remote storage only when the last-stable-offset is ahead of the passive segment end offset. When B2 was restored, B1 was the last standing replica and died with disk failure, so B2 can only be elected with unclean leader election.

When B2 elected as unclean leader, it will have 2 segments [1-11] and [12-101], where the second segment [12 - 101] contains messages written by more than one leader epoch (LE0:12-80 and LE1: 81-101). B2 cannot be able to upload the segment S2 (12:101) as the highest offset uploaded so far in the remote storage for LE0 is 112. B2 can only be able to upload from segments which contain offset 113.

When B1 was brought back as replica, it truncates itself up-to the largest common log prefix offset (target-offset) including the leader-epoch-checkpoint file so it cannot serve the uploaded segments from remote storage. If the target-offset is lesser than the local-log-start-offset, then the uploaded remote log segments will be cleared eventually when smallest (leader_epoch, start_offset) in the leader-epoch-checkpoint file is greater the unreferenced segment epoch.

RemoteLogLeaderEpochState is maintained for one epoch, we cannot extend it for multiple leader_epochs.

In remote storage, the records for the offsets from [81-112] will be maintained for both LE0 and LE1. With KIP-320, the fetcher can detect and handle the log truncation themselves.

Thanks for the detailed review! Let me know if I'm missing something.

[kamalcph]

Thinking more on this, we may have to update the logic to find the copiedOffset such that it's minimum of (end_offset_for_that_epoch_in_checkpoint_file, highest_remote_offset). In this case, when B2 was leader, copied_offset should be min(80, 112) where 80 is end_offset for LE0 and 112 is the highest_remote_offset for LE0.

So, that B2 can able to upload the segment S2 (12:101)

[Nickstery]

Hi, thank you for explanation, sounds reasonable.

Sorry for being too questionable. One more point:
What could happen if we instead:

while (lastEntry != null && lastEntry.getKey() >= startOffset && highestLogOffset <= leaderEpochEndOffset)
....
if (highestLogOffset == null || leaderEpochEndOffset > highestLogOffset) {
            highestLogOffset = leaderEpochEndOffset;
}

will do

while (lastEntry != null && lastEntry.getKey() >= startOffset)
....
highestLogOffset = leaderEpochEndOffset;

How I see it:

• In case we sent twice same segment [somehow], It will be replaced with the same one
• In case unclean.leader.election set to true, we are going to have up to date data, in that case TS will work same as local storage function wise [data loss due to dead leader]. Less discrepancy - easier to maintain and understand
• I do believe we do not send from leader and all the followers same segments. The leader is someone who sends data to tiered storage, so we expect at the same time send only 1 segment for partition-X, in case second request comes for same partition, it means that new leader is elected and it is in charge and its data is more valuable.
• It is possible that even ISR won't be synced fully replica.lag.time.max.ms [default= 30s] allows that, it means that it is possible that few messages wont be synced and potentially deleted in case traffic stops coming few messages after Leader broker died and new leader got them. What if Broker 1 uploaded segment [0, 100], Broker-2 was insync by replicated [0,90]. Broker-1 died, Broker-2 elected as a Leader and got 5 messages, Tiered storage won't be in charge of those 5 messages and segment could be deleted. After traffic being restored new segment is going to roll and its offset starts with 96, when last offset in TS is 100, so we delete segment and push new one starting from 96, and we could have a gap afterwards.

Of course this example is an edge case and configuration should be quite agressive for that in terms for retention times etc. But are the any drawbacks by removing highestLogOffset <= leaderEpochEndOffset from the while? What kind of problem we can face and why having this check is more safe rather than remove it?

[Nickstery]

Test example

@Test
    void handleSegmentWithCopySegmentFinishedStateForDuplicateSegmentsWithSecondSegmentEarlierEndOffset() {
        // Broker-1 is a Leader for Partition-0, Broker-2 is follower
        RemoteLogSegmentId segmentId1BRoker1 = new RemoteLogSegmentId(tpId, Uuid.randomUuid());
        //Leader and follower are in sync
        epochState.handleSegmentWithCopySegmentFinishedState(10L, segmentId1BRoker1, 100L);

        //Broker-1 received some messages and part of them were replicated by follower
        // Broker-2 was not able to replicate all the messages because Broker-1 is died. 
        // But it was able to upload segment before death [101, 200]
        RemoteLogSegmentId segmentId2Broker1 = new RemoteLogSegmentId(tpId, Uuid.randomUuid());
        epochState.handleSegmentWithCopySegmentFinishedState(101L, segmentId2Broker1, 200L);
        
        // Broker-2 still in-sync according to `replica.lag.time.max.ms`
        // Last offset in sync is 150L
        // Broker-2 gets the leadership and fills the segment 101L-190L and uploads to Tiered storage
        RemoteLogSegmentId segmentId2Broker2 = new RemoteLogSegmentId(tpId, Uuid.randomUuid());
        epochState.handleSegmentWithCopySegmentFinishedState(101L, segmentId2Broker2, 190L);
        
        //Traffic stops
        
        //Since LeaderEpoch was changed, I expect that more fresh data will be on the TS
        assertEquals(190L, epochState.highestLogOffset());
        //segment 2 uploaded by Broker-1 is expected to be unreferenced
        assertTrue(epochState.unreferencedSegmentIds().contains(segmentId2Broker1));
        assertEquals(1, epochState.unreferencedSegmentIds().size());
    }
    
Then it could be part of integration test, we can upload new segment [191L, 300L] and consume from the beginning from TS. With current logic we have data loss from segment 2 where BROKER-2 was leader [151L,190L], but in case of Kafka local storage functionality, it would not happen

[satishd]

Why would segment data for [151L,190L] is lost? This will be part of the next leader epoch state(viz 1) which should contain this segment. @kamalcph @Nickstery Am I missing anything here?

[kamalcph]

@satishd @Nickstery Correct me If I'm wrong:

The logic to find the copiedOffset from remoteStorage doesn't take in account of the current-leader-epoch checkpoint file. In the above test, when B2 becomes leader, it's leader-epoch-checkpoint file will look like:

(The case mentioned in the test can happen when acks is set to 1)

0
2
0 0
1 151

The logic to find the copied offset traverses the checkpoint file from latest-epoch. So, when B2 tries to find the copied offset:

For epoch(1), there won't any uploaded segments, so it returns empty.
For epoch(0), the highest copied offset will be 200

So, B2 will skip the segment S2 (101-190) which means there is a data loss from [151-190]

@Nickstery This can be fixed if we update the logic to find the copiedOffset:

find-highest-remote-offset = min(end-offset-for-epoch-in-the-checkpoint, highest-remote-offset-for-epoch)

[kamalcph]

But are the any drawbacks by removing highestLogOffset <= leaderEpochEndOffset from the while? What kind of problem we can face and why having this check is more safe rather than remove it?

Assume there are multiple back-to-back unclean leader elections happened and only one replica is available at any point of time. Both B1 and B2 may not be aware of all the leader-epochs. If the consumer reads the data from the beginning of the topic, then we should be able to serve the respective remote log segments for the (epoch, start_offset) present in both B1 and B2.

This patch only mark the segment as unreferenced if the current segment is a superset of all the previously uploaded segments for the same epoch which means the message is same across the segments.

[satishd]

Thanks @kamalcph for the detailed explanation. It is better to return the epoch along with highest-offset and then check the minimum of the respective end offset in the local leader epoch chain for that epoch and the highest offset uploaded to remote storage for that epoch.

[satishd]

Filed https://issues.apache.org/jira/browse/KAFKA-15241 to followup on this.

[satishd]

Thanks @kamalcph for addressing the comments. I will merge it to trunk once the Jenkins jobs look good.