This is an automated email from the ASF dual-hosted git repository.
cryptoe pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/druid.git
The following commit(s) were added to refs/heads/master by this push:
new 53e814e87c6 fix: Batch deletes in DurableStorageCleaner to avoid
overlord OOM (#20425)
53e814e87c6 is described below
commit 53e814e87c6b28d5ee9710f36a61565b7456be89
Author: Akshat Jain <[email protected]>
AuthorDate: Fri Sep 25 16:10:47 2026 +0530
fix: Batch deletes in DurableStorageCleaner to avoid overlord OOM (#20425)
* Batch deletes in DurableStorageCleaner to avoid overlord OOM
* Fix forbidden check
* Minor changes
---
.../indexing/cleaner/DurableStorageCleaner.java | 22 +++++++-----
.../msq/indexing/DurableStorageCleanerTest.java | 40 ++++++++++++++++++++++
2 files changed, 54 insertions(+), 8 deletions(-)
diff --git
a/multi-stage-query/src/main/java/org/apache/druid/msq/indexing/cleaner/DurableStorageCleaner.java
b/multi-stage-query/src/main/java/org/apache/druid/msq/indexing/cleaner/DurableStorageCleaner.java
index 525da3c63fb..dae4848630c 100644
---
a/multi-stage-query/src/main/java/org/apache/druid/msq/indexing/cleaner/DurableStorageCleaner.java
+++
b/multi-stage-query/src/main/java/org/apache/druid/msq/indexing/cleaner/DurableStorageCleaner.java
@@ -50,6 +50,7 @@ import java.util.stream.Collectors;
public class DurableStorageCleaner implements OverlordDuty
{
private static final Logger LOG = new Logger(DurableStorageCleaner.class);
+ private static final int DELETE_BATCH_SIZE = 10000;
private final DurableStorageCleanerConfig config;
private final StorageConnector storageConnector;
@@ -101,6 +102,7 @@ public class DurableStorageCleaner implements OverlordDuty
.collect(Collectors.toSet());
Set<String> filesToRemove = new HashSet<>();
+ int removedCount = 0;
while (allFiles.hasNext()) {
String currentFile = allFiles.next();
String nextDirName =
DurableStorageUtils.getNextDirNameWithPrefixFromPath(currentFile);
@@ -112,19 +114,23 @@ public class DurableStorageCleaner implements OverlordDuty
// do nothing, query results should not be cleaned if the task is
created before retain period
} else {
filesToRemove.add(currentFile);
+
+ if (filesToRemove.size() >= DELETE_BATCH_SIZE) {
+ storageConnector.deleteFiles(filesToRemove);
+ removedCount += filesToRemove.size();
+ filesToRemove.clear();
+ LOG.info("Removed [%d] files so far which are not associated with
any MSQ task.", removedCount);
+ }
}
}
}
- if (filesToRemove.isEmpty()) {
- LOG.info("There are no leftover directories to delete.");
- } else {
- LOG.info(
- "Removing [%d] files which are not associated with any MSQ task.",
- filesToRemove.size()
- );
- LOG.debug("Files to remove:\n[%s]\n", filesToRemove);
+ if (!filesToRemove.isEmpty()) {
storageConnector.deleteFiles(filesToRemove);
+ removedCount += filesToRemove.size();
+ LOG.info("Removed [%d] files so far which are not associated with any
MSQ task.", removedCount);
+ } else if (removedCount == 0) {
+ LOG.info("There are no leftover directories to delete.");
}
}
diff --git
a/multi-stage-query/src/test/java/org/apache/druid/msq/indexing/DurableStorageCleanerTest.java
b/multi-stage-query/src/test/java/org/apache/druid/msq/indexing/DurableStorageCleanerTest.java
index 5f4d804ed69..e2524f06a06 100644
---
a/multi-stage-query/src/test/java/org/apache/druid/msq/indexing/DurableStorageCleanerTest.java
+++
b/multi-stage-query/src/test/java/org/apache/druid/msq/indexing/DurableStorageCleanerTest.java
@@ -45,6 +45,8 @@ import org.mockito.ArgumentCaptor;
import org.mockito.ArgumentMatchers;
import org.mockito.Mockito;
+import java.util.ArrayList;
+import java.util.HashSet;
import java.util.List;
import java.util.Set;
@@ -135,6 +137,44 @@ public class DurableStorageCleanerTest
Assertions.assertEquals(Sets.newHashSet(STRAY_DIR, intermediateFilesPath),
capturedArguments.getValue());
}
+ @Test
+ public void testRunDeletesInBatchesAndKeepsTheRemainder() throws Exception
+ {
+ final List<String> strayFiles = new ArrayList<>();
+ for (int i = 0; i < 10_001; i++) {
+ strayFiles.add(STRAY_DIR + i);
+ }
+
+ final List<String> listing = ImmutableList.<String>builder()
+ .addAll(strayFiles)
+ .add(DurableStorageUtils.getControllerDirectory(TASK_ID))
+ .build();
+
+ Mockito.when(TASK_STORAGE.getCompletedTasksInfo(ArgumentMatchers.any(),
ArgumentMatchers.isNull())).thenReturn(List.of());
+
Mockito.when(STORAGE_CONNECTOR.listDir(ArgumentMatchers.anyString())).thenReturn(listing.iterator());
+ Mockito.when(TASK_RUNNER_WORK_ITEM.getTaskId()).thenReturn(TASK_ID);
+
Mockito.doReturn(ImmutableList.of(TASK_RUNNER_WORK_ITEM)).when(TASK_RUNNER).getRunningTasks();
+
Mockito.when(TASK_MASTER.getTaskRunner()).thenReturn(Optional.of(TASK_RUNNER));
+
+ final Set<String> deleted = new HashSet<>();
+ final List<Integer> batchSizes = new ArrayList<>();
+ Mockito.doAnswer(invocation -> {
+ final Iterable<String> batch = invocation.getArgument(0);
+ int size = 0;
+ for (String file : batch) {
+ deleted.add(file);
+ size++;
+ }
+ batchSizes.add(size);
+ return null;
+ }).when(STORAGE_CONNECTOR).deleteFiles(ArgumentMatchers.any());
+
+ durableStorageCleaner.run();
+
+ Assertions.assertTrue(batchSizes.size() > 1, "expected more than one
batch, got " + batchSizes);
+ Assertions.assertEquals(Sets.newHashSet(strayFiles), deleted);
+ }
+
@Test
public void testGetSchedule()
{
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
