nevzheng opened a new issue, #14394: URL: https://github.com/apache/gravitino/issues/14394
### Describe the proposal This epic reports an audit of the async Iceberg REST table purge (`IcebergCleanupManager`, `iceberg_cleanup_job`, opt-in with `X-Gravitino-Async-Purge`; design in `design-docs/async-iceberg-rest-hard-deletion.md`). The audit found five design weak points. The bugs found are listed under each weak point as symptoms, each with a reproducer, and are not filed separately. **Scope.** In scope: the drop path in [IcebergTableOperationExecutor](https://github.com/apache/gravitino/blob/bb634defdc5663a8124be5fca1c8d9e52ac096ac/iceberg/iceberg-rest-server/src/main/java/org/apache/gravitino/iceberg/service/dispatcher/IcebergTableOperationExecutor.java#L133-L170), the name guard on create, register and rename, the worker, retry and heartbeat logic in [IcebergCleanupManager](https://github.com/apache/gravitino/blob/bb634defdc5663a8124be5fca1c8d9e52ac096ac/iceberg/iceberg-rest-server/src/main/java/org/apache/gravitino/iceberg/service/cleanup/IcebergCleanupManager.java), and the job store SQL. Out of scope: views, sync purge, and the maintenance jobs. **Method.** 1. Wrote a state model of one table's lifecycle: the catalog entry, metadata and data files, and the cleanup row (`PENDING`, `RUNNING`, `SUCCEEDED`, `FAILED`). It includes the events that can interleave: commits, create, register, crashes, graceful restarts, storage outages, and row pruning. 2. Explored every interleaving within small bounds with a breadth-first model checker, against four rules: - a live table never loses a file it needs; - every file of a dropped table is either deleted or tracked by a row; - `SUCCEEDED` means no files are left; - `FAILED` happens only after a lasting storage fault. The model also finds #14333 and #14388 on its own. 3. Turned each counterexample into a JUnit test against the real classes (H2 backend). Each test asserts the current behavior, so a passing test means the problem reproduces. 4. Grouped the results by root cause. A group counts as a weak point only if one design change would remove all of its symptoms. Each weak point was also used to predict new failures, which were then tested (v3 deletion vectors, a 403 error, a table created at the dropped location). One prediction did not hold, and it is listed below. ```mermaid stateDiagram-v2 direction LR [*] --> Live Live --> Captured: drop with purge reads metadata location Captured --> Unlisted: catalog entry removed Unlisted --> Pending: cleanup row inserted Unlisted --> Untracked: row insert fails (W3) Pending --> Running: worker claims Running --> Pending: failure, retried at once (W4) Running --> Failed: attempts used up (W2, W4) Running --> Succeeded: files deleted (W1, W2 can hit live files) Failed --> Untracked: name released (#14388) Succeeded --> [*] Untracked --> [*]: files remain ``` **Priority scale.** P1: can destroy or expose data of a live table. P2: leaks files or blocks purging valid tables. P3: operational robustness. Priority reflects impact if hit, not scheduling. Async purge is opt-in and runs only in auxiliary mode, and none of these is a release blocker. This epic documents weak points for discussion and later prioritization; no fix is scheduled as part of this report. **Weak points** | Weak point | Priority | | ---------- | -------- | | #14389 The reservation is on the table name, not on the files | P1 | | #14390 Purge logic differs from `CatalogUtil.dropTableData` | P1 | | #14391 Catalog drop and cleanup-row insert are not atomic | P2 | | #14392 All failures are handled alike and retried at once | P3 | | #14393 Cleanup rows copy the catalog properties | P1 | **Observed symptoms** (one row per reproducer) | Symptom | Weak point | Test | Effect | | ------- | ---------- | ---- | ------ | | Registering a dropped table's metadata under a new name before the job runs; the job then deleted the registered table's data | W1 | `testF1RegisterDroppedMetadataUnderNewNameLosesData` | Data loss on a live table | | Rename onto a name being purged is not checked | W1 | code reading | Name guard bypassed | | A `FAILED` job releases the name while files may remain | W1 | #14388 (already tracked) | Untracked files | | `gc.enabled=false` ignored; the purge deleted a data file that sync purge kept | W2 | `testF2AsyncPurgeIgnoresGcDisabled` | Data loss in another table (for example a Spark snapshot's source) | | v2 delete manifests: every attempt failed after deleting part of the data | W2 | `testF3DeleteManifestMakesPurgeFailEveryTime` | Tables with row-level deletes are never purged | | v3 deletion vectors: same failure | W2 | `testN2DeletionVectorsAlsoBreakPurge` | Same | | Failed deletes swallowed, job `SUCCEEDED` | W2 | #14333 (already tracked) | Untracked files | | Row insert failed after the catalog drop; table gone, no row | W3 | `TestAsyncPurgeDropNotAtomic` | Untracked files, name freed | | A commit between metadata read and drop was not purged; job `SUCCEEDED` | W3 | `testF5CommitBetweenReadAndDropLeaks` | Untracked files (sync purge has a similar window) | | 5 attempts within 26 ms during a 2 s outage | W4 | `testF6ShortOutageExhaustsAllAttempts` | Spurious `FAILED` | | A graceful restart counted as a failed attempt | W4 | `testF7RestartsCountAsFailedAttempts` | Spurious `FAILED` | | A 403 used all 5 attempts within 28 ms | W4 | `testN4PermissionDeniedBurnsAllAttempts` | Spurious `FAILED` | | Catalog secrets stored in plain text in `file_io_props` | W5 | `testF8CatalogSecretsStoredInCleanupRow` | Secrets stored in plain text for up to 720 h after the job ends | | A table created at the dropped table's location kept its files (prediction checked, did not hold) | W1 | `testN1aCreateAtDroppedLocationIsSafeForPurge` | None | **Related.** #12392 and #12393 (IRC soft delete roadmap and design) plan to build on this cleanup path. W1, W3 and W4 in particular decide what soft delete can safely reuse. **Reproducers.** Copy the classes below into `iceberg/iceberg-rest-server/src/test/java/org/apache/gravitino/iceberg/service/{cleanup,dispatcher}/` and run: ``` ./gradlew :iceberg:iceberg-rest-server:test --tests '*TestAsyncPurge*' -PskipITs ``` <details><summary>TestAsyncPurgeFailureModes.java</summary> ```java /* * Licensed to the Apache Software Foundation (ASF) under one * or more contributor license agreements. See the NOTICE file * distributed with this work for additional information * regarding copyright ownership. The ASF licenses this file * to you under the Apache License, Version 2.0 (the * "License"); you may not use this file except in compliance * with the License. You may obtain a copy of the License at * * http://www.apache.org/licenses/LICENSE-2.0 * * Unless required by applicable law or agreed to in writing, * software distributed under the License is distributed on an * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY * KIND, either express or implied. See the License for the * specific language governing permissions and limitations * under the License. */ package org.apache.gravitino.iceberg.service.cleanup; import com.google.common.collect.ImmutableMap; import java.util.Collections; import java.util.HashMap; import java.util.List; import java.util.Map; import java.util.concurrent.CopyOnWriteArrayList; import java.util.concurrent.CountDownLatch; import java.util.concurrent.TimeUnit; import org.apache.commons.lang3.reflect.FieldUtils; import org.apache.gravitino.GravitinoEnv; import org.apache.gravitino.iceberg.common.IcebergConfig; import org.apache.gravitino.iceberg.service.cleanup.po.IcebergCleanupJobPO; import org.apache.gravitino.storage.RandomIdGenerator; import org.apache.gravitino.storage.relational.TestJDBCBackend; import org.apache.iceberg.BaseTable; import org.apache.iceberg.CatalogUtil; import org.apache.iceberg.DataFile; import org.apache.iceberg.DataFiles; import org.apache.iceberg.DeleteFile; import org.apache.iceberg.FileFormat; import org.apache.iceberg.FileMetadata; import org.apache.iceberg.PartitionSpec; import org.apache.iceberg.Schema; import org.apache.iceberg.Table; import org.apache.iceberg.TableMetadata; import org.apache.iceberg.TableProperties; import org.apache.iceberg.catalog.Namespace; import org.apache.iceberg.catalog.TableIdentifier; import org.apache.iceberg.inmemory.InMemoryCatalog; import org.apache.iceberg.inmemory.InMemoryFileIO; import org.apache.iceberg.io.FileIO; import org.apache.iceberg.io.InputFile; import org.apache.iceberg.io.OutputFile; import org.apache.iceberg.io.SupportsBulkOperations; import org.apache.iceberg.types.Types; import org.awaitility.Awaitility; import org.junit.jupiter.api.AfterAll; import org.junit.jupiter.api.Assertions; import org.junit.jupiter.api.BeforeAll; import org.junit.jupiter.api.BeforeEach; import org.junit.jupiter.api.TestTemplate; import org.slf4j.Logger; import org.slf4j.LoggerFactory; /** * Repro tests for async purge failure modes found by the state model in the async purge audit epic. Each test * asserts the CURRENT behavior, so a passing test means the problem reproduces. */ class TestAsyncPurgeFailureModes extends TestJDBCBackend { private static final Logger LOG = LoggerFactory.getLogger(TestAsyncPurgeFailureModes.class); private static final long CATALOG_ID = 7L; private static final Schema SCHEMA = new Schema(Types.NestedField.required(1, "id", Types.IntegerType.get())); private IcebergCleanupJobStore store; private Object originalConfig; private Object originalIdGenerator; @BeforeAll public void snapshotGravitinoEnv() throws IllegalAccessException { originalConfig = FieldUtils.readField(GravitinoEnv.getInstance(), "config", true); originalIdGenerator = FieldUtils.readField(GravitinoEnv.getInstance(), "idGenerator", true); } @AfterAll public void restoreGravitinoEnv() throws IllegalAccessException { FieldUtils.writeField(GravitinoEnv.getInstance(), "config", originalConfig, true); FieldUtils.writeField(GravitinoEnv.getInstance(), "idGenerator", originalIdGenerator, true); } @BeforeEach public void prepareStore() { store = new IcebergCleanupJobStore(new RandomIdGenerator()); } // F1: a dropped table's metadata.json can be registered under a new name while its purge is // pending. The purge then deletes the new table's files. @TestTemplate void testF1RegisterDroppedMetadataUnderNewNameLosesData() { InMemoryCatalog catalog = newCatalog("f1"); TableIdentifier t = TableIdentifier.of("db", "t"); TableIdentifier t2 = TableIdentifier.of("db", "t2"); String data = "memory://f1/db/t/data/d1.parquet"; BaseTable table = createWithFile(catalog, t, data, ImmutableMap.of()); String metadataLocation = table.operations().current().metadataFileLocation(); FileIO io = table.io(); // Async drop: catalog entry removed (files kept), PENDING row written for the old name. catalog.dropTable(t, false); store.addJob(job("t", metadataLocation)); IcebergCleanupManager svc = new IcebergCleanupManager(store, new IcebergConfig(new HashMap<>())); try { Assertions.assertTrue(svc.isNameOccupied(CATALOG_ID, "db", "t")); // The register guard only checks the target name; t2 is free. Assertions.assertFalse(svc.isNameOccupied(CATALOG_ID, "db", "t2")); catalog.registerTable(t2, metadataLocation); Assertions.assertTrue(catalog.tableExists(t2)); svc.cleanupFiles(io, metadataLocation); } finally { svc.close(); } // t2 is still listed in the catalog, but its data and metadata.json are gone. Assertions.assertTrue(catalog.listTables(Namespace.of("db")).contains(t2)); Assertions.assertFalse(((InMemoryFileIO) io).fileExists(data)); Assertions.assertThrows(RuntimeException.class, () -> catalog.loadTable(t2)); } // F2: async purge deletes data files even when gc.enabled=false. Iceberg's own purge // (CatalogUtil.dropTableData, used by sync purge) skips them, because such tables (e.g. made by // Spark's snapshot procedure) point at another table's data files. @TestTemplate void testF2AsyncPurgeIgnoresGcDisabled() { InMemoryCatalog catalog = newCatalog("f2"); Map<String, String> noGc = ImmutableMap.of(TableProperties.GC_ENABLED, "false"); String asyncData = "memory://f2/src/data/shared-a.parquet"; BaseTable asyncTable = createWithFile(catalog, TableIdentifier.of("db", "a"), asyncData, noGc); IcebergCleanupManager svc = new IcebergCleanupManager(store, new IcebergConfig(new HashMap<>())); try { svc.cleanupFiles(asyncTable.io(), asyncTable.operations().current().metadataFileLocation()); } finally { svc.close(); } String syncData = "memory://f2/src/data/shared-s.parquet"; BaseTable syncTable = createWithFile(catalog, TableIdentifier.of("db", "s"), syncData, noGc); CatalogUtil.dropTableData(syncTable.io(), syncTable.operations().current()); InMemoryFileIO io = (InMemoryFileIO) asyncTable.io(); Assertions.assertFalse(io.fileExists(asyncData), "async purge deleted the shared data file"); Assertions.assertTrue(io.fileExists(syncData), "sync purge kept the shared data file"); } // F3: a table with delete files (format v2 merge-on-read) can never be async-purged. // ManifestFiles.readPaths throws on a DELETE manifest, after earlier data files are already // deleted. Sync purge handles the same table fine. @TestTemplate void testF3DeleteManifestMakesPurgeFailEveryTime() { InMemoryCatalog catalog = newCatalog("f3"); String data = "memory://f3/db/t/data/d1.parquet"; String deletes = "memory://f3/db/t/data/pos-deletes.parquet"; BaseTable table = createMorTable(catalog, TableIdentifier.of("db", "t"), data, deletes); InMemoryFileIO io = (InMemoryFileIO) table.io(); String metadataLocation = table.operations().current().metadataFileLocation(); IcebergCleanupManager svc = new IcebergCleanupManager(store, new IcebergConfig(new HashMap<>())); try { RuntimeException first = Assertions.assertThrows( RuntimeException.class, () -> svc.cleanupFiles(io, metadataLocation)); LOG.info("F3 first attempt failed with: {}", first.toString(), first); Assertions.assertTrue( first.getCause() instanceof IllegalArgumentException && first.getCause().getMessage().contains("delete manifest"), String.valueOf(first.getCause())); // Partial: the data file is gone, the delete file and metadata.json remain. Assertions.assertFalse(io.fileExists(data)); Assertions.assertTrue(io.fileExists(deletes)); Assertions.assertTrue(io.fileExists(metadataLocation)); // A retry fails the same way, so the job can only end FAILED. Assertions.assertThrows(RuntimeException.class, () -> svc.cleanupFiles(io, metadataLocation)); } finally { svc.close(); } String data2 = "memory://f3/db/s/data/d1.parquet"; String deletes2 = "memory://f3/db/s/data/pos-deletes.parquet"; BaseTable syncTable = createMorTable(catalog, TableIdentifier.of("db", "s"), data2, deletes2); CatalogUtil.dropTableData(syncTable.io(), syncTable.operations().current()); Assertions.assertFalse(io.fileExists(data2)); Assertions.assertFalse(io.fileExists(deletes2)); } // F5: the drop path reads the metadata location, then drops. A commit that lands in between is // not in the captured metadata, so its files and metadata.json are never deleted, yet the job // ends SUCCEEDED. @TestTemplate void testF5CommitBetweenReadAndDropLeaks() { InMemoryCatalog catalog = newCatalog("f5"); TableIdentifier t = TableIdentifier.of("db", "t"); String d1 = "memory://f5/db/t/data/d1.parquet"; BaseTable table = createWithFile(catalog, t, d1, ImmutableMap.of()); // IcebergTableOperationExecutor.dropTable step 1: wrapper.loadTableMetadata String captured = table.operations().current().metadataFileLocation(); // Another writer commits before step 2. String d2 = "memory://f5/db/t/data/d2.parquet"; appendFile(catalog, t, d2); String latest = ((BaseTable) catalog.loadTable(t)).operations().current().metadataFileLocation(); // Step 2: wrapper.dropTable. Step 3: addJob(captured). catalog.dropTable(t, false); IcebergCleanupManager svc = new IcebergCleanupManager(store, new IcebergConfig(new HashMap<>())); try { svc.cleanupFiles(table.io(), captured); } finally { svc.close(); } InMemoryFileIO io = (InMemoryFileIO) table.io(); Assertions.assertFalse(io.fileExists(d1)); Assertions.assertTrue(io.fileExists(d2), "file from the late commit leaked"); Assertions.assertTrue(io.fileExists(latest), "latest metadata.json leaked"); } // F6: retries have no delay. One short storage outage uses all attempts and the job is FAILED. @TestTemplate void testF6ShortOutageExhaustsAllAttempts() { Map<String, String> config = new HashMap<>(); config.put("async-cleanup.worker-threads", "1"); config.put("async-cleanup.poll-interval-secs", "5"); // default config.put("async-cleanup.max-attempts", "5"); // default long outageMs = 2_000L; List<Long> attemptTimes = new CopyOnWriteArrayList<>(); IcebergCleanupManager svc = new IcebergCleanupManager(store, new IcebergConfig(config)) { @Override void cleanupFiles(FileIO io, String metadataLocation) { long now = System.currentTimeMillis(); attemptTimes.add(now); if (now - attemptTimes.get(0) < outageMs) { throw new RuntimeException("503 Slow Down"); } } }; long id = store.addJob(job("t", "memory://f6/db/t/metadata/0.json")); svc.start(); try { Awaitility.await() .atMost(30, TimeUnit.SECONDS) .until( () -> store.stateOf(id) != IcebergCleanupJob.State.PENDING && store.stateOf(id) != IcebergCleanupJob.State.RUNNING); } finally { svc.close(); } long spanMs = attemptTimes.get(attemptTimes.size() - 1) - attemptTimes.get(0); LOG.info( "F6: {} attempts in {} ms, final state {}", attemptTimes.size(), spanMs, store.stateOf(id)); Assertions.assertEquals(IcebergCleanupJob.State.FAILED, store.stateOf(id)); Assertions.assertEquals(5, attemptTimes.size()); Assertions.assertTrue( spanMs < outageMs, "all 5 attempts fell inside a " + outageMs + " ms outage"); } // F7: a graceful shutdown interrupts a running job, and runJob records that as a failed // attempt. max-attempts restarts during one long purge turn the job FAILED. @TestTemplate void testF7RestartsCountAsFailedAttempts() { Map<String, String> config = new HashMap<>(); config.put("async-cleanup.worker-threads", "1"); config.put("async-cleanup.poll-interval-secs", "1"); config.put("async-cleanup.max-attempts", "2"); long id = store.addJob(job("t", "memory://f7/db/t/metadata/0.json")); for (int restart = 1; restart <= 2; restart++) { CountDownLatch deleting = new CountDownLatch(1); IcebergCleanupManager svc = new IcebergCleanupManager(store, new IcebergConfig(config)) { @Override void cleanupFiles(FileIO io, String metadataLocation) { // A long purge: the real bulk-delete path, blocked on slow storage. deleteAll(new SlowFileIO(deleting), Collections.singletonList("big-file")); } }; svc.start(); try { Assertions.assertTrue(await(deleting)); } finally { svc.close(); // rolling restart / deploy } LOG.info("F7: state after restart {} = {}", restart, store.stateOf(id)); } Assertions.assertEquals(IcebergCleanupJob.State.FAILED, store.stateOf(id)); } // F8: every catalog property, including secrets, is copied as plain JSON into // iceberg_cleanup_job.file_io_props, and kept until the row is pruned (720 h after it ends). @TestTemplate void testF8CatalogSecretsStoredInCleanupRow() { Map<String, String> catalogConfig = new HashMap<>(); catalogConfig.put("s3-secret-access-key", "SECRET-S3-KEY"); catalogConfig.put("jdbc-password", "SECRET-DB-PASSWORD"); Map<String, String> fileIOProps = new IcebergConfig(catalogConfig) .getIcebergCatalogProperties(); // wrapper.fileIOProperties() IcebergCleanupJob job = new IcebergCleanupJob(0L, CATALOG_ID, "db", "t", "s3://b/m.json", "io", fileIOProps, "u"); IcebergCleanupJobPO po = IcebergCleanupJobPO.fromCleanupJob(job, 1L, 0L); Assertions.assertTrue(po.getFileIOProps().contains("SECRET-S3-KEY"), po.getFileIOProps()); Assertions.assertTrue(po.getFileIOProps().contains("SECRET-DB-PASSWORD"), po.getFileIOProps()); } private static boolean await(CountDownLatch latch) { try { return latch.await(10, TimeUnit.SECONDS); } catch (InterruptedException e) { Thread.currentThread().interrupt(); return false; } } private static IcebergCleanupJob job(String table, String metadataLocation) { return new IcebergCleanupJob( 0L, CATALOG_ID, "db", table, metadataLocation, TestIcebergCleanupManager.NoopFileIO.class.getName(), ImmutableMap.of(), "alice"); } private static InMemoryCatalog newCatalog(String warehouse) { InMemoryCatalog catalog = new InMemoryCatalog(); catalog.initialize(warehouse, ImmutableMap.of("warehouse", "memory://" + warehouse + "/")); catalog.createNamespace(Namespace.of("db")); return catalog; } private static BaseTable createWithFile( InMemoryCatalog catalog, TableIdentifier id, String dataPath, Map<String, String> props) { catalog.buildTable(id, SCHEMA).withProperties(props).create(); appendFile(catalog, id, dataPath); return (BaseTable) catalog.loadTable(id); } private static void appendFile(InMemoryCatalog catalog, TableIdentifier id, String dataPath) { Table table = catalog.loadTable(id); DataFile file = DataFiles.builder(PartitionSpec.unpartitioned()) .withPath(dataPath) .withFileSizeInBytes(10L) .withRecordCount(1L) .build(); table.newAppend().appendFile(file).commit(); ((InMemoryFileIO) table.io()).addFile(dataPath, new byte[] {1}); } private static BaseTable createMorTable( InMemoryCatalog catalog, TableIdentifier id, String dataPath, String deletePath) { catalog.buildTable(id, SCHEMA).withProperty(TableProperties.FORMAT_VERSION, "2").create(); appendFile(catalog, id, dataPath); Table table = catalog.loadTable(id); DeleteFile deletes = FileMetadata.deleteFileBuilder(PartitionSpec.unpartitioned()) .ofPositionDeletes() .withPath(deletePath) .withFormat(FileFormat.PARQUET) .withFileSizeInBytes(10L) .withRecordCount(1L) .build(); table.newRowDelta().addDeletes(deletes).commit(); ((InMemoryFileIO) table.io()).addFile(deletePath, new byte[] {1}); BaseTable base = (BaseTable) catalog.loadTable(id); TableMetadata current = base.operations().current(); Assertions.assertEquals(2, current.snapshots().size()); return base; } /** Bulk delete that blocks like a slow object store until the thread is interrupted. */ static class SlowFileIO implements SupportsBulkOperations { private final CountDownLatch started; SlowFileIO(CountDownLatch started) { this.started = started; } @Override public InputFile newInputFile(String path) { throw new UnsupportedOperationException(); } @Override public OutputFile newOutputFile(String path) { throw new UnsupportedOperationException(); } @Override public void deleteFile(String path) { deleteFiles(Collections.singletonList(path)); } @Override public void deleteFiles(Iterable<String> paths) { started.countDown(); try { Thread.sleep(60_000L); } catch (InterruptedException e) { Thread.currentThread().interrupt(); throw new RuntimeException("delete interrupted", e); } } } } ``` </details> <details><summary>TestAsyncPurgePredictions.java</summary> ```java /* * Licensed to the Apache Software Foundation (ASF) under one * or more contributor license agreements. See the NOTICE file * distributed with this work for additional information * regarding copyright ownership. The ASF licenses this file * to you under the Apache License, Version 2.0 (the * "License"); you may not use this file except in compliance * with the License. You may obtain a copy of the License at * * http://www.apache.org/licenses/LICENSE-2.0 * * Unless required by applicable law or agreed to in writing, * software distributed under the License is distributed on an * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY * KIND, either express or implied. See the License for the * specific language governing permissions and limitations * under the License. */ package org.apache.gravitino.iceberg.service.cleanup; import com.google.common.collect.ImmutableMap; import java.util.HashMap; import java.util.List; import java.util.Map; import java.util.concurrent.CopyOnWriteArrayList; import java.util.concurrent.TimeUnit; import org.apache.commons.lang3.reflect.FieldUtils; import org.apache.gravitino.GravitinoEnv; import org.apache.gravitino.iceberg.common.IcebergConfig; import org.apache.gravitino.storage.RandomIdGenerator; import org.apache.gravitino.storage.relational.TestJDBCBackend; import org.apache.iceberg.BaseTable; import org.apache.iceberg.DataFile; import org.apache.iceberg.DataFiles; import org.apache.iceberg.DeleteFile; import org.apache.iceberg.FileFormat; import org.apache.iceberg.FileMetadata; import org.apache.iceberg.PartitionSpec; import org.apache.iceberg.Schema; import org.apache.iceberg.Table; import org.apache.iceberg.TableProperties; import org.apache.iceberg.catalog.Namespace; import org.apache.iceberg.catalog.TableIdentifier; import org.apache.iceberg.exceptions.ForbiddenException; import org.apache.iceberg.inmemory.InMemoryCatalog; import org.apache.iceberg.inmemory.InMemoryFileIO; import org.apache.iceberg.io.FileIO; import org.apache.iceberg.io.InputFile; import org.apache.iceberg.io.OutputFile; import org.apache.iceberg.types.Types; import org.awaitility.Awaitility; import org.junit.jupiter.api.AfterAll; import org.junit.jupiter.api.Assertions; import org.junit.jupiter.api.BeforeAll; import org.junit.jupiter.api.BeforeEach; import org.junit.jupiter.api.TestTemplate; import org.slf4j.Logger; import org.slf4j.LoggerFactory; /** * Tests of bugs predicted by the design weak points in the async purge audit epic. A passing test means * the prediction held (or, for N1a, that the predicted hole does not exist). */ class TestAsyncPurgePredictions extends TestJDBCBackend { private static final Logger LOG = LoggerFactory.getLogger(TestAsyncPurgePredictions.class); private static final Schema SCHEMA = new Schema(Types.NestedField.required(1, "id", Types.IntegerType.get())); private IcebergCleanupJobStore store; private Object originalConfig; private Object originalIdGenerator; @BeforeAll public void snapshotGravitinoEnv() throws IllegalAccessException { originalConfig = FieldUtils.readField(GravitinoEnv.getInstance(), "config", true); originalIdGenerator = FieldUtils.readField(GravitinoEnv.getInstance(), "idGenerator", true); } @AfterAll public void restoreGravitinoEnv() throws IllegalAccessException { FieldUtils.writeField(GravitinoEnv.getInstance(), "config", originalConfig, true); FieldUtils.writeField(GravitinoEnv.getInstance(), "idGenerator", originalIdGenerator, true); } @BeforeEach public void prepareStore() { store = new IcebergCleanupJobStore(new RandomIdGenerator()); } // N1a: a new table created at the dropped table's location while the purge is pending. The // purge deletes by reachability, so the new table's files must survive. Expected: no damage. @TestTemplate void testN1aCreateAtDroppedLocationIsSafeForPurge() { InMemoryCatalog catalog = newCatalog("n1a"); TableIdentifier t = TableIdentifier.of("db", "t"); BaseTable old = createWithFile(catalog, t, "memory://n1a/db/t/data/old.parquet", Map.of()); String oldMetadata = old.operations().current().metadataFileLocation(); String location = old.location(); catalog.dropTable(t, false); TableIdentifier t2 = TableIdentifier.of("db", "t2"); catalog.buildTable(t2, SCHEMA).withLocation(location).create(); String newData = location + "/data/new.parquet"; appendFile(catalog, t2, newData); IcebergCleanupManager svc = new IcebergCleanupManager(store, new IcebergConfig(new HashMap<>())); try { svc.cleanupFiles(old.io(), oldMetadata); } finally { svc.close(); } Assertions.assertTrue(((InMemoryFileIO) old.io()).fileExists(newData)); Assertions.assertDoesNotThrow(() -> catalog.loadTable(t2).currentSnapshot()); } // N2: v3 deletion vectors live in delete manifests, so async purge must fail on them too. @TestTemplate void testN2DeletionVectorsAlsoBreakPurge() { InMemoryCatalog catalog = newCatalog("n2"); TableIdentifier t = TableIdentifier.of("db", "t"); catalog.buildTable(t, SCHEMA).withProperty(TableProperties.FORMAT_VERSION, "3").create(); String data = "memory://n2/db/t/data/d1.parquet"; appendFile(catalog, t, data); Table table = catalog.loadTable(t); String dv = "memory://n2/db/t/data/dv.puffin"; DeleteFile vector = FileMetadata.deleteFileBuilder(PartitionSpec.unpartitioned()) .ofPositionDeletes() .withPath(dv) .withFormat(FileFormat.PUFFIN) .withFileSizeInBytes(100L) .withRecordCount(1L) .withReferencedDataFile(data) .withContentOffset(4L) .withContentSizeInBytes(40L) .build(); table.newRowDelta().addDeletes(vector).commit(); ((InMemoryFileIO) table.io()).addFile(dv, new byte[] {1}); BaseTable base = (BaseTable) catalog.loadTable(t); IcebergCleanupManager svc = new IcebergCleanupManager(store, new IcebergConfig(new HashMap<>())); try { RuntimeException e = Assertions.assertThrows( RuntimeException.class, () -> svc.cleanupFiles(base.io(), base.operations().current().metadataFileLocation())); LOG.info("N2: v3 purge failed with {}", String.valueOf(e.getCause())); } finally { svc.close(); } Assertions.assertTrue(((InMemoryFileIO) base.io()).fileExists(dv)); } // N4: a permanent permission error (403) is retried with no delay, exactly like a transient one. @TestTemplate void testN4PermissionDeniedBurnsAllAttempts() { Map<String, String> config = new HashMap<>(); config.put("async-cleanup.worker-threads", "1"); List<Long> attempts = new CopyOnWriteArrayList<>(); ForbiddenFileIO.ATTEMPTS = attempts; IcebergCleanupManager svc = new IcebergCleanupManager(store, new IcebergConfig(config)); long id = store.addJob( new IcebergCleanupJob( 0L, 7L, "db", "t", "s3://bucket/db/t/metadata/00001.metadata.json", ForbiddenFileIO.class.getName(), ImmutableMap.of(), "alice")); svc.start(); try { Awaitility.await() .atMost(30, TimeUnit.SECONDS) .until(() -> store.stateOf(id) == IcebergCleanupJob.State.FAILED); } finally { svc.close(); } long span = attempts.get(attempts.size() - 1) - attempts.get(0); LOG.info("N4: {} attempts on a 403 in {} ms", attempts.size(), span); Assertions.assertEquals(5, attempts.size()); } /** FileIO whose every read is rejected the way S3 rejects a missing permission. */ public static class ForbiddenFileIO implements FileIO { static volatile List<Long> ATTEMPTS = new CopyOnWriteArrayList<>(); @Override public InputFile newInputFile(String path) { ATTEMPTS.add(System.currentTimeMillis()); throw new ForbiddenException("403 Access Denied: %s", path); } @Override public OutputFile newOutputFile(String path) { throw new UnsupportedOperationException(); } @Override public void deleteFile(String path) { throw new ForbiddenException("403 Access Denied: %s", path); } } private static InMemoryCatalog newCatalog(String warehouse) { InMemoryCatalog catalog = new InMemoryCatalog(); catalog.initialize(warehouse, ImmutableMap.of("warehouse", "memory://" + warehouse + "/")); catalog.createNamespace(Namespace.of("db")); return catalog; } private static BaseTable createWithFile( InMemoryCatalog catalog, TableIdentifier id, String dataPath, Map<String, String> props) { catalog.buildTable(id, SCHEMA).withProperties(props).create(); appendFile(catalog, id, dataPath); return (BaseTable) catalog.loadTable(id); } private static void appendFile(InMemoryCatalog catalog, TableIdentifier id, String dataPath) { Table table = catalog.loadTable(id); DataFile file = DataFiles.builder(PartitionSpec.unpartitioned()) .withPath(dataPath) .withFileSizeInBytes(10L) .withRecordCount(1L) .build(); table.newAppend().appendFile(file).commit(); ((InMemoryFileIO) table.io()).addFile(dataPath, new byte[] {1}); } } ``` </details> <details><summary>TestAsyncPurgeDropNotAtomic.java</summary> ```java /* * Licensed to the Apache Software Foundation (ASF) under one * or more contributor license agreements. See the NOTICE file * distributed with this work for additional information * regarding copyright ownership. The ASF licenses this file * to you under the Apache License, Version 2.0 (the * "License"); you may not use this file except in compliance * with the License. You may obtain a copy of the License at * * http://www.apache.org/licenses/LICENSE-2.0 * * Unless required by applicable law or agreed to in writing, * software distributed under the License is distributed on an * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY * KIND, either express or implied. See the License for the * specific language governing permissions and limitations * under the License. */ package org.apache.gravitino.iceberg.service.dispatcher; import static org.mockito.ArgumentMatchers.any; import static org.mockito.Mockito.mock; import static org.mockito.Mockito.mockStatic; import static org.mockito.Mockito.verify; import static org.mockito.Mockito.when; import java.util.Collections; import java.util.Optional; import org.apache.gravitino.GravitinoEnv; import org.apache.gravitino.catalog.CatalogManager; import org.apache.gravitino.catalog.CatalogTestUtils; import org.apache.gravitino.connector.BaseCatalog; import org.apache.gravitino.iceberg.service.CatalogWrapperForREST; import org.apache.gravitino.iceberg.service.IcebergCatalogWrapperManager; import org.apache.gravitino.iceberg.service.authorization.IcebergRESTServerContext; import org.apache.gravitino.iceberg.service.cleanup.IcebergCleanupManager; import org.apache.gravitino.iceberg.service.provider.IcebergConfigProvider; import org.apache.gravitino.listener.api.event.IcebergRequestContext; import org.apache.gravitino.meta.CatalogEntity; import org.apache.iceberg.TableMetadata; import org.apache.iceberg.catalog.TableIdentifier; import org.junit.jupiter.api.Assertions; import org.junit.jupiter.api.BeforeEach; import org.junit.jupiter.api.Test; import org.mockito.MockedStatic; /** * F4 repro: async drop removes the catalog entry before it writes the cleanup row. If the row write * fails, the table is gone, no row tracks its files, and the name is free at once. */ class TestAsyncPurgeDropNotAtomic { private static final TableIdentifier TABLE = TableIdentifier.of("db", "t"); @BeforeEach void setUp() { IcebergConfigProvider provider = mock(IcebergConfigProvider.class); when(provider.getMetalakeName()).thenReturn("metalake"); when(provider.getDefaultCatalogName()).thenReturn("cat"); IcebergRESTServerContext.create(provider, false, false, true, null); } @Test void testF4RowWriteFailureAfterCatalogDropLeavesNoRecord() { CatalogWrapperForREST wrapper = mock(CatalogWrapperForREST.class); IcebergCleanupManager cleanup = mock(IcebergCleanupManager.class); TableMetadata metadata = mock(TableMetadata.class); when(metadata.metadataFileLocation()).thenReturn("s3://b/db/t/metadata/0.json"); when(wrapper.loadTableMetadata(any())).thenReturn(metadata); when(wrapper.fileIOImpl()).thenReturn("io"); when(wrapper.fileIOProperties()).thenReturn(Collections.emptyMap()); when(cleanup.addJob(any())).thenThrow(new RuntimeException("entity store unavailable")); IcebergCatalogWrapperManager manager = mock(IcebergCatalogWrapperManager.class); when(manager.getCatalogWrapper("cat")).thenReturn(wrapper); IcebergRequestContext context = mock(IcebergRequestContext.class); when(context.catalogName()).thenReturn("cat"); when(context.userName()).thenReturn("alice"); when(context.asyncPurge()).thenReturn(true); try (MockedStatic<GravitinoEnv> ignored = mockCatalogId()) { IcebergTableOperationExecutor executor = new IcebergTableOperationExecutor(manager, Optional.of(cleanup)); // The client gets an error for a drop that already happened. Assertions.assertThrows( RuntimeException.class, () -> executor.dropTable(context, TABLE, true)); } verify(wrapper).dropTable(TABLE); } private static MockedStatic<GravitinoEnv> mockCatalogId() { MockedStatic<GravitinoEnv> envStatic = mockStatic(GravitinoEnv.class); GravitinoEnv env = mock(GravitinoEnv.class); CatalogManager catalogManager = mock(CatalogManager.class); BaseCatalog<?> catalog = mock(BaseCatalog.class); CatalogEntity entity = mock(CatalogEntity.class); envStatic.when(GravitinoEnv::getInstance).thenReturn(env); when(env.catalogManager()).thenReturn(catalogManager); CatalogTestUtils.mockDoWithCatalog(catalogManager, catalog); when(catalog.entity()).thenReturn(entity); when(entity.id()).thenReturn(42L); return envStatic; } } ``` </details> ### Task list - [ ] #14389 - [ ] #14390 - [ ] #14391 - [ ] #14392 - [ ] #14393 - [ ] #14333 - [ ] #14388 -- 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]
