Samrat002 commented on code in PR #29129:
URL: https://github.com/apache/flink/pull/29129#discussion_r3971397481
##########
flink-filesystems/flink-s3-fs-native/src/main/java/org/apache/flink/fs/s3native/NativeS3FileSystem.java:
##########
@@ -406,6 +419,87 @@ public boolean delete(Path path, boolean recursive) throws
IOException {
}
}
+ /**
+ * Deletes every object under the given key prefix. Lists all keys with a
single flat (i.e.
+ * non-delimited) listing, then either issues one {@code DeleteObjects}
batch request per
+ * {@value #DELETE_BATCH_SIZE} keys, or falls back to one {@code
DeleteObject} request per key
+ * when {@link #deleteBatchEnabled} is {@code false} (e.g. for
S3-compatible stores that don't
+ * support multi-object delete).
+ */
+ private void deleteRecursively(S3Client s3Client, String key) throws
IOException {
+ final String prefix = key.endsWith("/") ? key : key + "/";
+ final List<String> keysToDelete = new ArrayList<>();
Review Comment:
keysToDelete retains every object key below the prefix until the entire
paginated listing completes, before issuing the first delete. Recursive
deletion of a large checkpoint/savepoint prefix can therefore consume heap
proportional to the whole subtree (and deleteBatch then creates another list
per batch), potentially OOMing the TaskManager during cleanup. IMO using a
bounded batch/page-based approach so memory remains bounded while preserving
correct pagination.
--
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: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
