nevzheng opened a new issue, #14333: URL: https://github.com/apache/gravitino/issues/14333
### Version main branch Affects 1.3.0, 1.3.1 and `main` (checked at [`e2ed33cb`](https://github.com/apache/gravitino/commit/e2ed33cb7deb0f92dc5600860830bb8223939520); the same line is in [v1.3.0](https://github.com/apache/gravitino/blob/40fdf6ab96ac87b47e6d3e14e7c4dc0d815e68f0/iceberg/iceberg-rest-server/src/main/java/org/apache/gravitino/iceberg/service/cleanup/IcebergCleanupManager.java#L237)). ### Describe what's wrong When storage refuses deletes, async table purge (`DELETE ...?purgeRequested=true` with `X-Gravitino-Async-Purge: true`) marks the cleanup job `SUCCEEDED`. It records no error and does not retry, and the table name is released while every file of the old table is still on storage. **Why care** - **The status is wrong.** `SUCCEEDED` is the operator's signal that the data is gone. Here it means "we tried": every file is still there. - **Storage keeps costing money.** The data, manifests and metadata stay on storage indefinitely, and nothing records that they exist. - **The name moves onto old files.** A `SUCCEEDED` job frees the name at once, so a recreated `db.t` lands on the same `<warehouse>/db/t` prefix, next to the old files. A later manual cleanup of "leftovers" under that prefix would also delete the new table's files. - **Operators get no signal.** `last_error` is empty and `attempts` is 0. The only trace is a `WARN Failed to bulk delete ... cleanup files` log line. **Why this is a bug and not a choice** Sync purge has the same best-effort behaviour, since Iceberg's own [`CatalogUtil.dropTableData`](https://github.com/apache/iceberg/blob/apache-iceberg-1.11.0/core/src/main/java/org/apache/iceberg/CatalogUtil.java#L101-L103) logs and continues past delete failures. Async purge is different: it adds a tracked job with retries, `last_error` and a `FAILED` state, and the design defines `SUCCEEDED` strictly: - [Design doc, Finish](https://github.com/apache/gravitino/blob/e2ed33cb7deb0f92dc5600860830bb8223939520/design-docs/async-iceberg-rest-hard-deletion.md#L360): "Once every reachable file is gone the job is `SUCCEEDED`". The [outcome table](https://github.com/apache/gravitino/blob/e2ed33cb7deb0f92dc5600860830bb8223939520/design-docs/async-iceberg-rest-hard-deletion.md#L484-L488) says the same: "All files deleted (or already gone)". - The code intends failures to propagate. `deleteAll` has a [`catch (ExecutionException)`](https://github.com/apache/gravitino/blob/e2ed33cb7deb0f92dc5600860830bb8223939520/iceberg/iceberg-rest-server/src/main/java/org/apache/gravitino/iceberg/service/cleanup/IcebergCleanupManager.java#L245-L250) that skips `NotFoundException` and otherwise throws "Bulk delete batch failed". That branch can never run. **Where it goes wrong** 1. [`deleteAll`](https://github.com/apache/gravitino/blob/e2ed33cb7deb0f92dc5600860830bb8223939520/iceberg/iceberg-rest-server/src/main/java/org/apache/gravitino/iceberg/service/cleanup/IcebergCleanupManager.java#L228-L237) deletes each batch with `CatalogUtil.deleteFiles(io, batch, "cleanup", true)`. 2. In Iceberg 1.11, [`CatalogUtil.deleteFiles`](https://github.com/apache/iceberg/blob/apache-iceberg-1.11.0/core/src/main/java/org/apache/iceberg/CatalogUtil.java#L222-L248) catches and logs every failure: `BulkDeletionFailureException` and any other `RuntimeException` on the bulk path, and per-file failures on the non-bulk path (`suppressFailureWhenFinished`). It never throws. 3. So [`runJob`](https://github.com/apache/gravitino/blob/e2ed33cb7deb0f92dc5600860830bb8223939520/iceberg/iceberg-rest-server/src/main/java/org/apache/gravitino/iceberg/service/cleanup/IcebergCleanupManager.java#L280-L293) never reaches `recordFailure` and calls `markSucceeded` instead. 4. [`selectUnfinishedJobId`](https://github.com/apache/gravitino/blob/e2ed33cb7deb0f92dc5600860830bb8223939520/iceberg/iceberg-rest-server/src/main/java/org/apache/gravitino/iceberg/service/cleanup/mapper/provider/base/IcebergCleanupJobBaseSQLProvider.java#L163-L174) only matches `PENDING`/`RUNNING`, so the name is released. ```mermaid stateDiagram-v2 [*] --> PENDING : "async drop, 204" PENDING --> RUNNING : "worker claims job" RUNNING --> SUCCEEDED : "expected: every file deleted" RUNNING --> PENDING : "expected: delete failed, retry (attempts+1)" RUNNING --> FAILED : "expected: delete failed at max-attempts" RUNNING --> SUCCEEDED : "actual: deletes denied, failure swallowed" note right of SUCCEEDED : "name released at once; files may still exist" ``` **Realistic triggers:** credentials or a bucket policy without `s3:DeleteObject` (least-privilege roles, a read-only bucket after a migration), throttling that is still failing once the SDK gives up retrying on a large table, and transient network errors on part of a batch. S3 Object Lock is not a trigger: in a versioned bucket, `DeleteObject` without a version id writes a delete marker and succeeds. ### Error message and/or stacktrace There is no stacktrace and no error on the job, only warnings, then `SUCCEEDED`: ``` WARN [iceberg-cleanup-delete] CatalogUtil:228 - Failed to bulk delete 1 cleanup files ... (5 times, one per batch) ``` ### How to reproduce I reproduced this with a JUnit test in `iceberg-rest-server` on the H2 backend. The test uses the real `IcebergTableOperationExecutor`, a `CatalogWrapperForREST` on a JDBC/SQLite Iceberg catalog with a local warehouse, and the real `IcebergCleanupManager` worker and job store. The only test double is the FileIO: a `HadoopFileIO` subclass that denies deletes, standing in for a role without `DeleteObject`. `HadoopFileIO` implements `SupportsBulkOperations`, so this exercises the bulk path that S3FileIO uses. Manual steps: 1. Run the Iceberg REST service as an auxiliary service (async cleanup is only wired in that mode) with a relational entity store. 2. Create `db.t` and append a data file. 3. Remove delete permission from the warehouse credentials, but keep read permission. 4. Send `DELETE .../namespaces/db/tables/t?purgeRequested=true` with `X-Gravitino-Async-Purge: true`. 5. `SELECT state, attempts, last_error FROM iceberg_cleanup_job` returns `SUCCEEDED`, `0`, `NULL`. 6. All 5 files of the old table (data file, manifest, manifest list, 2 × `metadata.json`, 14,049 bytes) are still under `<warehouse>/db/t/`. 7. `POST .../namespaces/db/tables` with `name: t` succeeds, at the same `<warehouse>/db/t` location. <details> <summary>Test (FileIO fault injection and the test method, trimmed from the full file)</summary> ```java /** Local HadoopFileIO that can simulate a role without DeleteObject. */ public static class FaultyFileIO extends HadoopFileIO { public enum Mode { NORMAL, DENY_DELETE } static volatile Mode mode = Mode.NORMAL; public FaultyFileIO() { super(new Configuration()); } @Override public void deleteFile(String path) { if (mode != Mode.NORMAL) { throw new RuntimeIOException("Access Denied (simulated DeleteObject): %s", path); } super.deleteFile(path); } @Override public void deleteFiles(Iterable<String> paths) throws BulkDeletionFailureException { if (mode != Mode.NORMAL) { throw new BulkDeletionFailureException(Iterables.size(paths)); } super.deleteFiles(paths); } } @TestTemplate void testDeleteOnlyFailureIsReportedAsSucceeded() throws Exception { OldTable old = createTableWithData(T); // creates db.t, appends one data file, lists its files executor.dropTable(context(), T, true); // context() has asyncPurge() == true long jobId = store.findUnfinishedJobId(CATALOG_ID, "db", "t").orElseThrow(); FaultyFileIO.mode = FaultyFileIO.Mode.DENY_DELETE; manager.start(); Awaitility.await().atMost(20, TimeUnit.SECONDS).until(() -> { IcebergCleanupJob.State s = store.stateOf(jobId); return s != IcebergCleanupJob.State.PENDING && s != IcebergCleanupJob.State.RUNNING; }); Assertions.assertEquals(IcebergCleanupJob.State.SUCCEEDED, store.stateOf(jobId)); // passes today FaultyFileIO.mode = FaultyFileIO.Mode.NORMAL; old.assertAllFilesStillExist(); // every file is still on disk LoadTableResponse recreated = executor.createTable(context(), DB, createReq()); Assertions.assertEquals(old.location, recreated.tableMetadata().location()); // same prefix } ``` The test class extends `TestJDBCBackend` and wires the executor to the cleanup manager with `IO_IMPL = FaultyFileIO`, `worker-threads=1`, `poll-interval-secs=1`, `max-attempts=2`. ``` ./gradlew :iceberg:iceberg-rest-server:test --tests '*TestIcebergAsyncPurgeFailedCleanupRepro.testDeleteOnlyFailureIsReportedAsSucceeded' -PskipITs -PskipDockerTests=true ``` Output on `e2ed33cb` (passes, which shows the bug): ``` [repro] delete-only failure: job 1589609940454036763 SUCCEEDED, denied deletes=5, leftovers: [.../db/t/data/00000-0-old.parquet, .../metadata/00000-....metadata.json, .../metadata/00001-....metadata.json, ...-m0.avro, snap-....avro] BUILD SUCCESSFUL ``` </details> ### Additional context **Proposed fix.** Delete each batch with a helper that throws, so the existing retry and `FAILED` path applies: ```diff - batch -> - futures.add( - deleteExecutor.submit( - () -> CatalogUtil.deleteFiles(io, batch, "cleanup", true)))); + batch -> futures.add(deleteExecutor.submit(() -> deleteBatch(io, batch)))); ... + // Unlike CatalogUtil.deleteFiles, which only logs delete failures, this throws so the attempt + // fails and the job retries or becomes FAILED instead of SUCCEEDED with files left behind. + private static void deleteBatch(FileIO io, List<String> batch) { + if (io instanceof SupportsBulkOperations) { + ((SupportsBulkOperations) io).deleteFiles(batch); + return; + } + Tasks.foreach(batch) + .noRetry() + .throwFailureWhenFinished() + .run( + file -> { + try { + io.deleteFile(file); + } catch (NotFoundException alreadyGone) { + // Already deleted by an earlier attempt. + } + }); + } ``` I tried this locally: - With it, a delete-only failure ends `FAILED` after `max-attempts` instead of `SUCCEEDED`. - An attempt with denied deletes throws, and a retry once access is back deletes all 5 files (14,049 bytes). - All 10 existing `TestIcebergCleanupManager` tests still pass. The only failure among the 16 tests run is the repro above, which asserts the old behaviour. Retries stay safe because levels are deleted leaf-first, with the root `metadata.json` last, so each attempt re-enumerates from an intact root. Bulk deletes of keys that are already gone succeed on S3 and `FileSystem`, so `NotFoundException` only needs handling on the non-bulk path. On that path, `throwFailureWhenFinished` still deletes everything it can before failing. Two smaller notes for the fix: - `BulkDeletionFailureException` carries only a count, so `last_error` will read "Failed to delete N files". The per-file causes are already logged by the FileIO. - A delete failure while data files are being streamed is currently re-wrapped as `Failed to read manifest <path>` ([`IcebergCleanupManager.java#L359-L360`](https://github.com/apache/gravitino/blob/e2ed33cb7deb0f92dc5600860830bb8223939520/iceberg/iceberg-rest-server/src/main/java/org/apache/gravitino/iceberg/service/cleanup/IcebergCleanupManager.java#L359-L360)). With the fix, that `catch (Exception e)` should keep the delete cause visible, so `last_error` doesn't point operators at the manifest. What happens after a job does reach `FAILED` is a separate problem: the name is released then too, while files may remain. @roryqi, could you take a look? You know the design trade-offs best. -- 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]
