KAFKA-19952: Don't upload to remote if the segment had already expired according to remote retention configure

[jiafu1115]

This is another solution for the bug #21049 (The solution is to check on
deleting) Also it is one demo solution for the comment from @kamalcph
for KIP:1241

When the remote copy is configured to be lazy, What is the behaviour
when local and complete retention values are set to the same?
Do we upload the data to remote, then immediately delete it from
both remote and local? Or, do we skip uploading the segment to remote?

The idea is to skip the upload and update the LogStartOffset. Checking
on the uploading.

[jiafu1115]

I copied the bug description here.

Title: When remote storage keep outage status. the local segment never get deleted

Test Case:

Topic enable remote storage:
Local retention time: 10 minutes
Remote retenton time: 20 minutes

Keep run producer to send. and pick one time after 20 minutes to set the AWS S3 permisssion to let the remote upload fail.

After 1 hour. you will see the local segments keep increase and save about 1 hours even the both local/remote retention time < 20 minutes.

So we need one protect for this case to avoid the local disk keep increase forever util the outage recovered.

[kamalcph]

Thanks for the PR, @jiafu1115.

1. We cannot delete the segments in-middle when the maxTimestamp of the candidate segment is less than the retention.ms and move the log-start-offset. This will break the deletion breach by retention time logic. In UnifiedLog#deleteRetentionMsBreachedSegments, the segments are being scanned from the beginning of the log. We cannot expect that the maxTimestamp in the segments will increase monotonically.

2. This optimization can be applied only to the deletion breached by size logic and can be moved to UnifiedLog instead. Similar to the DELETE_RECORDS API, that moves the log-start-offset. The segment deletion logic due to breach by size can increment the log-start-offset instead of local-log-start-offset when the local-log-size > complete-retention-time. See UnifiedLog for more details.

[jiafu1115]

@kamalcph thanks for your comments. And I need more time to research/understand your statements and give the feedback later.
One quick question here. Do you have some other suggestion for the corner case handle: local time=remote time's case handle improvement for the KIP. Or I just need to note this case in this phase for future as I already done with KIP updated? After all. it is corner case.
WDYT?

[kamalcph]

I just need to note this case in this phase for future as I already done with KIP updated?

It is not a blocker for the KIP-1241. You can document it as same as the existing behaviour; the segment will be uploaded to remote, then allowed for local-log deletion.

[jiafu1115]

@kamalcph After a deeper dive, I understand it now. Thanks a lot for pointing this out.

The maxTimestamp in log segments may not increase monotonically due to the use of ProduceTime, or because of NTP / system clock adjustments. It is different with offset.

Based on this, I think I should update the code by changing 'continue' to "break" in the scan process with the next segment be evaluated. This way, the behavior will keep consistent with the all time-based retention deleting logic.

I can take some time to work on this change. Once I feel it’s ready, I’d really appreciate your help to review it again. For now, I’ll focus on the KIP as first.

Thanks again.

[jiafu1115]

@kamalcph Sorry to trouble you again. I’ve just completed an update to the code.
We can leave this PR until the KIP is finalized, but could you please take a quick look at the current implementation to see whether the overall flow looks reasonable to you? I’m simply very curious to know whether this approach would be viable in the future.
Thanks a lot!