[GitHub] [kafka] showuon commented on a diff in pull request #14330: KAFKA-15410: Delete records with tiered storage integration test (4/4)

[via GitHub]

showuon commented on code in PR #14330:
URL: https://github.com/apache/kafka/pull/14330#discussion_r1317166423


##########
core/src/main/java/kafka/log/remote/RemoteLogManager.java:
##########
@@ -954,24 +953,29 @@ private void cleanupExpiredRemoteLogSegments() throws 
RemoteStorageException, Ex
 
             RemoteLogRetentionHandler remoteLogRetentionHandler = new 
RemoteLogRetentionHandler(retentionSizeData, retentionTimeData);
             Iterator<Integer> epochIterator = 
epochWithOffsets.navigableKeySet().iterator();
-            boolean isSegmentDeleted = true;
-            while (isSegmentDeleted && epochIterator.hasNext()) {
+            boolean canProcess = true;
+            while (canProcess && epochIterator.hasNext()) {
                 Integer epoch = epochIterator.next();
                 Iterator<RemoteLogSegmentMetadata> segmentsIterator = 
remoteLogMetadataManager.listRemoteLogSegments(topicIdPartition, epoch);
-                while (isSegmentDeleted && segmentsIterator.hasNext()) {
+                while (canProcess && segmentsIterator.hasNext()) {
                     if (isCancelled() || !isLeader()) {
                         logger.info("Returning from remote log segments 
cleanup for the remaining segments as the task state is changed.");
                         return;
                     }
                     RemoteLogSegmentMetadata metadata = 
segmentsIterator.next();
 
                     // check whether the segment contains the required epoch 
range with in the current leader epoch lineage.
-                    if (isRemoteSegmentWithinLeaderEpochs(metadata, 
logEndOffset, epochWithOffsets)) {
+                    boolean isValidSegment = 
isRemoteSegmentWithinLeaderEpochs(metadata, logEndOffset, epochWithOffsets);
+                    boolean isSegmentDeleted = false;
+                    if (isValidSegment) {
                         isSegmentDeleted =
                                 
remoteLogRetentionHandler.deleteRetentionTimeBreachedSegments(metadata) ||
-                                        
remoteLogRetentionHandler.deleteRetentionSizeBreachedSegments(metadata) ||
-                                        
remoteLogRetentionHandler.deleteLogStartOffsetBreachedSegments(metadata, 
logStartOffset);
+                                        
remoteLogRetentionHandler.deleteRetentionSizeBreachedSegments(metadata);
+                    }
+                    if (!isSegmentDeleted) {
+                        isSegmentDeleted = 
remoteLogRetentionHandler.deleteLogStartOffsetBreachedSegments(metadata, 
logStartOffset);

Review Comment:
   Is it possible we run `deleteLogStartOffsetBreachedSegments` first before 
sizeBreach/timeBreach? Ex:
   
   ```
   isSegmentDeleted = 
remoteLogRetentionHandler.deleteLogStartOffsetBreachedSegments(
                                   metadata, logStartOffset, epochWithOffsets);
   
    boolean isValidSegment = isRemoteSegmentWithinLeaderEpochs(metadata, 
logEndOffset, epochWithOffsets);
   
   if (isValidSegment) {
                           isSegmentDeleted =         
remoteLogRetentionHandler.deleteRetentionTimeBreachedSegments(metadata) || 
remoteLogRetentionHandler.deleteRetentionSizeBreachedSegments(metadata));
   ```



-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: jira-unsubscr...@kafka.apache.org

For queries about this service, please contact Infrastructure at:
us...@infra.apache.org