diff --git a/CHANGELOG.md b/CHANGELOG.md index 5085cc7dc0c..71754dd8b52 100644 --- a/CHANGELOG.md +++ b/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 diff --git a/runtime/service/src/main/java/org/apache/polaris/service/task/FileCleanupTaskHandler.java b/runtime/service/src/main/java/org/apache/polaris/service/task/FileCleanupTaskHandler.java index 86c9c845198..1e4c3e8b4f0 100644 --- a/runtime/service/src/main/java/org/apache/polaris/service/task/FileCleanupTaskHandler.java +++ b/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 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 diff --git a/runtime/service/src/test/java/org/apache/polaris/service/task/BatchFileCleanupTaskHandlerTest.java b/runtime/service/src/test/java/org/apache/polaris/service/task/BatchFileCleanupTaskHandlerTest.java index 4d3372719bf..f428c4b3303 100644 --- a/runtime/service/src/test/java/org/apache/polaris/service/task/BatchFileCleanupTaskHandlerTest.java +++ b/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 tryDelete( - TableIdentifier tableId, - FileIO fileIO, - Iterable 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 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 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(); + } } } diff --git a/runtime/service/src/test/java/org/apache/polaris/service/task/TaskExecutorImplTest.java b/runtime/service/src/test/java/org/apache/polaris/service/task/TaskExecutorImplTest.java index 91369e647f9..012ba93c6fe 100644 --- a/runtime/service/src/test/java/org/apache/polaris/service/task/TaskExecutorImplTest.java +++ b/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";