KAFKA-15352: Update log-start-offset before initiating deletion of remote segments

[clolov]

This pull request tries to solve is to first update the log-start-offset and then delete remote segments.
In the previous version if a read request arrives between us deleting a segment and updating the log-start-offset we won't be able to service it.

[kamalcph]

The changes LGTM. Could you please cover the patch with unit tests?

[kamalcph]

Can we and the result of remoteLogRetentionHandler.deleteRemoteLogSegment for all the segments and log an error statement if it is not able to delete the segments?

remoteLogRetentionHandler.deleteRemoteLogSegment(segmentMetadata, x -> !isCancelled() && isLeader());

[kamalcph]

@divijvaidya

Could you please review this patch? Discussion thread:

This patch addresses the case-2 mentioned in the KAFKA-15352 ticket partially. Only the leader will move the log-start-offset and starts to delete the remote log segments. In middle of deleting the remote log segments, if leader switch happens then there can be two cases:

1. The log-start-offset changes was propagated to all the replicas including the new-leader. In this case, both the old-leader and new-leader can delete the remote log segments since the log-start-offset can be updated only with the monotonically increasing value.
2. The log-start-offset changes was not propagated to the new leader. In this case, the new leader assumes the log-start-offset (with KAFKA-15351) which can be deleted immediately by the old leader and the log-start-offset can be stale until the next segment deletion, we can address this case by ensuring that the log-start-offset propagates to all the replicas before deleting the remote log segments, we can take this up later (or) file a JIRA.

[divijvaidya]

I have some meta questions:

1. The race condition between start offset getting updated, leadership getting moved and deletion by RLM is a tricky one. Is there a way to add integ tests for the different scenarios?

2. Log start offset is updated only by expiration related work. In this PR we update log start offset even though we might not perform expiration. Isn't it safer to update log start offset only if we are leader? Yes, we will still have a case where leadership changes between updating log start and deleting this which is handled by this PR but the probability of that happening is low since there is no compute intensive statements between these two. My point is, should we check for leadership even before updating the log start offset?

[satishd]

Thanks @clolov for the PR. Please rebase it with the latest trunk and resolve the conflicts to make it available for review.

[kamalcph]

My point is, should we check for leadership even before updating the log start offset?

yes, this is the expectation and being done inside the handleLogStartOffsetUpdate method:

public void handleLogStartOffsetUpdate(TopicPartition topicPartition, long remoteLogStartOffset) {
    if (isLeader()) {
        logger.debug("Updating {} with remoteLogStartOffset: {}", topicPartition, remoteLogStartOffset);
        updateRemoteLogStartOffset.accept(topicPartition, remoteLogStartOffset);
    }
}

[satishd]

Thanks @clolov for the PR. Had an initial review of the changes.

[kamalcph]

Can we use list of tuples instead of list of map?

[kamalcph]

Instead of asserting the log-start-offset via events. Can we assert it similar to testLogStartOffsetUpdatedOnStartup method?

AtomicLong logStartOffset = new AtomicLong(0);
try (RemoteLogManager remoteLogManager = new RemoteLogManager(remoteLogManagerConfig, brokerId, logDir, clusterId, time,
        tp -> Optional.of(mockLog),
        (topicPartition, offset) ->  logStartOffset.set(offset),
        brokerTopicStats) {
    public RemoteLogMetadataManager createRemoteLogMetadataManager() {
        return remoteLogMetadataManager;
    }
}) {
    RemoteLogManager.RLMTask task = remoteLogManager.new RLMTask(leaderTopicIdPartition, 128);
    task.convertToLeader(0);
    task.run();
    assertEquals(200L, logStartOffset.get());
    verify(remoteStorageManager).deleteLogSegmentData(remoteLogSegmentMetadatas.get(0));
    verify(remoteStorageManager, never()).deleteLogSegmentData(remoteLogSegmentMetadatas.get(1));
}

[clolov]

I am happy to change it to this. The reason why I implemented it with events is that it allowed me to carry out the assertion before deletes were initiated. With your approach we assert that the log start offset has been updated, but we do not assert that it was updated before deletes were carried out. Would you still like me to change it to your proposal?

[clolov]

Or do you mean that the problem is not with the location of the assertion, but that I do not need a list of events where a single AtomicLong will be sufficient?

[kamalcph]

yes, single AtomicLong will be sufficient. We can assert before and after calling the task.run() method.

[kamalcph]

LGTM. Thanks @clolov for addressing the review comments.

[showuon]

Overall LGTM! Thanks for the fix. Left some comments.

[showuon]

nit: These if block and L1004-1006 could be put before canProcess = isSegmentDeleted || !isValidSegment;. I.e.

boolean shouldDeleteSegment = remoteLogRetentionHandler.deleteLogStartOffsetBreachedSegments(
                            metadata, logStartOffset, epochWithOffsets);
boolean isValidSegment = false;

if (!shouldDeleteSegment) {
    ...
  if (isValidSegment) {
      shouldDeleteSegment = remoteLogRetentionHandler.deleteRetentionTimeBreachedSegments(metadata) || remoteLogRetentionHandler.deleteRetentionSizeBreachedSegments(metadata);
  }
}
if (shouldDeleteSegment) {
   segmentsToDelete.add(metadata);
}

canProcess = isSegmentDeleted || !isValidSegment;

[clolov]

Thanks for the spot! Hopefully addressed in the next commit

[showuon]

I think this is a good place to continue to verify the situation we want to protect:

// If the follower HAS picked up the changes, and they become the leader this replica won't successfully complete the deletion.
// However, the new leader will correctly pick up all breaching segments as log start offset breaching ones
// and delete them accordingly.

// If the follower HAS NOT picked up the changes, and they become the leader then they will go through this process
// again and delete them with the original deletion reason i.e. size, time or log start offset breach.

So, I'm thinking we can continue the test with sth like:

RemoteLogManager.RLMTask task = remoteLogManager.new RLMTask(followerTopicIdPartition, 128);
task.convertToLeader(1);
....
task.run();

assertEquals(200L, logStartOffset.get());

// verify the 2nd log segment will be deleted by the new leader.
verify(remoteStorageManager).deleteLogSegmentData(remoteLogSegmentMetadatas.get(1));

WDYT?

[clolov]

This is a great idea, apologies for not doing it myself. Hopefully the next commit addresses this

[clolov]

Thanks for the latest comments! I will review and respond today

[divijvaidya]

Please don't wait for comments from me to merge this. I am good with the code changes here.

[showuon]

LGTM! Thanks for the fix!

[showuon]

@satishd , do you want to have another look?

[satishd]

Thanks @clolov , went through the source code changes and LGTM.

This change is about updating the log-start-offset before the segments are deleted from remote storage. This will do a best effort mechanism for followers to receive log-start-offset and they can update their log-start-offset before it becomes a leader.

I could not take a closer look at the tests. I do not want the PR to be blocking, I am fine as others already reviewed these tests.

[satishd]

There are a few unrelated test failures. Merging it to trunk and 3.6