[GitHub] [kafka] satishd commented on a diff in pull request #14349: KAFKA-15352: Update log-start-offset before initiating deletion of remote segments

[via GitHub]

satishd commented on code in PR #14349:
URL: https://github.com/apache/kafka/pull/14349#discussion_r1319375686


##########
core/src/main/java/kafka/log/remote/RemoteLogManager.java:
##########
@@ -818,56 +818,51 @@ public 
RemoteLogRetentionHandler(Optional<RetentionSizeData> retentionSizeData,
                 remainingBreachedSize = retentionSizeData.map(sizeData -> 
sizeData.remainingBreachedSize).orElse(0L);
             }
 
-            private boolean 
deleteRetentionSizeBreachedSegments(RemoteLogSegmentMetadata metadata) throws 
RemoteStorageException, ExecutionException, InterruptedException {
+            private boolean 
deleteRetentionSizeBreachedSegments(RemoteLogSegmentMetadata metadata) {
                 if (!retentionSizeData.isPresent()) {
                     return false;
                 }
 
-                boolean isSegmentDeleted = deleteRemoteLogSegment(metadata, x 
-> {
-                    // Assumption that segments contain size >= 0
-                    if (remainingBreachedSize > 0) {
-                        long remainingBytes = remainingBreachedSize - 
x.segmentSizeInBytes();
-                        if (remainingBytes >= 0) {
-                            remainingBreachedSize = remainingBytes;
-                            return true;
-                        }
+                boolean isSegmentDeleted = false;
+                // Assumption that segments contain size >= 0
+                if (remainingBreachedSize > 0) {
+                    long remainingBytes = remainingBreachedSize - 
metadata.segmentSizeInBytes();
+                    if (remainingBytes >= 0) {
+                        remainingBreachedSize = remainingBytes;
+                        isSegmentDeleted = true;
                     }
+                }
 
-                    return false;
-                });
                 if (isSegmentDeleted) {
                     logStartOffset = OptionalLong.of(metadata.endOffset() + 1);
-                    logger.info("Deleted remote log segment {} due to 
retention size {} breach. Log size after deletion will be {}.",
+                    logger.info("About to delete remote log segment {} due to 
retention size {} breach. Log size after deletion will be {}.",
                             metadata.remoteLogSegmentId(), 
retentionSizeData.get().retentionSize, remainingBreachedSize + 
retentionSizeData.get().retentionSize);
                 }
                 return isSegmentDeleted;
             }
 
-            public boolean 
deleteRetentionTimeBreachedSegments(RemoteLogSegmentMetadata metadata)
-                    throws RemoteStorageException, ExecutionException, 
InterruptedException {
+            public boolean 
deleteRetentionTimeBreachedSegments(RemoteLogSegmentMetadata metadata) {
                 if (!retentionTimeData.isPresent()) {
                     return false;
                 }
 
-                boolean isSegmentDeleted = deleteRemoteLogSegment(metadata,
-                        x -> x.maxTimestampMs() <= 
retentionTimeData.get().cleanupUntilMs);
+                boolean isSegmentDeleted = metadata.maxTimestampMs() <= 
retentionTimeData.get().cleanupUntilMs;

Review Comment:
   Please change in other delete retention check methods.



-- 
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