nevzheng opened a new issue, #14388:
URL: https://github.com/apache/gravitino/issues/14388
### What would you like to be improved?
**Summary.** A cleanup job that ends `FAILED` releases the table name while
its files may still exist. A recreate at that name lands on the old prefix, the
old data stays readable there, and the documented manual cleanup of the failed
job would then delete the new table's files. **Proposal:** keep `FAILED`
reserving the name, report it on create, and give operators a reset; longer
term, unique table locations.
Async table purge (`DELETE
/iceberg/v1/{prefix}/namespaces/{namespace}/tables/{table}?purgeRequested=true`
with `X-Gravitino-Async-Purge: true`, 1.3.x and `main`) reserves the dropped
identifier while the cleanup job runs: `createTable` and `registerTable` at
that name return `409 Conflict`. The
[docs](https://github.com/apache/gravitino/blob/bb634de/docs/iceberg-rest-service.md#L165)
describe the reservation as held "until the file cleanup finishes".
The reservation is only held while the job is `PENDING` or `RUNNING`. A job
that exhausts `max-attempts` becomes `FAILED`, and `FAILED` releases the name
exactly like `SUCCEEDED`, although `FAILED` by definition means some files may
remain:
-
[`selectUnfinishedJobId`](https://github.com/apache/gravitino/blob/bb634de/iceberg/iceberg-rest-server/src/main/java/org/apache/gravitino/iceberg/service/cleanup/mapper/provider/base/IcebergCleanupJobBaseSQLProvider.java#L163-L174)
filters `state IN ('PENDING', 'RUNNING')`. It backs
[`isNameOccupied`](https://github.com/apache/gravitino/blob/bb634de/iceberg/iceberg-rest-server/src/main/java/org/apache/gravitino/iceberg/service/cleanup/IcebergCleanupManager.java#L135)
and
[`rejectIfBeingPurged`](https://github.com/apache/gravitino/blob/bb634de/iceberg/iceberg-rest-server/src/main/java/org/apache/gravitino/iceberg/service/dispatcher/IcebergCleanupHelper.java#L50-L74).
-
[`recordFailure`](https://github.com/apache/gravitino/blob/bb634de/iceberg/iceberg-rest-server/src/main/java/org/apache/gravitino/iceberg/service/cleanup/mapper/provider/base/IcebergCleanupJobBaseSQLProvider.java#L137)
sets `FAILED` at the attempt ceiling, from
[`runJob`](https://github.com/apache/gravitino/blob/bb634de/iceberg/iceberg-rest-server/src/main/java/org/apache/gravitino/iceberg/service/cleanup/IcebergCleanupManager.java#L280-L291).
- `FAILED` is terminal:
[`selectCandidateJobs`](https://github.com/apache/gravitino/blob/bb634de/iceberg/iceberg-rest-server/src/main/java/org/apache/gravitino/iceberg/service/cleanup/mapper/provider/base/IcebergCleanupJobBaseSQLProvider.java#L69-L70)
never re-selects it, there is no reset or re-enqueue API, and the row is
pruned after `retention-hours` ([default 720
h](https://github.com/apache/gravitino/blob/bb634de/iceberg/iceberg-rest-server/src/main/java/org/apache/gravitino/iceberg/service/cleanup/mapper/provider/base/IcebergCleanupJobBaseSQLProvider.java#L180-L184)).
- The row is keyed by `(catalog_id, namespace, table_name)`
([schema](https://github.com/apache/gravitino/blob/bb634de/scripts/mysql/schema-1.3.0-mysql.sql#L603-L620))
and stores neither the table UUID nor the table root location.
```mermaid
stateDiagram-v2
direction LR
[*] --> PENDING : addJob (async drop)
PENDING --> RUNNING : takePendingJob
RUNNING --> PENDING : recordFailure, attempts + 1 < max-attempts
RUNNING --> SUCCEEDED : markSucceeded (all files gone)
RUNNING --> FAILED : recordFailure at max-attempts (files may remain)
SUCCEEDED --> [*] : pruned after retention-hours
FAILED --> [*] : pruned after retention-hours, never re-selected
note right of PENDING : name RESERVED (409)
note right of RUNNING : name RESERVED (409)
note right of SUCCEEDED : name FREE
note right of FAILED : name FREE
```
**Why this matters.** The reservation exists because a recreate without an
explicit `location` lands on `<warehouse>/<namespace>/<table>`, the prefix the
old files still occupy ([design
note](https://github.com/apache/gravitino/blob/bb634de/iceberg/iceberg-rest-server/src/main/java/org/apache/gravitino/iceberg/service/dispatcher/IcebergCleanupHelper.java#L50-L54),
[design doc
§5.7](https://github.com/apache/gravitino/blob/bb634de/design-docs/async-iceberg-rest-hard-deletion.md#L527-L545)).
That reason applies at least as much after a failure as during the run:
1. The old table's data and metadata stay readable under the new table's
prefix for anyone with storage access. This is the same-name, same-location
recreate the tombstone was meant to close, reached through the failure path.
2. The documented remediation for a `FAILED` row, a manual delete or a
lifecycle rule on the table's prefix
([§5.6](https://github.com/apache/gravitino/blob/bb634de/design-docs/async-iceberg-rest-hard-deletion.md#L496-L506)),
becomes destructive once a new table sits on that prefix, and the row carries
nothing to tell the two tables apart.
3. The window is short and silent. With defaults (`poll-interval-secs=5`,
`max-attempts=5`) a hard failure reaches `FAILED` in about 25 to 30 s, and a
client that saw `409` now gets `200` with no indication that the purge gave up.
What does not go wrong: the worker deletes by reachability from the old
`metadata_location` and data files are UUID-named, so a re-run could not delete
the new table's files, and the new table does not read the leftovers. The
concern is leftover-data exposure and unsafe manual cleanup, not catalog
corruption.
**This is a documented trade-off, not an oversight.**
[§5.6](https://github.com/apache/gravitino/blob/bb634de/design-docs/async-iceberg-rest-hard-deletion.md#L496-L506)
and
[§5.7](https://github.com/apache/gravitino/blob/bb634de/design-docs/async-iceberg-rest-hard-deletion.md#L543-L545)
state that `FAILED` lifts the tombstone so "a failed cleanup never permanently
wedges the name", with the leak handled out of band. This issue asks whether
that default should lean the other way, or at least be visible and
operator-controllable.
### Reproduction
<details>
<summary>Sequence: async drop, 409 while reserved, FAILED after read
failures, then create succeeds on the old prefix</summary>
```mermaid
sequenceDiagram
participant C as Client
participant S as Iceberg REST
participant J as iceberg_cleanup_job
participant W as Cleanup worker
participant F as Storage (reads denied)
C->>S: DELETE .../namespaces/db/tables/t?purgeRequested=true
(X-Gravitino-Async-Purge: true)
S->>J: addJob(db.t) PENDING
S-->>C: 204
C->>S: POST .../namespaces/db/tables {name: t}
S-->>C: 409 "Table db.t is being purged; retry after cleanup completes"
loop max-attempts
W->>F: read metadata.json
F-->>W: AccessDenied
W->>J: recordFailure
end
J->>J: FAILED (no file deleted)
C->>S: POST .../namespaces/db/tables {name: t}
S-->>C: 200, new db.t at <warehouse>/db/t, old metadata.json and data
still there
```
</details>
<details>
<summary>State machine walk on H2 with the real DDL and SQL (proves the
predicate: FAILED is the moment isNameOccupied flips to free)</summary>
State transitions, executed against H2 with the real DDL
([`schema-1.3.0-h2.sql`](https://github.com/apache/gravitino/blob/bb634de/scripts/h2/schema-1.3.0-h2.sql#L616-L633))
and the exact statements from `IcebergCleanupJobBaseSQLProvider`,
`max-attempts=2`:
```text
addJob (DELETE
/iceberg/v1/{prefix}/namespaces/db/tables/t?purgeRequested=true,
X-Gravitino-Async-Purge: true) | state=PENDING attempts=0 last_error=null |
isNameOccupied(db.t): RESERVED (create/register -> 409)
attempt 1: takePendingJob -> markRunning
| state=RUNNING attempts=0 last_error=null | isNameOccupied(db.t): RESERVED
(create/register -> 409)
attempt 1: cleanupFiles threw -> recordFailure(maxAttempts=2)
| state=PENDING attempts=1 last_error=AccessDenied: GetObject metadata.json |
isNameOccupied(db.t): RESERVED (create/register -> 409)
attempt 2: takePendingJob -> markRunning
| state=RUNNING attempts=1 last_error=AccessDenied: GetObject metadata.json |
isNameOccupied(db.t): RESERVED (create/register -> 409)
attempt 2: cleanupFiles threw -> recordFailure(maxAttempts=2)
| state=FAILED attempts=2 last_error=AccessDenied: GetObject metadata.json |
isNameOccupied(db.t): FREE (create/register succeeds)
selectCandidateJobs after FAILED -> no rows (never re-selected)
```
</details>
**Context: #14333.** That issue is in a different method with a different
wrong value: `CatalogUtil.deleteFiles` swallows delete failures inside
[`deleteAll`](https://github.com/apache/gravitino/blob/bb634de/iceberg/iceberg-rest-server/src/main/java/org/apache/gravitino/iceberg/service/cleanup/IcebergCleanupManager.java#L228-L237),
so a job whose deletes are denied is marked `SUCCEEDED`. This issue is the
reservation rule itself. Today a job only reaches `FAILED` through a non-delete
failure (metadata or manifest unreadable, `FileIO` cannot be instantiated).
Once #14333 propagates delete failures, denied deletes land in `FAILED` too,
and this rule then releases the name for them as well. Fixing #14333 alone
makes this more reachable, not less.
End to end: `TestIcebergCleanupFailedJobFreesName` (below), run on `main` at
[`bb634de`](https://github.com/apache/gravitino/commit/bb634de), H2 backend, 3
tests pass. It drives the real `IcebergTableOperationExecutor`,
`IcebergCleanupManager` and relational job store against an `InMemoryCatalog`;
the only double is the worker's `FileIO`.
| Test | Storage fault | Job ends | Files | Name |
| --- | --- | --- | --- | --- |
| `testFailedJobFreesNameWhileFilesRemain` | reads denied | `FAILED` after 2
attempts | all present | released; `createTable` succeeds at the old location |
| `testDeleteDeniedJobIsMarkedSucceededNotFailed` (#14333 contrast) |
deletes denied | `SUCCEEDED` | all present | released |
| `testRenameOntoReservedNameIsNotRejected` | none | `PENDING` | n/a |
`createTable` 409, `renameTable` onto the same name succeeds |
<details>
<summary>JUnit repro (TestIcebergCleanupFailedJobFreesName.java, passes on
main at bb634de) and how to run it</summary>
```
./gradlew :iceberg:iceberg-rest-server:test --tests
'org.apache.gravitino.iceberg.service.cleanup.TestIcebergCleanupFailedJobFreesName'
-PskipITs -PskipDockerTests=true
```
```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 static org.mockito.ArgumentMatchers.any;
import static org.mockito.ArgumentMatchers.anyBoolean;
import static org.mockito.Mockito.doAnswer;
import static org.mockito.Mockito.mock;
import static org.mockito.Mockito.when;
import com.google.common.collect.ImmutableMap;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.Optional;
import java.util.concurrent.TimeUnit;
import org.apache.commons.lang3.reflect.FieldUtils;
import org.apache.gravitino.GravitinoEnv;
import org.apache.gravitino.auth.AuthConstants;
import org.apache.gravitino.catalog.CatalogManager;
import org.apache.gravitino.catalog.CatalogTestUtils;
import org.apache.gravitino.connector.BaseCatalog;
import org.apache.gravitino.iceberg.common.IcebergConfig;
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.dispatcher.IcebergTableOperationExecutor;
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.gravitino.storage.RandomIdGenerator;
import org.apache.gravitino.storage.relational.TestJDBCBackend;
import org.apache.iceberg.BaseTable;
import org.apache.iceberg.CatalogProperties;
import org.apache.iceberg.DataFile;
import org.apache.iceberg.DataFiles;
import org.apache.iceberg.ManifestFile;
import org.apache.iceberg.PartitionSpec;
import org.apache.iceberg.Schema;
import org.apache.iceberg.Snapshot;
import org.apache.iceberg.Table;
import org.apache.iceberg.TableMetadata;
import org.apache.iceberg.TableMetadataParser;
import org.apache.iceberg.catalog.Namespace;
import org.apache.iceberg.catalog.TableIdentifier;
import org.apache.iceberg.exceptions.AlreadyExistsException;
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.rest.requests.CreateTableRequest;
import org.apache.iceberg.rest.requests.RenameTableRequest;
import org.apache.iceberg.rest.responses.LoadTableResponse;
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;
/**
* Reproduces gaps in the async-purge name reservation, end to end through
{@link
* IcebergTableOperationExecutor} with a real {@link IcebergCleanupManager},
the relational job
* store ({@link TestJDBCBackend}: H2 by default) and a real Iceberg table
in an {@link
* InMemoryCatalog}. Only the {@link FileIO} the worker instantiates is a
test double:
*
* <ol>
* <li>a cleanup job that reaches {@code FAILED} after {@code
max-attempts} frees the identifier
* although none of the old table's files were deleted, so a create at
the same name succeeds
* and lands on the old table's location (reached here through a
storage <em>read</em>
* failure, the only kind of failure that currently makes a job {@code
FAILED});
* <li>for contrast, a storage <em>delete</em> failure never reaches
{@code FAILED}: {@code
* CatalogUtil.deleteFiles} swallows it and the job is marked {@code
SUCCEEDED} (that is
* apache/gravitino#14333, a different defect in a different method);
* <li>{@code renameTable} onto an identifier held by a {@code PENDING}
job is not rejected,
* unlike {@code createTable}.
* </ol>
*/
class TestIcebergCleanupFailedJobFreesName extends TestJDBCBackend {
private static final long CATALOG_ID = 100L;
private static final String CATALOG_NAME = "cat";
private static final Namespace DB = Namespace.of("db");
private static final TableIdentifier TABLE = TableIdentifier.of(DB, "t");
private static final TableIdentifier OTHER = TableIdentifier.of(DB,
"other");
private static final Schema SCHEMA =
new Schema(Types.NestedField.required(1, "id",
Types.IntegerType.get()));
private static final String DATA_FILE =
"memory://warehouse/db/t/data/00000-0.parquet";
private IcebergCleanupJobStore store;
private InMemoryCatalog catalog;
// See TestIcebergCleanupManager: TestJDBCBackend swaps GravitinoEnv's
config/idGenerator and
// never restores them; we also install a mock CatalogManager so
IcebergCleanupHelper.catalogId
// resolves. Snapshot everything and put it back after the class.
private Object originalConfig;
private Object originalIdGenerator;
private Object originalCatalogManager;
@BeforeAll
public void snapshotGravitinoEnv() throws IllegalAccessException {
GravitinoEnv env = GravitinoEnv.getInstance();
originalConfig = FieldUtils.readField(env, "config", true);
originalIdGenerator = FieldUtils.readField(env, "idGenerator", true);
originalCatalogManager = FieldUtils.readField(env, "catalogManager",
true);
IcebergConfigProvider provider = mock(IcebergConfigProvider.class);
when(provider.getMetalakeName()).thenReturn("metalake");
when(provider.getDefaultCatalogName()).thenReturn(CATALOG_NAME);
IcebergRESTServerContext.create(provider, false, false, true, null);
CatalogManager catalogManager = mock(CatalogManager.class);
BaseCatalog<?> baseCatalog = mock(BaseCatalog.class);
CatalogEntity entity = mock(CatalogEntity.class);
CatalogTestUtils.mockDoWithCatalog(catalogManager, baseCatalog);
when(baseCatalog.entity()).thenReturn(entity);
when(entity.id()).thenReturn(CATALOG_ID);
FieldUtils.writeField(env, "catalogManager", catalogManager, true);
}
@AfterAll
public void restoreGravitinoEnv() throws IllegalAccessException {
GravitinoEnv env = GravitinoEnv.getInstance();
FieldUtils.writeField(env, "config", originalConfig, true);
FieldUtils.writeField(env, "idGenerator", originalIdGenerator, true);
FieldUtils.writeField(env, "catalogManager", originalCatalogManager,
true);
}
@BeforeEach
public void prepare() {
store = new IcebergCleanupJobStore(new RandomIdGenerator());
catalog = new InMemoryCatalog();
catalog.initialize(
"test", ImmutableMap.of(CatalogProperties.WAREHOUSE_LOCATION,
"memory://warehouse"));
catalog.createNamespace(DB);
DelegatingFileIO.delegate = null;
}
/**
* Claim 1: FAILED frees the name while every file of the old table is
still there. The worker
* cannot read the old metadata.json (think: storage credentials revoked
or expired), so every
* attempt throws before any delete, and after max-attempts the job is
FAILED.
*/
@TestTemplate
void testFailedJobFreesNameWhileFilesRemain() {
BaseTable old = createTableWithDataFile();
InMemoryFileIO io = (InMemoryFileIO) old.io();
String oldLocation = old.location();
String oldMetadata = old.operations().current().metadataFileLocation();
List<String> oldFiles = reachableFiles(old);
oldFiles.forEach(file -> Assertions.assertTrue(io.fileExists(file),
file));
DelegatingFileIO.delegate = io;
// max-attempts=2, poll every second: a hard failure reaches FAILED in
~2 polls.
IcebergCleanupManager manager = new IcebergCleanupManager(store,
fastFailConfig());
IcebergTableOperationExecutor executor = executor(manager,
ReadDeniedFileIO.class);
try {
// DELETE
/iceberg/v1/{prefix}/namespaces/db/tables/t?purgeRequested=true
// with X-Gravitino-Async-Purge: true
executor.dropTable(context(true), TABLE, true);
Assertions.assertFalse(catalog.tableExists(TABLE));
long jobId = store.findUnfinishedJobId(CATALOG_ID, "db", "t").get();
Assertions.assertEquals(IcebergCleanupJob.State.PENDING,
store.stateOf(jobId));
Assertions.assertTrue(manager.isNameOccupied(CATALOG_ID, "db", "t"));
// POST /iceberg/v1/{prefix}/namespaces/db/tables {name: t} -> 409
while PENDING.
AlreadyExistsException rejected =
Assertions.assertThrows(
AlreadyExistsException.class,
() -> executor.createTable(context(false), DB, createReq()));
Assertions.assertEquals(
"Table db.t is being purged; retry after cleanup completes",
rejected.getMessage());
// Let the worker burn through max-attempts against storage that
denies every read.
manager.start();
Awaitility.await()
.atMost(20, TimeUnit.SECONDS)
.until(() -> store.stateOf(jobId) ==
IcebergCleanupJob.State.FAILED);
// Nothing was deleted, yet the reservation is gone ...
oldFiles.forEach(file -> Assertions.assertTrue(io.fileExists(file),
file));
Assertions.assertFalse(manager.isNameOccupied(CATALOG_ID, "db", "t"));
Assertions.assertTrue(store.findUnfinishedJobId(CATALOG_ID, "db",
"t").isEmpty());
// ... so the same POST now succeeds, and the new db.t sits on the old
table's prefix.
LoadTableResponse created = executor.createTable(context(false), DB,
createReq());
TableMetadata fresh = created.tableMetadata();
Assertions.assertEquals(oldLocation, fresh.location());
Assertions.assertNotEquals(oldMetadata, fresh.metadataFileLocation());
Assertions.assertTrue(oldMetadata.startsWith(fresh.location() +
"/metadata/"));
Assertions.assertTrue(DATA_FILE.startsWith(fresh.location() +
"/data/"));
// The old metadata.json is still readable under the new table's
prefix and still points at
// the old data file; the new table's own metadata does not reference
it.
TableMetadata leftover = TableMetadataParser.read(io, oldMetadata);
Assertions.assertEquals(1, leftover.snapshots().size());
Assertions.assertEquals(0, fresh.snapshots().size());
Assertions.assertTrue(io.fileExists(DATA_FILE));
} finally {
manager.close();
DelegatingFileIO.delegate = null;
}
}
/**
* Contrast (apache/gravitino#14333): when storage denies deletes instead
of reads, the job does
* NOT reach FAILED. CatalogUtil.deleteFiles logs and swallows every
delete failure, so runJob
* marks the job SUCCEEDED with all files still present. The name is freed
either way, but through
* a different state and a different method than claim 1.
*/
@TestTemplate
void testDeleteDeniedJobIsMarkedSucceededNotFailed() {
BaseTable old = createTableWithDataFile();
InMemoryFileIO io = (InMemoryFileIO) old.io();
List<String> oldFiles = reachableFiles(old);
DelegatingFileIO.delegate = io;
IcebergCleanupManager manager = new IcebergCleanupManager(store,
fastFailConfig());
IcebergTableOperationExecutor executor = executor(manager,
DeleteDeniedFileIO.class);
try {
executor.dropTable(context(true), TABLE, true);
long jobId = store.findUnfinishedJobId(CATALOG_ID, "db", "t").get();
Assertions.assertTrue(manager.isNameOccupied(CATALOG_ID, "db", "t"));
manager.start();
Awaitility.await()
.atMost(20, TimeUnit.SECONDS)
.until(
() ->
store.stateOf(jobId) != IcebergCleanupJob.State.PENDING
&& store.stateOf(jobId) !=
IcebergCleanupJob.State.RUNNING);
Assertions.assertEquals(IcebergCleanupJob.State.SUCCEEDED,
store.stateOf(jobId));
oldFiles.forEach(file -> Assertions.assertTrue(io.fileExists(file),
file));
Assertions.assertFalse(manager.isNameOccupied(CATALOG_ID, "db", "t"));
Assertions.assertDoesNotThrow(() ->
executor.createTable(context(false), DB, createReq()));
} finally {
manager.close();
DelegatingFileIO.delegate = null;
}
}
/** Claim 2: rename is not checked against the reservation, create is. */
@TestTemplate
void testRenameOntoReservedNameIsNotRejected() {
catalog.createTable(OTHER, SCHEMA);
IcebergCleanupManager manager =
new IcebergCleanupManager(store, new IcebergConfig(new HashMap<>()));
IcebergTableOperationExecutor executor = executor(manager,
ReadDeniedFileIO.class);
try {
// A PENDING job holds db.t (worker never started, so it stays
PENDING).
store.addJob(
new IcebergCleanupJob(
0L,
CATALOG_ID,
"db",
"t",
"memory://warehouse/db/t/metadata/0.json",
ReadDeniedFileIO.class.getName(),
ImmutableMap.of(),
"alice"));
Assertions.assertTrue(manager.isNameOccupied(CATALOG_ID, "db", "t"));
// POST /iceberg/v1/{prefix}/namespaces/db/tables {name: t} -> 409.
Assertions.assertThrows(
AlreadyExistsException.class,
() -> executor.createTable(context(false), DB, createReq()));
// POST /iceberg/v1/{prefix}/tables/rename {source: db.other,
destination: db.t} -> 204.
RenameTableRequest rename =
RenameTableRequest.builder().withSource(OTHER).withDestination(TABLE).build();
Assertions.assertDoesNotThrow(() ->
executor.renameTable(context(false), rename));
Assertions.assertTrue(catalog.tableExists(TABLE));
Assertions.assertFalse(catalog.tableExists(OTHER));
Assertions.assertTrue(manager.isNameOccupied(CATALOG_ID, "db", "t"));
} finally {
manager.close();
}
}
// --- helpers ---
private static IcebergConfig fastFailConfig() {
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");
return new IcebergConfig(config);
}
private BaseTable createTableWithDataFile() {
Table table = catalog.createTable(TABLE, SCHEMA);
DataFile dataFile =
DataFiles.builder(PartitionSpec.unpartitioned())
.withPath(DATA_FILE)
.withFileSizeInBytes(10L)
.withRecordCount(1L)
.build();
table.newAppend().appendFile(dataFile).commit();
BaseTable base = (BaseTable) catalog.loadTable(TABLE);
((InMemoryFileIO) base.io()).addFile(DATA_FILE, new byte[] {1});
return base;
}
private static List<String> reachableFiles(BaseTable table) {
List<String> files = new ArrayList<>();
files.add(table.operations().current().metadataFileLocation());
files.add(DATA_FILE);
for (Snapshot snapshot : table.snapshots()) {
files.add(snapshot.manifestListLocation());
for (ManifestFile manifest : snapshot.allManifests(table.io())) {
files.add(manifest.path());
}
}
return files;
}
/** Executor over a mocked REST wrapper that forwards to the in-memory
Iceberg catalog. */
private IcebergTableOperationExecutor executor(
IcebergCleanupManager manager, Class<? extends FileIO> workerFileIO) {
CatalogWrapperForREST wrapper = mock(CatalogWrapperForREST.class);
when(wrapper.loadTableMetadata(any()))
.thenAnswer(
inv -> ((BaseTable)
catalog.loadTable(inv.getArgument(0))).operations().current());
doAnswer(
inv -> {
catalog.dropTable(inv.getArgument(0), false);
return null;
})
.when(wrapper)
.dropTable(any());
when(wrapper.fileIOImpl()).thenReturn(workerFileIO.getName());
when(wrapper.fileIOProperties()).thenReturn(ImmutableMap.of());
when(wrapper.createTable(any(), any(), anyBoolean()))
.thenAnswer(
inv -> {
Namespace namespace = inv.getArgument(0);
CreateTableRequest request = inv.getArgument(1);
Table table =
catalog.createTable(
TableIdentifier.of(namespace, request.name()),
request.schema());
return LoadTableResponse.builder()
.withTableMetadata(((BaseTable)
table).operations().current())
.build();
});
doAnswer(
inv -> {
RenameTableRequest request = inv.getArgument(0);
catalog.renameTable(request.source(), request.destination());
return null;
})
.when(wrapper)
.renameTable(any());
IcebergCatalogWrapperManager wrapperManager =
mock(IcebergCatalogWrapperManager.class);
when(wrapperManager.getCatalogWrapper(CATALOG_NAME)).thenReturn(wrapper);
return new IcebergTableOperationExecutor(wrapperManager,
Optional.of(manager));
}
private static IcebergRequestContext context(boolean asyncPurge) {
IcebergRequestContext context = mock(IcebergRequestContext.class);
when(context.catalogName()).thenReturn(CATALOG_NAME);
when(context.userName()).thenReturn(asyncPurge ? "alice" :
AuthConstants.ANONYMOUS_USER);
when(context.asyncPurge()).thenReturn(asyncPurge);
return context;
}
private static CreateTableRequest createReq() {
return
CreateTableRequest.builder().withName("t").withSchema(SCHEMA).build();
}
/**
* FileIO the cleanup worker instantiates by class name (public no-arg
constructor for {@code
* CatalogUtil.loadFileIO}); forwards to the in-memory store shared with
the catalog.
*/
public static class DelegatingFileIO implements FileIO {
static volatile InMemoryFileIO delegate;
@Override
public InputFile newInputFile(String path) {
return delegate.newInputFile(path);
}
@Override
public OutputFile newOutputFile(String path) {
return delegate.newOutputFile(path);
}
@Override
public void deleteFile(String path) {
delegate.deleteFile(path);
}
}
/** Every read fails (credentials revoked / expired); the worker never
gets to a delete. */
public static class ReadDeniedFileIO extends DelegatingFileIO {
@Override
public InputFile newInputFile(String path) {
throw new RuntimeException("AccessDenied: GetObject " + path);
}
}
/** Reads work, every delete fails (a role without DeleteObject). */
public static class DeleteDeniedFileIO extends DelegatingFileIO {
@Override
public void deleteFile(String path) {
throw new RuntimeException("AccessDenied: DeleteObject " + path);
}
}
}
```
</details>
**Related: rename bypasses the reservation.**
[`renameTable`](https://github.com/apache/gravitino/blob/bb634de/iceberg/iceberg-rest-server/src/main/java/org/apache/gravitino/iceberg/service/dispatcher/IcebergTableOperationExecutor.java#L200-L205)
validates the destination name but never calls `rejectIfBeingPurged`, so `POST
/iceberg/v1/{prefix}/tables/rename` onto a `PENDING` or `RUNNING` identifier
succeeds while `createTable` at that identifier returns `409`. Small enough to
fix alongside; can be split into its own issue if preferred.
**A terminal state that needs human intervention has to be visible and
actionable.** Today `FAILED` is neither: no metric, no event, no distinct error
on the next create, and no reset path. The design planned all four and they are
still open ([§5.9
checklist](https://github.com/apache/gravitino/blob/bb634de/design-docs/async-iceberg-rest-hard-deletion.md#L698-L712)).
Releasing the name on top of that means the one moment somebody would notice,
the next create, passes silently. Either `FAILED` keeps the name until it is
acted on, or it has to ship with the alerting and the reset path the design
describes. As it stands the behavior contradicts the feature's own contract
("409 until the file cleanup finishes") in the one case where the files are
known not to be gone.
### How should we improve?
Other REST catalogs avoid the hazard at the location level rather than by
reserving names: Polaris refuses overlapping table locations and can add an
unpredictable suffix to default locations
(`DEFAULT_UNIQUE_TABLE_LOCATION_ENABLED`, [config
reference](https://polaris.apache.org/releases/1.7.0/configuration/configuration-reference/));
Lakekeeper puts a `{uuid}` segment in every default location and rejects
parent or sub-folder overlap ([storage
docs](https://docs.lakekeeper.io/docs/0.13.x/storage/)). Gravitino's REST
server passes the backend catalog's `<warehouse>/<namespace>/<table>` through
unchanged, so the name reservation is currently the only guard.
**Proposed:** 1 + 2 + 3 now; 5 as the longer-term direction. The full set of
alternatives:
1. **Keep `FAILED` reserving the name** (add `FAILED` to the
`selectUnfinishedJobId` predicate) until an operator clears the row. One
predicate and a test; honours "until the file cleanup finishes"; makes the
failure loud at the next create. Needs (3) to be practical, otherwise the name
stays blocked until the row is deleted by hand or pruned.
2. **Say why on create.** With or without (1), a `FAILED` row should produce
a distinct message ("previous purge of db.t failed after N attempts:
<last_error>; files may remain at <location>") instead of a silent success or
the generic "being purged".
3. **Operator reset / re-enqueue** for `FAILED` rows (`FAILED -> PENDING`,
`attempts = 0`), as an admin endpoint or a documented procedure. The design
already calls re-enqueue a manual action; there is no supported way to do it.
4. **Record table UUID and root location in the job row**, so a `FAILED` row
can be reconciled against the live catalog and the reservation can also be
checked by location prefix (covers explicit-`location` recreates).
5. **Unique default locations** for tables created through the REST server
(UUID suffix, as Polaris and Lakekeeper do). Removes the hazard for new tables
regardless of job state; a larger change with layout implications.
6. **Config flag** (`async-cleanup.failed-blocks-reuse`) if there is no
consensus on the default.
Fixing the rename gap (call `rejectIfBeingPurged` on the destination) is
independent of the choice above.
cc @roryqi, since this touches the documented trade-off in the async purge
design.
--
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]
