Preserve failed batch cleanup tasks for retry

CHANGELOG.md

@@ -99,6 +99,10 @@ request adding CHANGELOG notes for breaking (!) changes and possibly other secti
 ### Fixes
 
 - Iceberg REST: renaming a table or view with a missing `source` or `destination` now returns `400 Bad Request` instead of `500 Internal Server Error`.
+- Batch file cleanup now retries storage deletion failures reported by `FileIO` and keeps failed
+  tasks persisted, instead of reporting success and dropping tasks while metadata files remain
+  undeleted. Iceberg's `ADLSFileIO` suppresses deletion failures, so Azure remains affected
+  ([#5482](https://github.com/apache/polaris/issues/5482)).
 - Python CLI `catalogs create --type external` now validates `--storage-type` and `--default-base-location` up front, matching the behavior for internal catalogs and the flags' documented "(Required)" status. Previously, omitting either produced an opaque pydantic `ValidationError` at request-build time.
 - Iceberg REST: server-side JSON processing failures (HTTP 500) now return the standard Iceberg
   error envelope (`{"error": {...}}`) instead of a flat `{"code", "message"}` body, so Iceberg

runtime/service/src/main/java/org/apache/polaris/service/task/FileCleanupTaskHandler.java

@@ -21,10 +21,12 @@
 import java.util.concurrent.CompletableFuture;
 import java.util.concurrent.ExecutorService;
 import java.util.concurrent.TimeUnit;
-import org.apache.iceberg.CatalogUtil;
 import org.apache.iceberg.catalog.TableIdentifier;
 import org.apache.iceberg.exceptions.NotFoundException;
 import org.apache.iceberg.io.FileIO;
+import org.apache.iceberg.io.SupportsBulkOperations;
+import org.apache.iceberg.util.Tasks;
+import org.apache.iceberg.util.ThreadPools;
 import org.apache.polaris.core.StructuredLogKeys;
 import org.apache.polaris.core.context.CallContext;
 import org.apache.polaris.core.entity.TaskEntity;
@@ -162,7 +164,26 @@ public CompletableFuture<Void> tryDelete(
       return CompletableFuture.failedFuture(e);
     }
     return CompletableFuture.runAsync(
-            () -> CatalogUtil.deleteFiles(fileIO, files, type, isConcurrent), executorService)
+            () -> {
+              // CatalogUtil.deleteFiles suppresses failures, preventing retries and causing
+              // unfinished cleanup tasks to be dropped as successful.
+              if (fileIO instanceof SupportsBulkOperations bulkIO) {
+                bulkIO.deleteFiles(files);
+              } else {
+                Tasks.foreach(files)
+                    .executeWith(isConcurrent ? ThreadPools.getWorkerPool() : null)
+                    .noRetry() // The batch retry below owns the retry budget.
+                    .run(
+                        file -> {
+                          try {
+                            fileIO.deleteFile(file);
+                          } catch (NotFoundException nfe) {
+                            // Already removed, including by an earlier attempt of this batch.
+                          }
+                        });
+              }
+            },
+            executorService)
         .exceptionallyComposeAsync(
             newEx -> {
               LOGGER

runtime/service/src/test/java/org/apache/polaris/service/task/BatchFileCleanupTaskHandlerTest.java

@@ -23,6 +23,7 @@
 import static org.assertj.core.api.Assertions.assertThatCode;
 import static org.assertj.core.api.Assertions.assertThatNoException;
 import static org.assertj.core.api.Assertions.assertThatPredicate;
+import static org.assertj.core.api.Assertions.assertThatThrownBy;
 
 import io.quarkus.test.InjectMock;
 import io.quarkus.test.junit.QuarkusMock;
@@ -31,7 +32,6 @@
 import java.io.IOException;
 import java.util.List;
 import java.util.UUID;
-import java.util.concurrent.CompletableFuture;
 import java.util.concurrent.ExecutorService;
 import java.util.concurrent.Executors;
 import java.util.concurrent.atomic.AtomicInteger;
@@ -44,8 +44,10 @@
 import org.apache.iceberg.catalog.Namespace;
 import org.apache.iceberg.catalog.TableIdentifier;
 import org.apache.iceberg.inmemory.InMemoryFileIO;
+import org.apache.iceberg.io.BulkDeletionFailureException;
 import org.apache.iceberg.io.FileIO;
 import org.apache.iceberg.io.InputFile;
+import org.apache.iceberg.io.SupportsBulkOperations;
 import org.apache.polaris.core.PolarisCallContext;
 import org.apache.polaris.core.context.CallContext;
 import org.apache.polaris.core.context.RealmContext;
@@ -55,6 +57,7 @@
 import org.junit.jupiter.api.BeforeEach;
 import org.junit.jupiter.api.Test;
 import org.junit.jupiter.params.ParameterizedTest;
+import org.junit.jupiter.params.provider.CsvSource;
 import org.junit.jupiter.params.provider.ValueSource;
 import org.mockito.Mockito;
 
@@ -273,80 +276,103 @@ public void testExistenceCheckIsPerformedOncePerFile() {
   }
 
   @ParameterizedTest
-  @ValueSource(ints = {1, 2, 3})
-  public void testCleanupWithRetries(int maxRetries) throws IOException {
-    AtomicInteger batchRetryCounter = new AtomicInteger(0);
-    FileIO fileIO =
-        new InMemoryFileIO() {
-          @Override
-          public void close() {
-            // no-op
-          }
-        };
-    TableIdentifier tableIdentifier = TableIdentifier.of(Namespace.of("db1", "schema1"), "table1");
-    Mockito.when(taskFileIOSupplier.apply(Mockito.any(), Mockito.any())).thenReturn(fileIO);
-    BatchFileCleanupTaskHandler handler =
-        new BatchFileCleanupTaskHandler(taskFileIOSupplier, executor) {
-          @Override
-          public CompletableFuture<Void> tryDelete(
-              TableIdentifier tableId,
-              FileIO fileIO,
-              Iterable<String> files,
-              String type,
-              Boolean isConcurrent,
-              Throwable e,
-              int attempt) {
-            if (attempt <= maxRetries) {
-              batchRetryCounter.incrementAndGet();
-              return tryDelete(tableId, fileIO, files, type, isConcurrent, e, attempt + 1);
-            } else {
-              return super.tryDelete(tableId, fileIO, files, type, isConcurrent, e, attempt);
-            }
-          }
-        };
-    long snapshotId = 100L;
-    ManifestFile manifestFile =
-        TaskTestUtils.manifestFile(
-            fileIO, "manifest1.avro", snapshotId, "dataFile1.parquet", "dataFile2.parquet");
-    TestSnapshot snapshot =
-        TaskTestUtils.newSnapshot(fileIO, "manifestList.avro", 1, snapshotId, 99L, manifestFile);
-    String metadataFile = "v1-49494949.metadata.json";
-    StatisticsFile statisticsFile =
-        TaskTestUtils.writeStatsFile(
-            snapshot.snapshotId(),
-            snapshot.sequenceNumber(),
-            "/metadata/" + UUID.randomUUID() + ".stats",
-            fileIO);
-    TaskTestUtils.writeTableMetadata(fileIO, metadataFile, List.of(statisticsFile), snapshot);
-    assertThat(TaskUtils.exists(statisticsFile.path(), fileIO)).isTrue();
+  @CsvSource({
+    "false, 0",
+    "false, 1",
+    "false, 2",
+    "false, 3",
+    "true, 0",
+    "true, 1",
+    "true, 2",
+    "true, 3"
+  })
+  public void testCleanupWithRetries(boolean bulk, int failures) {
+    String file = "s3://bucket/metadata.json";
+    FileIO fileIO = bulk ? Mockito.mock(SupportsBulkOperations.class) : Mockito.mock(FileIO.class);
+    InputFile inputFile = Mockito.mock(InputFile.class);
+    Mockito.when(inputFile.exists()).thenReturn(true);
+    Mockito.when(fileIO.newInputFile(file)).thenReturn(inputFile);
+    AtomicInteger deleteCalls = new AtomicInteger();
+    RuntimeException failure =
+        bulk ? new BulkDeletionFailureException(1) : new IllegalStateException("delete failed");
+    var deletion =
+        Mockito.doAnswer(
+            invocation -> {
+              if (deleteCalls.incrementAndGet() <= failures) {
+                throw failure;
+              }
+              return null;
+            });
+    if (bulk) {
+      deletion.when((SupportsBulkOperations) fileIO).deleteFiles(List.of(file));
+    } else {
+      deletion.when(fileIO).deleteFile(file);
+    }
 
+    BatchFileCleanupTaskHandler handler = newBatchFileCleanupTaskHandler(fileIO);
     TaskEntity task =
         new TaskEntity.Builder()
             .withTaskType(AsyncTaskType.BATCH_FILE_CLEANUP)
             .withData(
                 new BatchFileCleanupTaskHandler.BatchFileCleanupTask(
-                    tableIdentifier,
-                    List.of(statisticsFile.path()),
+                    TableIdentifier.of("db", "table"),
+                    List.of(file),
                     BatchFileCleanupTaskHandler.BatchFileType.TABLE_METADATA))
             .setName(UUID.randomUUID().toString())
             .build();
 
-    CompletableFuture<Void> future =
-        CompletableFuture.runAsync(
-            () -> {
-              var newTask = addTaskLocation(task);
-              assertThatPredicate(handler::canHandleTask).accepts(newTask);
-              handler.handleTask(
-                  newTask, polarisCallContext); // this will schedule the batch deletion
-            });
+    if (failures >= FileCleanupTaskHandler.MAX_ATTEMPTS) {
+      assertThatThrownBy(() -> handler.handleTask(task, polarisCallContext)).hasRootCause(failure);
+    } else {
+      assertThatNoException().isThrownBy(() -> handler.handleTask(task, polarisCallContext));
+    }
+    assertThat(deleteCalls.get())
+        .isEqualTo(Math.min(failures + 1, FileCleanupTaskHandler.MAX_ATTEMPTS));
+    if (bulk) {
+      Mockito.verify(fileIO, Mockito.never()).deleteFile(Mockito.anyString());
+    }
+  }
 
-    // Wait for all async tasks to finish
-    future.join();
+  @ParameterizedTest
+  @ValueSource(booleans = {false, true})
+  public void testPartialDeletionRetriesWithAlreadyDeletedFiles(boolean concurrent)
+      throws IOException {
+    AtomicInteger deleteCalls = new AtomicInteger();
+    String retryFile = UUID.randomUUID() + ".metadata.json";
+    FileIO fileIO =
+        new InMemoryFileIO() {
+          @Override
+          public void deleteFile(String path) {
+            if (path.equals(retryFile)
+                && deleteCalls.incrementAndGet() < FileCleanupTaskHandler.MAX_ATTEMPTS) {
+              throw new IllegalStateException("transient delete failure");
+            }
+            super.deleteFile(path);
+          }
+        };
+    List<String> files = List.of(UUID.randomUUID() + ".metadata.json", retryFile);
+    for (String file : files) {
+      fileIO.newOutputFile(file).create().close();
+    }
+    BatchFileCleanupTaskHandler handler = newBatchFileCleanupTaskHandler(fileIO);
 
-    // Ensure that retries happened as expected
-    assertThat(batchRetryCounter.get()).isEqualTo(maxRetries);
+    assertThatNoException()
+        .isThrownBy(
+            () ->
+                handler
+                    .tryDelete(
+                        TableIdentifier.of("db", "table"),
+                        fileIO,
+                        files,
+                        "table_metadata",
+                        concurrent,
+                        null,
+                        1)
+                    .join());
 
-    // Check if the file was successfully deleted after retries
-    assertThat(TaskUtils.exists(statisticsFile.path(), fileIO)).isFalse();
+    assertThat(deleteCalls.get()).isEqualTo(FileCleanupTaskHandler.MAX_ATTEMPTS);
+    for (String file : files) {
+      assertThat(TaskUtils.exists(file, fileIO)).isFalse();
+    }
   }
 }

runtime/service/src/test/java/org/apache/polaris/service/task/TaskExecutorImplTest.java

@@ -21,11 +21,19 @@
 import static org.assertj.core.api.Assertions.assertThat;
 import static org.assertj.core.api.Assertions.assertThatThrownBy;
 
+import java.io.IOException;
+import java.util.List;
+import java.util.UUID;
+import java.util.concurrent.Executors;
+import java.util.concurrent.atomic.AtomicBoolean;
 import java.util.concurrent.atomic.AtomicInteger;
+import org.apache.iceberg.catalog.TableIdentifier;
+import org.apache.iceberg.inmemory.InMemoryFileIO;
 import org.apache.polaris.core.PolarisCallContext;
 import org.apache.polaris.core.context.CallContext;
 import org.apache.polaris.core.context.RealmContext;
 import org.apache.polaris.core.entity.AsyncTaskType;
+import org.apache.polaris.core.entity.PolarisEntityType;
 import org.apache.polaris.core.entity.TaskEntity;
 import org.apache.polaris.core.persistence.PolarisMetaStoreManager;
 import org.apache.polaris.service.TestServices;
@@ -38,6 +46,7 @@
 import org.apache.polaris.service.events.listeners.InMemoryEventCollector;
 import org.junit.jupiter.api.Assertions;
 import org.junit.jupiter.api.Test;
+import org.mockito.Mockito;
 
 /** Unit tests for TaskExecutorImpl */
 public class TaskExecutorImplTest {
@@ -231,6 +240,92 @@ public void handleTask(TaskEntity task, CallContext callContext) {
     assertThat(afterEvent.attributes().getRequired(EventAttributes.TASK_SUCCESS)).isEqualTo(false);
   }
 
+  @Test
+  void failedBatchCleanupRemainsRetryable() throws IOException {
+    String realm = "myrealm";
+    TestServices services = TestServices.builder().realmContext(() -> realm).build();
+    PolarisCallContext context = services.newCallContext();
+    PolarisMetaStoreManager store = services.metaStoreManager();
+    AtomicBoolean failDeletion = new AtomicBoolean(true);
+    InMemoryFileIO fileIO =
+        new InMemoryFileIO() {
+          @Override
+          public void deleteFile(String path) {
+            if (failDeletion.get()) {
+              throw new IllegalStateException("delete failed");
+            }
+            super.deleteFile(path);
+          }
+
+          @Override
+          public void close() {
+            // Keep the files available across task attempts.
+          }
+        };
+    String file = UUID.randomUUID() + ".metadata.json";
+    fileIO.newOutputFile(file).create().close();
+    TaskFileIOSupplier supplier = Mockito.mock(TaskFileIOSupplier.class);
+    Mockito.when(supplier.apply(Mockito.any(), Mockito.any())).thenReturn(fileIO);
+    TaskEntity task =
+        new TaskEntity.Builder()
+            .setName("batch-cleanup-task")
+            .setId(store.generateNewEntityId(context).getId())
+            .setCreateTimestamp(services.clock().millis())
+            .withTaskType(AsyncTaskType.BATCH_FILE_CLEANUP)
+            .withData(
+                new BatchFileCleanupTaskHandler.BatchFileCleanupTask(
+                    TableIdentifier.of("db", "table"),
+                    List.of(file),
+                    BatchFileCleanupTaskHandler.BatchFileType.TABLE_METADATA))
+            .build();
+    assertThat(store.createEntityIfNotExists(context, null, task).isSuccess()).isTrue();
+    InMemoryEventCollector events = (InMemoryEventCollector) services.polarisEventDispatcher();
+    PolarisEventMetadata metadata = PolarisEventMetadata.builder().realmId(realm).build();
+
+    try (var deletionExecutor = Executors.newSingleThreadExecutor()) {
+      TaskExecutorImpl executor =
+          new TaskExecutorImpl(
+              Runnable::run,
+              null,
+              services.clock(),
+              services.metaStoreManagerFactory(),
+              supplier,
+              new RealmContextHolder(),
+              services.polarisEventDispatcher(),
+              services.eventMetadataFactory(),
+              null,
+              new PolarisPrincipalHolder(),
+              services.principal());
+      executor.addTaskHandler(new BatchFileCleanupTaskHandler(supplier, deletionExecutor));
+
+      assertThatThrownBy(() -> executor.handleTask(task.getId(), context, metadata, 1))
+          .hasRootCauseInstanceOf(IllegalStateException.class)
+          .hasRootCauseMessage("delete failed");
+      assertThat(TaskUtils.exists(file, fileIO)).isTrue();
+      assertThat(store.loadEntity(context, 0L, task.getId(), PolarisEntityType.TASK).getEntity())
+          .isNotNull();
+      assertThat(
+              events
+                  .getLatest(PolarisEventType.AFTER_ATTEMPT_TASK)
+                  .attributes()
+                  .getRequired(EventAttributes.TASK_SUCCESS))
+          .isFalse();
+
+      failDeletion.set(false);
+      executor.handleTask(task.getId(), context, metadata, 2);
+
+      assertThat(TaskUtils.exists(file, fileIO)).isFalse();
+      assertThat(store.loadEntity(context, 0L, task.getId(), PolarisEntityType.TASK).getEntity())
+          .isNull();
+      assertThat(
+              events
+                  .getLatest(PolarisEventType.AFTER_ATTEMPT_TASK)
+                  .attributes()
+                  .getRequired(EventAttributes.TASK_SUCCESS))
+          .isTrue();
+    }
+  }
+
   @Test
   void asyncRetryIsTriggeredWhenHandlerThrows() throws InterruptedException {
     String realm = "myrealm";