This is an automated email from the ASF dual-hosted git repository.
roryqi pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/gravitino.git
The following commit(s) were added to refs/heads/main by this push:
new df532dbf03 [#11298] feat(iceberg-rest): add async cleanup manager
(#11299)
df532dbf03 is described below
commit df532dbf035fa6eeee48c92276341d61f82630e9
Author: roryqi <[email protected]>
AuthorDate: Tue Jun 2 17:18:41 2026 +0800
[#11298] feat(iceberg-rest): add async cleanup manager (#11299)
### What changes were proposed in this pull request?
- Add `IcebergCleanupManager`: a server-wide worker pool that polls the
`iceberg_cleanup_job` store for `PENDING` jobs, claims them via a
heartbeat lease, deletes the dropped table's reachable files in bulk
through a shared `deleteExecutor` (with `CallerRunsPolicy`
back-pressure), renews heartbeats on a separate scheduler thread, drives
the retry state machine (transient failure → back to `PENDING`, give up
at `max-attempts` → `FAILED`), and prunes finished rows past the
retention window.
- Add `TestIcebergCleanupManager` unit tests, running against the
H2/MySQL/PostgreSQL backend matrix via `TestJDBCBackend`.
- Sync the async hard-deletion design doc with the merged persistence
layer (the `dropTable` enqueue snippet and the testing section).
### Why are the changes needed?
Second of three stacked PRs for async hard deletion in the Iceberg REST
catalog. The persistence layer (#11266) added the durable job store;
this PR adds the worker engine that drains jobs and deletes files. REST
integration follows in PR 3.
Fix: #11298
### Does this PR introduce _any_ user-facing change?
No. The manager is not yet wired into the REST drop path (that lands in
PR 3), and the config keys (`async-cleanup.*`) were introduced with the
persistence layer (#11266).
### How was this patch tested?
`./gradlew :iceberg:iceberg-rest-server:test --tests
"org.apache.gravitino.iceberg.service.cleanup.TestIcebergCleanupManager"
-PskipITs` — 7 tests pass (lifecycle to `SUCCEEDED`, transient-retry to
`FAILED`, bulk-delete batching, already-deleted-file tolerance,
reachable-file cleanup, and tombstone delegation).
---
design-docs/async-iceberg-rest-hard-deletion.md | 253 ++++++++-----
.../service/cleanup/IcebergCleanupJobStore.java | 42 +--
.../service/cleanup/IcebergCleanupManager.java | 389 ++++++++++++++++++++
.../cleanup/mapper/IcebergCleanupJobMapper.java | 6 +-
.../IcebergCleanupJobSQLProviderFactory.java | 10 +-
.../base/IcebergCleanupJobBaseSQLProvider.java | 20 +-
.../AbstractIcebergCleanupJobStoreBackendTest.java | 46 ++-
.../service/cleanup/TestIcebergCleanupManager.java | 406 +++++++++++++++++++++
8 files changed, 1022 insertions(+), 150 deletions(-)
diff --git a/design-docs/async-iceberg-rest-hard-deletion.md
b/design-docs/async-iceberg-rest-hard-deletion.md
index 34315ae98e..69a702d451 100644
--- a/design-docs/async-iceberg-rest-hard-deletion.md
+++ b/design-docs/async-iceberg-rest-hard-deletion.md
@@ -19,12 +19,12 @@
# Design: Asynchronous Hard Deletion for the Gravitino Iceberg REST Server
-| Field | Value |
-| -------- | ------------------------------------------------------- |
-| Status | Complete |
-| Authors | @roryqi |
-| Created | 2026-05-19 |
-| Module | `iceberg/iceberg-rest-server`, `iceberg/iceberg-common` |
+| Field | Value |
+| ------- | ------------------------------------------------------- |
+| Status | Complete |
+| Authors | @roryqi |
+| Created | 2026-05-19 |
+| Module | `iceberg/iceberg-rest-server`, `iceberg/iceberg-common` |
---
@@ -98,14 +98,14 @@ simpler design with a smaller bug surface. We chose a
single async
implementation gated behind a per-request opt-in, with synchronous deletion
as the default.
-| Approach | Pros
|
Cons
| Decision
|
-|-------------------------------------------------------------|------------------------------------------------------------------------------------------|--------------------------------------------------------------------------------------------------|-------------------------------------------------------------------------|
-| Synchronous only (status quo) | Simplest;
strongest "deleted means gone" guarantee
| Exceeds HTTP timeouts, saturates Jetty, no retry/audit on large tables
| **Rejected** — the problem we are solving
|
-| Pluggable `IcebergPurger` SPI | Extensible
without code changes |
Real added surface (SPI, discovery factory, context) with **no** second
implementation in flight | **Rejected** — revisit when a second implementation
has a real customer |
-| Reuse `RelationalGarbageCollector` | Proven
worker/scheduling pattern
| Different IO surface (object store vs. JDBC) and failure model
| **Rejected** — share patterns, not code
|
-| External job system (Quartz / Temporal) | Mature
scheduling, retries, observability
| Heavy operational burden on every operator
| **Rejected** — disproportionate for one workload
|
-| Enumerate files at enqueue time | Worker needs
no metadata re-read |
Slow on large tables (defeats the latency goal), bloats job rows
| **Rejected** — store `metadata_location`, re-read at run
time |
-| Object-store job markers (no DB table) | No schema /
migration |
Hand-rolled lease + renewal, no indexed scheduling, fragments per-bucket
| **Rejected** — higher net complexity (see below)
|
+| Approach | Pros
| Cons
| Decision
|
+| -------------------------------------------------------------------- |
----------------------------------------------------------------------------------------
|
------------------------------------------------------------------------------------------------
| ----------------------------------------------------------------------- |
+| Synchronous only (status quo) |
Simplest; strongest "deleted means gone" guarantee
| Exceeds HTTP timeouts, saturates Jetty, no retry/audit on large
tables | **Rejected** — the problem we are solving
|
+| Pluggable `IcebergPurger` SPI |
Extensible without code changes
| Real added surface (SPI, discovery factory, context) with **no**
second implementation in flight | **Rejected** — revisit when a second
implementation has a real customer |
+| Reuse `RelationalGarbageCollector` |
Proven worker/scheduling pattern
| Different IO surface (object store vs. JDBC) and failure model
| **Rejected** — share patterns, not code
|
+| External job system (Quartz / Temporal) |
Mature scheduling, retries, observability
| Heavy operational burden on every operator
| **Rejected** — disproportionate for one workload
|
+| Enumerate files at enqueue time |
Worker needs no metadata re-read
| Slow on large tables (defeats the latency goal), bloats job rows
| **Rejected** — store `metadata_location`, re-read
at run time |
+| Object-store job markers (no DB table) | No
schema / migration
| Hand-rolled lease + renewal, no indexed scheduling, fragments
per-bucket | **Rejected** — higher net complexity (see
below) |
| **JDBC job table + worker pool, async opt-in (synchronous default)** |
Smallest bug surface; restart-safe and cluster-safe via CAS claim; one code
path to test | No built-in extension point
| **Chosen**
|
**Why not an S3-only control plane?** Conditional writes (`If-None-Match`)
@@ -183,21 +183,22 @@ public void dropTable(IcebergRequestContext ctx,
TableIdentifier id,
TableMetadata metadata = w.loadTableMetadata(id);
w.dropTable(id); // metadata-only drop in the
catalog
- purgeJobStore.enqueue(
- IcebergPurgeJob.builder()
- .catalogName(ctx.catalogName())
- .tableIdentifier(id)
- .metadataLocation(metadata.metadataFileLocation())
- .fileIoImpl(w.fileIoImpl())
- .fileIoProperties(w.fileIoProperties())
- .createdBy(ctx.userPrincipal())
- .build());
+ cleanupManager.addJob(
+ new IcebergCleanupJob(
+ 0L, // id assigned by IdGenerator at
enqueue
+ catalogId, // resolved catalog entity id
+ id.namespace().toString(),
+ id.name(),
+ metadata.metadataFileLocation(),
+ w.fileIOImpl(),
+ w.fileIOProperties(),
+ ctx.userPrincipal()));
}
```
Order matters on the async path: load metadata location → drop catalog
entry → enqueue the job. A cleanup job exists only for a table already gone
-from the catalog. `fileIoProperties` is captured at enqueue time so the
+from the catalog. `fileIOProperties` is captured at enqueue time so the
worker can reconstruct `FileIO` even if the catalog is later reconfigured.
The enqueued row also serves as the **tombstone** that blocks name reuse
while the files still exist (§5.7).
@@ -217,33 +218,57 @@ already terminal, so it is consistent with the §5.7
tombstone table.
```sql
CREATE TABLE IF NOT EXISTS `iceberg_cleanup_job` (
- `id` BIGINT(20) UNSIGNED NOT NULL AUTO_INCREMENT,
- `metalake_name` VARCHAR(128) NOT NULL,
- `catalog_name` VARCHAR(128) NOT NULL,
- `namespace` VARCHAR(512) NOT NULL,
- `table_name` VARCHAR(256) NOT NULL,
- `metadata_location` VARCHAR(1024) NOT NULL,
- `file_io_impl` VARCHAR(256) NOT NULL,
- `file_io_props` MEDIUMTEXT NOT NULL COMMENT 'JSON',
+ `id` BIGINT(20) UNSIGNED NOT NULL COMMENT 'globally unique
cleanup job id',
+ `catalog_id` BIGINT(20) UNSIGNED NOT NULL COMMENT 'globally unique
id of the owning catalog, stable across catalog rename',
+ `namespace` VARCHAR(512) NOT NULL COMMENT 'namespace of the table
to be cleaned up',
+ `table_name` VARCHAR(256) NOT NULL COMMENT 'name of the table to be
cleaned up',
+ `metadata_location` MEDIUMTEXT NOT NULL COMMENT 'location of the table
metadata file to purge',
+ `file_io_impl` VARCHAR(256) NOT NULL COMMENT 'FileIO implementation
class used to access the table files',
+ `file_io_props` MEDIUMTEXT NOT NULL COMMENT 'JSON-encoded FileIO
properties',
`state` VARCHAR(16) NOT NULL COMMENT
'PENDING|RUNNING|SUCCEEDED|FAILED',
- `attempts` INT(10) NOT NULL DEFAULT 0,
- `last_error` VARCHAR(2048) NULL COMMENT 'truncated reason for the
most recent failure; NULL until a job fails',
- `heartbeat_at` BIGINT(20) NULL COMMENT 'last heartbeat from the
worker; NULL when unclaimed',
+ `attempts` INT(10) NOT NULL DEFAULT 0 COMMENT 'number of
processing attempts made so far',
+ `last_error` VARCHAR(2048) NULL COMMENT 'truncated reason for the
most recent failure, NULL until a job fails',
+ `heartbeat_at` BIGINT(20) NOT NULL DEFAULT 0 COMMENT 'last heartbeat
from the worker, 0 when not running',
`created_by` VARCHAR(128) NOT NULL COMMENT 'principal that requested
the drop (audit)',
- `updated_at` BIGINT(20) NOT NULL COMMENT 'last state change;
drives poll ordering and terminal-row pruning',
+ `updated_at` BIGINT(20) NOT NULL COMMENT 'last state change,
drives poll ordering and old finished-job cleanup',
PRIMARY KEY (`id`),
KEY `idx_state_updated` (`state`, `updated_at`),
- KEY `idx_object` (`catalog_name`, `namespace`, `table_name`, `state`)
-) ENGINE=InnoDB;
+ KEY `idx_object` (`catalog_id`, `namespace`(255), `table_name`(128), `state`)
+) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_bin COMMENT 'async
Iceberg table cleanup jobs';
```
We store only `metadata_location`, not the file list — enumeration is slow
on large tables, and `TableMetadataParser.read(io, location)` rebuilds the
snapshot graph deterministically when the worker runs.
+Row `id` is **not** a DB `AUTO_INCREMENT` column: the server allocates it from
+Gravitino's `IdGenerator` at enqueue time, and every timestamp (`heartbeat_at`,
+`updated_at`) is supplied by the application rather than the database. This
keeps
+one parameterized set of `INSERT` / `UPDATE` statements portable across H2,
+MySQL, and PostgreSQL, served by a single shared SQL provider (§5.5).
+The object is keyed by `catalog_id`, not by metalake/catalog *names*. Async
+cleanup only runs when the Iceberg REST server is embedded in Gravitino (the
job
+store reuses the entity store's relational backend), so a real catalog entity —
+and its id — always exists, and the request thread reads it for free:
+`catalogDispatcher.loadCatalog(...)` returns the `BaseCatalog`, whose
+`CatalogEntity` is already in memory, so `catalog.entity().id()` needs no extra
+entity-store round trip. `catalog_id` is globally unique (it is
+`catalog_meta`'s primary key), so it identifies the catalog without scoping by
+metalake, and it is **stable across catalog rename**: a rename mid-cleanup
would
+leave a name-keyed tombstone stranded while the physical warehouse prefix is
+unchanged, but the id-keyed tombstone still matches. The leaf `namespace` and
+`table_name` stay as strings — the table is already dropped at enqueue, so
there
+is no table entity/id to reference. `catalog_id` indexes in full, while
+`namespace` and `table_name` are prefix-indexed on MySQL (`namespace(255)`,
+`table_name(128)` — table names also cap at 128) to keep `idx_object` within
+MySQL's 3072-byte index key-length limit. The job store reuses the entity
+store's relational backend —
+its connection pool, transaction management, and per-backend dispatch — instead
+of opening its own JDBC connections.
+
A single column drives retries: `attempts` counts failures so the worker
gives up at the retry ceiling. A failed job returns to `PENDING` and is
-re-claimed on a later poll, so the poll cadence (`poll-interval-ms`,
+re-claimed on a later poll, so the poll cadence (`poll-interval-secs`,
§5.10) is the retry interval — no separate scheduling column is needed. The
ceiling is a config (`max-attempts`, §5.10), not a per-row `max_attempts`
column, since it is the same for every job.
@@ -268,7 +293,7 @@ polls. Capacity-gating is therefore by construction — no
central scheduler
claims jobs ahead of the threads, the pool never holds a claimed-but-unstarted
job, and a `RUNNING` row always maps to a thread actively deleting files (its
heartbeat reflects real progress, not a job idling in a queue). A poll that
-finds nothing claimable waits `poll-interval-ms` and retries. `worker-threads`
+finds nothing claimable waits `poll-interval-secs` and retries.
`worker-threads`
thus bounds the concurrent jobs per server; there is no separate claim-batch
size.
@@ -287,59 +312,90 @@ first row it wins:
-- a free thread reads a small candidate window
SELECT id FROM iceberg_cleanup_job
WHERE state = 'PENDING'
- OR (state = 'RUNNING'
- AND (heartbeat_at IS NULL OR heartbeat_at < :now - :heartbeatTimeout))
+ OR (state = 'RUNNING' AND heartbeat_at < :now - :heartbeatTimeout)
ORDER BY updated_at LIMIT :window;
-- claim one candidate; the row is ours only if affected_rows = 1
UPDATE iceberg_cleanup_job
- SET state='RUNNING', heartbeat_at=:now
+ SET state='RUNNING', heartbeat_at=:now, updated_at=:now
WHERE id=:id
AND ( state='PENDING'
- OR (state='RUNNING'
- AND (heartbeat_at IS NULL OR heartbeat_at < :now -
:heartbeatTimeout)) );
+ OR (state='RUNNING' AND heartbeat_at < :now - :heartbeatTimeout) );
```
A loser sees `affected_rows = 0` and tries the next id in its window. The
window is a small constant that only gives a CAS loser alternatives — not a
-capacity control, since a free thread still claims exactly one job. Where
-`SELECT … FOR UPDATE SKIP LOCKED` is available (MySQL 8+, PostgreSQL) the
-worker uses it to cut contention up front; the CAS form is the portable
-fallback (H2, older MySQL). `SKIP LOCKED` is purely an optimization — the
-claiming `UPDATE` is the serialization point, so two replicas can never claim
-the same job. We do not use DB-specific advisory locks.
+capacity control, since a free thread still claims exactly one job. The same
+portable CAS runs on **every** backend: H2, MySQL, and PostgreSQL share one SQL
+provider, so we do **not** use `SELECT … FOR UPDATE SKIP LOCKED` or DB-specific
+advisory locks. The claiming `UPDATE` is the serialization point, so two
+replicas can never claim the same job; a backend that ever needs a divergent
+statement can register its own provider without touching the rest of the
design.
**Worker loop (one free thread):**
1. **Poll & claim.** Read the candidate window above and CAS the first
winnable row to `RUNNING` with a fresh `heartbeat_at`. If nothing is
- claimable, wait `poll-interval-ms` and repeat.
-2. **Load metadata.** `TableMetadataParser.read(io, metadata_location)`. A
- transient failure releases the job for retry; a terminal one fails it
+ claimable, wait `poll-interval-secs` and repeat.
+2. **Load metadata.** `TableMetadataParser.read(io, metadata_location)`. The
+ root `metadata.json` is the only file whose absence means "the table is
+ already gone": a `NotFoundException` here completes the job (a prior attempt
+ already deleted it). Any other failure (e.g. corrupt/unreadable metadata)
+ releases the job for retry, failing it once `attempts` hits the ceiling
(§5.6).
-3. **Stream & delete.** Walk the snapshot graph lazily, group reachable files
- into batches of `delete-batch-size`, and hand each batch to the shared
- `deleteExecutor` for a **bulk** delete; `NotFoundException` counts as
- deleted. The worker thread itself never issues the storage call directly,
- and a *separate* background task keeps `heartbeat_at` fresh throughout
- (see "Heartbeat is decoupled from deletion" below).
+3. **Delete leaves before parents.** Walk the snapshot graph and delete one
+ dependency level at a time — data files → manifests → manifest lists →
+ statistics → ancestor metadata → the root `metadata.json` **last** — only
+ advancing once the current level is fully gone. Within a level, files are
+ independent: group them into batches of `delete-batch-size` and hand each to
+ the shared `deleteExecutor` for a **bulk** delete; `NotFoundException`
counts
+ as deleted. Leaf-first ordering keeps every crash recoverable: the root
+ pointer (and the manifests/manifest lists above any leaf still on disk)
+ survives, so a retry re-enumerates and finishes. Deleting a parent first
+ would strand its children — and a `metadata.json` removed early would be
+ misread as "table already gone", leaking everything beneath it. The worker
+ thread itself never issues the storage call directly, and a *separate*
+ background task keeps `heartbeat_at` fresh throughout (see "Heartbeat is
+ decoupled from deletion" below).
4. **Finish.** Once every reachable file is gone the job is `SUCCEEDED`
(§5.6). The thread then loops back to step 1.
-Execution mirrors `CatalogHandlers.purgeTable`, but **streams** the
-reachable files instead of materializing them. A large table can reference
-millions of data files, so the worker walks the snapshot graph lazily and
-deletes them **in bulk batches**, never one file at a time:
+Execution mirrors `CatalogHandlers.purgeTable`, deleting **in bulk batches**
+and never one file at a time. To keep every crash recoverable it deletes one
+dependency level at a time — leaves first, the root `metadata.json` last — and
+each `deleteAll` blocks until its level is fully gone before the next begins.
+Data files are the only unbounded level, so they are streamed and deleted one
+manifest at a time rather than collected up front — a million-file table never
+materializes its whole path set; only the far smaller manifest, manifest-list
+and metadata path lists are held at once:
```java
TableMetadata meta = TableMetadataParser.read(io, job.metadataLocation());
-try (CloseableIterable<String> files = reachableFiles(meta)) { // lazy, not
materialized
- Iterators.partition(files.iterator(), deleteBatchSize) // bounded
batches
+Set<String> manifests = new LinkedHashSet<>();
+for (Snapshot s : meta.snapshots()) {
+ for (ManifestFile m : s.allManifests(io)) {
+ if (!manifests.add(m.path())) continue; // dedup shared
manifests
+ try (CloseableIterable<String> dataFiles = // lazy, one
batch resident at a time
+ ManifestFiles.readPaths(m, io, meta.specsById())) {
+ deleteAll(io, dataFiles); // leaves:
streamed, then deleted
+ }
+ }
+}
+deleteAll(io, manifests); // then their
manifests
+deleteAll(io, ReachableFileUtil.manifestListLocations(table));
+deleteAll(io, ReachableFileUtil.statisticsFilesLocations(table));
+deleteAll(io, ancestorMetadata); // older
metadata.json
+deleteAll(io, List.of(job.metadataLocation())); // root pointer,
deleted last
+
+// deleteAll: bulk-batch the (possibly lazy) iterable, then await it
+void deleteAll(FileIO io, Iterable<String> files) {
+ Iterators.partition(files.iterator(), deleteBatchSize) // bounded batches
.forEachRemaining(batch ->
- deleteExecutor.execute(() ->
+ futures.add(deleteExecutor.submit(() ->
// SupportsBulkOperations.deleteFiles when the FileIO supports it
// (e.g. S3FileIO's batch-delete API), else concurrent per-file
- CatalogUtil.deleteFiles(io, batch, "cleanup", /* bulk */ true)));
+ CatalogUtil.deleteFiles(io, batch, "cleanup", /* bulk */
true))));
+ awaitAll(futures); //
NotFoundException counts as deleted
}
```
@@ -417,19 +473,20 @@ rows age out for reclaim.
Per-file failures are logged but do not fail the whole job — the
synchronous purge has the same "best effort" stance. A job fails only if the
-**metadata phase** fails. `NotFoundException` from `deleteFile` counts as
-success. A transient failure goes back to `PENDING` and is retried when a
-free thread next polls (§5.5), incrementing `attempts` each time; when
`attempts`
-reaches `max-attempts` the job goes to `FAILED` instead of `PENDING`. A
-clearly terminal failure skips the retries and goes straight to `FAILED`.
-
-| Outcome | Action
|
-|------------------------------------------------------|---------------------------------------------------------------------------------------|
-| All files deleted (or already gone) | `state='SUCCEEDED'`
|
-| Transient failure, `attempts < max-attempts` | `attempts++`, set
`last_error`, back to `PENDING`, `heartbeat_at=NULL`; re-claimed later |
-| Transient failure, `attempts` reaches `max-attempts` | `attempts++`, set
`last_error` → `FAILED` (gave up retrying) |
-| Terminal failure (e.g. metadata gone/corrupt) | set `last_error` →
`FAILED` immediately; retrying cannot help |
-| Worker killed mid-job | `RUNNING` row's
heartbeat goes stale; another worker reclaims; deletes are idempotent |
+**metadata phase** fails. A `NotFoundException` counts as success rather than a
+failure: a missing root `metadata.json` means the table is already gone, and a
+missing data file, manifest, or manifest list (from `deleteFile` or while
+enumerating reachable files) is already deleted. Any other failure goes back to
+`PENDING` and is retried when a free thread next polls (§5.5), incrementing
+`attempts` each time; when `attempts` reaches `max-attempts` the job goes to
+`FAILED` instead of `PENDING`.
+
+| Outcome | Action
|
+| ------------------------------------------ |
-------------------------------------------------------------------------------------
|
+| All files deleted (or already gone) | `state='SUCCEEDED'`
|
+| Failure, `attempts < max-attempts` | `attempts++`, set `last_error`,
back to `PENDING`, `heartbeat_at=0`; re-claimed later |
+| Failure, `attempts` reaches `max-attempts` | `attempts++`, set `last_error`
→ `FAILED` (gave up retrying) |
+| Worker killed mid-job | `RUNNING` row's heartbeat goes
stale; another worker reclaims; deletes are idempotent |
**After `SUCCEEDED`.** The files are gone, so the tombstone lifts at once:
`createTable` / `register` at the identifier succeed again (§5.7). The
@@ -476,7 +533,7 @@ tombstone — no second table is needed. The REST server
consults the store on
the request thread (one indexed lookup via `idx_object`):
| Operation | Active job exists
| No active job (`SUCCEEDED`/`FAILED`/none) |
-|---------------------------------------------|-----------------------------------|-------------------------------------------|
+| ------------------------------------------- |
--------------------------------- | ----------------------------------------- |
| `loadTable` / `HEAD`, `alterTable` / commit | `404 NoSuchTableException`
| `404` |
| `dropTable` (same identifier, repeated) | `404 NoSuchTableException`
| `404` |
| `createTable` (same identifier) | **`409 Conflict`** — being
purged | succeeds |
@@ -575,15 +632,15 @@ per-request **client** choice via the
`X-Gravitino-Async-Purge` header
(absent ⇒ synchronous; `true` opts into async deletion). The server-side keys
only tune the worker pool and retries:
-| Key | Default |
Description |
-|---------------------------------------------------------------|----------|------------------------------------------------------------------------------|
-| `gravitino.iceberg-rest.async-purge.worker-threads` | `2` |
Worker pool size per server (concurrent jobs). |
-| `gravitino.iceberg-rest.async-purge.delete-threads` | `4` |
Server-wide file-delete pool size, shared across all jobs. |
-| `gravitino.iceberg-rest.async-purge.delete-batch-size` | `1000` |
Files per bulk-delete batch handed to `deleteExecutor` (§5.5). |
-| `gravitino.iceberg-rest.async-purge.poll-interval-ms` | `5000` |
Worker poll interval. |
-| `gravitino.iceberg-rest.async-purge.heartbeat-timeout-ms` | `300000` |
Age after which a job with no heartbeat is reclaimable. |
-| `gravitino.iceberg-rest.async-purge.max-attempts` | `5` |
Attempts before `FAILED`. |
-| `gravitino.iceberg-rest.async-purge.retention-hours` | `720` |
How long terminal (`SUCCEEDED` / `FAILED`) rows are retained before pruning (30
days). |
+| Key | Default |
Description
|
+| ------------------------------------------------------------- | ------- |
--------------------------------------------------------------------------------------
|
+| `gravitino.iceberg-rest.async-cleanup.worker-threads` | `2` |
Worker pool size per server (concurrent jobs).
|
+| `gravitino.iceberg-rest.async-cleanup.delete-threads` | `4` |
Server-wide file-delete pool size, shared across all jobs.
|
+| `gravitino.iceberg-rest.async-cleanup.delete-batch-size` | `1000` |
Files per bulk-delete batch handed to `deleteExecutor` (§5.5).
|
+| `gravitino.iceberg-rest.async-cleanup.poll-interval-secs` | `5` |
Worker poll interval in seconds; also the retry interval.
|
+| `gravitino.iceberg-rest.async-cleanup.heartbeat-timeout-secs` | `300` |
Age in seconds after which a job with no heartbeat is reclaimable.
|
+| `gravitino.iceberg-rest.async-cleanup.max-attempts` | `5` |
Attempts before `FAILED`.
|
+| `gravitino.iceberg-rest.async-cleanup.retention-hours` | `720` |
How long terminal (`SUCCEEDED` / `FAILED`) rows are retained before pruning (30
days). |
The thread defaults are deliberately modest. Each `delete-threads` thread now
issues a *bulk* delete of up to `delete-batch-size` files per call (§5.5), so
@@ -635,8 +692,8 @@ opt-in path selected per request via the
`X-Gravitino-Async-Purge` header.
### Phase 1 (1.3): Async purge core
- [ ] Add the `iceberg_cleanup_job` schema and migrations (MySQL, H2,
PostgreSQL)
-- [ ] Implement `IcebergPurgeJobStore` enqueue path (persist job, return)
-- [ ] Implement the worker pool: CAS claiming with heartbeat ownership (`FOR
UPDATE SKIP LOCKED` where available), heartbeat renewal on a thread decoupled
from deletion, streaming **bulk** file deletion (`CatalogUtil.deleteFiles(...,
true)`) through a bounded `deleteExecutor` with `CallerRunsPolicy`
back-pressure, retry state machine that records `last_error` (re-claim failed
jobs on later polls, give up at `max-attempts`)
+- [ ] Implement `IcebergCleanupJobStore` enqueue path (persist job, return)
+- [ ] Implement the worker pool: portable CAS claiming with heartbeat
ownership (one SQL provider for H2/MySQL/PostgreSQL, no `SKIP LOCKED`),
heartbeat renewal on a thread decoupled from deletion, streaming **bulk** file
deletion (`CatalogUtil.deleteFiles(..., true)`) through a bounded
`deleteExecutor` with `CallerRunsPolicy` back-pressure, retry state machine
that records `last_error` (re-claim failed jobs on later polls, give up at
`max-attempts`)
- [ ] Honor the `X-Gravitino-Async-Purge` request header and wire both paths
into `IcebergTableOperationExecutor.dropTable`
- [ ] Add observability (§5.9): metrics and the informative `ErrorResponse`
message on the §5.7 `409`; operator read path in 1.3 is direct DB query
- [ ] Enforce tombstone semantics (§5.7): on `createTable`/`registerTable`,
reject with `409` when an active job exists (`idx_object` lookup on the request
thread)
@@ -659,11 +716,11 @@ opt-in path selected per request via the
`X-Gravitino-Async-Purge` header.
## 7. Testing
- Unit (`./gradlew :iceberg:iceberg-rest-server:test -PskipITs`):
- - `TestIcebergPurgeJobStore` — enqueue and row contents.
- - `TestIcebergPurgeStateMachine` — PENDING → RUNNING → SUCCEEDED;
- failure → retry (back to PENDING) → FAILED; `last_error` populated on
- each failure.
- - `TestIcebergPurgeWorker` — claiming, heartbeat renewal on a thread that
+ - `TestIcebergCleanupJobStore` — enqueue, row contents, and the state
+ machine: PENDING → RUNNING → SUCCEEDED; failure → retry (back to
+ PENDING) → FAILED; `last_error` populated on each failure. Runs against
+ the H2/MySQL/PostgreSQL backend matrix (`TestJDBCBackend`).
+ - `TestIcebergCleanupManager` — claiming, heartbeat renewal on a thread that
stays fresh while delete batches block, contention (H2), bulk-delete
batching, and `CallerRunsPolicy` back-pressure when the delete queue fills.
- `TestIcebergTableOperationExecutorAsyncPurge` — `X-Gravitino-Async-Purge:
diff --git
a/iceberg/iceberg-rest-server/src/main/java/org/apache/gravitino/iceberg/service/cleanup/IcebergCleanupJobStore.java
b/iceberg/iceberg-rest-server/src/main/java/org/apache/gravitino/iceberg/service/cleanup/IcebergCleanupJobStore.java
index 2bd3cd10a7..6f715ba1d9 100644
---
a/iceberg/iceberg-rest-server/src/main/java/org/apache/gravitino/iceberg/service/cleanup/IcebergCleanupJobStore.java
+++
b/iceberg/iceberg-rest-server/src/main/java/org/apache/gravitino/iceberg/service/cleanup/IcebergCleanupJobStore.java
@@ -93,52 +93,40 @@ public class IcebergCleanupJobStore {
}
/**
- * Marks a RUNNING job SUCCEEDED.
+ * Marks a RUNNING job SUCCEEDED, only if the caller still owns it.
*
* @param id job id
- * @return {@code true} iff the row was still RUNNING and was updated (i.e.
the caller still owned
- * the job)
+ * @param heartbeat the caller's heartbeat token; the update applies only if
the row's {@code
+ * heartbeat_at} still matches, so a reclaimed worker cannot flip a job
a peer now owns
+ * @return {@code true} iff the row was updated (still RUNNING and owned by
the caller)
*/
- public boolean markSucceeded(long id) {
+ public boolean markSucceeded(long id, long heartbeat) {
long now = System.currentTimeMillis();
return SessionUtils.doWithCommitAndFetchResult(
IcebergCleanupJobMapper.class,
- mapper -> mapper.markFinished(id,
IcebergCleanupJob.State.SUCCEEDED.name(), null, now))
+ mapper ->
+ mapper.markFinished(
+ id, IcebergCleanupJob.State.SUCCEEDED.name(), null, now,
heartbeat))
> 0;
}
/**
- * Marks a RUNNING job FAILED immediately.
- *
- * @param id job id
- * @param reason failure text
- * @return {@code true} iff the row was still RUNNING and was updated (i.e.
the caller still owned
- * the job)
- */
- public boolean markFailed(long id, String reason) {
- long now = System.currentTimeMillis();
- String err = truncate(reason);
- return SessionUtils.doWithCommitAndFetchResult(
- IcebergCleanupJobMapper.class,
- mapper -> mapper.markFinished(id,
IcebergCleanupJob.State.FAILED.name(), err, now))
- > 0;
- }
-
- /**
- * Records a transient failure: {@code attempts++}, then FAILED at the
ceiling else PENDING.
+ * Records a transient failure, only if the caller still owns the job:
{@code attempts++}, then
+ * FAILED at the ceiling else PENDING.
*
* @param id job id
* @param reason failure text
* @param maxAttempts ceiling from config
- * @return {@code true} iff the row was still RUNNING and was updated (i.e.
the caller still owned
- * the job)
+ * @param heartbeat the caller's heartbeat token; the update applies only if
the row's {@code
+ * heartbeat_at} still matches, so a reclaimed worker cannot disturb a
job a peer now owns
+ * @return {@code true} iff the row was updated (still RUNNING and owned by
the caller)
*/
- public boolean recordFailure(long id, String reason, int maxAttempts) {
+ public boolean recordFailure(long id, String reason, int maxAttempts, long
heartbeat) {
long now = System.currentTimeMillis();
String err = truncate(reason);
return SessionUtils.doWithCommitAndFetchResult(
IcebergCleanupJobMapper.class,
- mapper -> mapper.recordFailure(id, err, maxAttempts, now))
+ mapper -> mapper.recordFailure(id, err, maxAttempts, now,
heartbeat))
> 0;
}
diff --git
a/iceberg/iceberg-rest-server/src/main/java/org/apache/gravitino/iceberg/service/cleanup/IcebergCleanupManager.java
b/iceberg/iceberg-rest-server/src/main/java/org/apache/gravitino/iceberg/service/cleanup/IcebergCleanupManager.java
new file mode 100644
index 0000000000..f46b3c337d
--- /dev/null
+++
b/iceberg/iceberg-rest-server/src/main/java/org/apache/gravitino/iceberg/service/cleanup/IcebergCleanupManager.java
@@ -0,0 +1,389 @@
+/*
+ * 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.base.Throwables;
+import com.google.common.collect.Iterators;
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.LinkedHashSet;
+import java.util.List;
+import java.util.Map;
+import java.util.Optional;
+import java.util.Set;
+import java.util.concurrent.ArrayBlockingQueue;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.ExecutionException;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
+import java.util.concurrent.Future;
+import java.util.concurrent.ScheduledExecutorService;
+import java.util.concurrent.ThreadFactory;
+import java.util.concurrent.ThreadPoolExecutor;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicBoolean;
+import java.util.function.LongPredicate;
+import org.apache.gravitino.iceberg.common.IcebergConfig;
+import org.apache.iceberg.BaseTable;
+import org.apache.iceberg.CatalogUtil;
+import org.apache.iceberg.ManifestFile;
+import org.apache.iceberg.ManifestFiles;
+import org.apache.iceberg.ReachableFileUtil;
+import org.apache.iceberg.Snapshot;
+import org.apache.iceberg.StaticTableOperations;
+import org.apache.iceberg.Table;
+import org.apache.iceberg.TableMetadata;
+import org.apache.iceberg.TableMetadataParser;
+import org.apache.iceberg.exceptions.NotFoundException;
+import org.apache.iceberg.io.CloseableIterable;
+import org.apache.iceberg.io.FileIO;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+/**
+ * Server-wide async cleanup engine: claims {@code iceberg_cleanup_job} rows,
deletes the dropped
+ * table's files in bulk, and renews claim heartbeats on a thread decoupled
from deletion.
+ */
+public class IcebergCleanupManager implements AutoCloseable {
+
+ private static final Logger LOG =
LoggerFactory.getLogger(IcebergCleanupManager.class);
+
+ private final IcebergCleanupJobStore store;
+ private final int workerThreads;
+ private final int deleteBatchSize;
+ private final int maxAttempts;
+ private final int candidateWindow;
+ private final long pollIntervalMs;
+ private final long heartbeatTimeoutMs;
+ private final long retentionMs;
+ private final ThreadPoolExecutor deleteExecutor;
+ // Heartbeat token per job this manager currently owns, keyed by id. The
scheduler renews it and a
+ // worker reads it for the terminal CAS; refreshHeartbeats drops the entry
once a peer reclaims.
+ private final Map<Long, Long> ownedHeartbeats = new ConcurrentHashMap<>();
+
+ private final AtomicBoolean running = new AtomicBoolean(false);
+ private final AtomicBoolean closed = new AtomicBoolean(false);
+ private ExecutorService workers;
+ private ScheduledExecutorService scheduler;
+
+ /**
+ * Creates an async cleanup manager.
+ *
+ * @param store the cleanup job store backed by the entity store's
relational backend
+ * @param config Iceberg REST server config
+ */
+ public IcebergCleanupManager(IcebergCleanupJobStore store, IcebergConfig
config) {
+ this.store = store;
+ this.workerThreads =
config.get(IcebergConfig.ASYNC_CLEANUP_WORKER_THREADS);
+ int deleteThreads = config.get(IcebergConfig.ASYNC_CLEANUP_DELETE_THREADS);
+ this.deleteBatchSize =
config.get(IcebergConfig.ASYNC_CLEANUP_DELETE_BATCH_SIZE);
+ this.pollIntervalMs =
config.get(IcebergConfig.ASYNC_CLEANUP_POLL_INTERVAL_SECS) * 1000L;
+ this.heartbeatTimeoutMs =
+ config.get(IcebergConfig.ASYNC_CLEANUP_HEARTBEAT_TIMEOUT_SECS) * 1000L;
+ this.maxAttempts = config.get(IcebergConfig.ASYNC_CLEANUP_MAX_ATTEMPTS);
+ this.retentionMs = config.get(IcebergConfig.ASYNC_CLEANUP_RETENTION_HOURS)
* 3_600_000L;
+ // Scan more candidates than workers so a claim that loses its CAS still
has other rows to try
+ // in the same poll. workerThreads * 4 gives that headroom; the floor of 8
keeps the window
+ // useful when only one or two worker threads are configured.
+ this.candidateWindow = Math.max(8, workerThreads * 4);
+ this.deleteExecutor =
+ new ThreadPoolExecutor(
+ deleteThreads,
+ deleteThreads,
+ 60L,
+ TimeUnit.SECONDS,
+ new ArrayBlockingQueue<>(deleteThreads * 4),
+ daemon("iceberg-cleanup-delete"),
+ new ThreadPoolExecutor.CallerRunsPolicy());
+ }
+
+ /**
+ * Persists a new cleanup job.
+ *
+ * @param job job to persist
+ * @return generated id
+ */
+ public long addJob(IcebergCleanupJob job) {
+ return store.addJob(job);
+ }
+
+ /**
+ * Checks whether an unfinished cleanup job occupies a table identifier.
+ *
+ * @param catalogId globally unique id of the owning catalog
+ * @param namespace table namespace
+ * @param table table name
+ * @return true iff a PENDING or RUNNING job exists for the identifier
+ */
+ public boolean isNameOccupied(long catalogId, String namespace, String
table) {
+ return store.findUnfinishedJobId(catalogId, namespace, table).isPresent();
+ }
+
+ /** Starts worker threads and the heartbeat/prune scheduler. */
+ public void start() {
+ if (closed.get()) {
+ throw new IllegalStateException("Iceberg cleanup manager is already
closed");
+ }
+
+ // compareAndSet keeps concurrent or repeated start() calls from each
allocating a pool.
+ if (!running.compareAndSet(false, true)) {
+ return;
+ }
+
+ // We submit exactly workerThreads loops, so the queue is never used; it
is bounded only to
+ // avoid Executors.newFixedThreadPool's unbounded queue.
+ this.workers =
+ new ThreadPoolExecutor(
+ workerThreads,
+ workerThreads,
+ 0L,
+ TimeUnit.MILLISECONDS,
+ new ArrayBlockingQueue<>(workerThreads),
+ daemon("iceberg-cleanup-worker"));
+ for (int i = 0; i < workerThreads; i++) {
+ workers.submit(this::workerLoop);
+ }
+
+ // One scheduler thread runs both periodic tasks: heartbeat renewal and
row pruning.
+ this.scheduler = Executors.newScheduledThreadPool(1,
daemon("iceberg-cleanup-heartbeat-prune"));
+ long heartbeatIntervalMs = heartbeatTimeoutMs / 3L;
+ scheduler.scheduleAtFixedRate(
+ this::refreshHeartbeats, heartbeatIntervalMs, heartbeatIntervalMs,
TimeUnit.MILLISECONDS);
+ scheduler.scheduleAtFixedRate(this::prune, 1L, 1L, TimeUnit.HOURS);
+ }
+
+ @Override
+ public void close() {
+ if (!closed.compareAndSet(false, true)) {
+ return;
+ }
+
+ running.set(false);
+ if (scheduler != null) {
+ scheduler.shutdownNow();
+ }
+ if (workers != null) {
+ workers.shutdownNow();
+ awaitTermination(workers);
+ }
+ deleteExecutor.shutdownNow();
+ // Wait for in-flight delete batches to observe the interrupt and stop,
matching the workers
+ // above, so close() does not return while file deletions are still
running on a dying pool.
+ awaitTermination(deleteExecutor);
+ // The job store is backed by the entity store's shared relational
backend, which owns the
+ // connection pool lifecycle, so there is nothing to close here.
+ }
+
+ void cleanupFiles(FileIO io, String metadataLocation) {
+ TableMetadata metadata;
+ try {
+ metadata = TableMetadataParser.read(io, metadataLocation);
+ } catch (NotFoundException metadataAlreadyGone) {
+ // A missing root metadata.json means the table is already gone. Since
we delete it last, its
+ // absence proves every file under it was deleted first, so this is the
one NotFoundException
+ // we treat as success. Return and let runJob mark the job SUCCEEDED.
+ LOG.info("Cleanup metadata {} already absent; treating as done",
metadataLocation);
+ return;
+ }
+
+ Table table = new BaseTable(new StaticTableOperations(metadata, io),
"async-cleanup");
+
+ // Delete children before parents, root metadata.json last. Each deleteAll
blocks until its
+ // level is gone, so a crash always leaves the root (and the manifests
above any surviving file)
+ // readable for a retry to rebuild from. Deleting a parent first would
orphan its children.
+ //
+ // Data files are the only huge level, so they are streamed and deleted
one manifest at a time
+ // rather than all collected first; only the smaller
manifest/list/metadata paths are held.
+ Set<String> manifests = new LinkedHashSet<>();
+ deleteDataFiles(io, metadata, manifests);
+ deleteAll(io, manifests);
+ deleteAll(io, ReachableFileUtil.manifestListLocations(table));
+ deleteAll(io, ReachableFileUtil.statisticsFilesLocations(table));
+
+ // metadataFileLocations includes the current metadata.json; drop it so it
is deleted last.
+ Set<String> ancestorMetadata =
+ new LinkedHashSet<>(ReachableFileUtil.metadataFileLocations(table,
true));
+ ancestorMetadata.remove(metadataLocation);
+ deleteAll(io, ancestorMetadata);
+ deleteAll(io, Collections.singletonList(metadataLocation));
+ }
+
+ void deleteAll(FileIO io, Iterable<String> files) {
+ // Callers pass one manifest's files at a time (or a small fixed list), so
futures stay small;
+ // CallerRunsPolicy on deleteExecutor also throttles submission when the
pool is saturated.
+ List<Future<?>> futures = new ArrayList<>();
+ Iterators.partition(files.iterator(), deleteBatchSize)
+ .forEachRemaining(
+ batch ->
+ futures.add(
+ deleteExecutor.submit(
+ () -> CatalogUtil.deleteFiles(io, batch, "cleanup",
true))));
+
+ for (Future<?> future : futures) {
+ try {
+ future.get();
+ } catch (InterruptedException e) {
+ Thread.currentThread().interrupt();
+ throw new RuntimeException("Interrupted during bulk delete", e);
+ } catch (ExecutionException e) {
+ if
(Throwables.getCausalChain(e).stream().anyMatch(NotFoundException.class::isInstance))
{
+ LOG.debug("Ignoring already-deleted file during async cleanup", e);
+ continue;
+ }
+ throw new RuntimeException("Bulk delete batch failed", e);
+ }
+ }
+ }
+
+ private void workerLoop() {
+ while (running.get()) {
+ try {
+ long now = System.currentTimeMillis();
+ Optional<IcebergCleanupJob> job =
+ store.takePendingJob(now, heartbeatTimeoutMs, candidateWindow);
+ if (job.isEmpty()) {
+ sleep(pollIntervalMs);
+ continue;
+ }
+
+ ownedHeartbeats.put(job.get().id(), now);
+ runJob(job.get());
+ } catch (Throwable t) {
+ // The loop is submitted once, so if it exits the worker is gone for
good. Catch everything
+ // (including Errors) so a fault only backs off instead of killing the
worker.
+ if (t instanceof InterruptedException) {
+ Thread.currentThread().interrupt();
+ }
+ LOG.warn("Cleanup worker loop hit an unexpected error; backing off",
t);
+ sleep(pollIntervalMs);
+ }
+ }
+ }
+
+ private void runJob(IcebergCleanupJob job) {
+ long id = job.id();
+ // try-with-resources so the per-job FileIO (which may hold an S3 client /
connection pool) is
+ // closed on every path: success, transient failure, and the early return
inside cleanupFiles.
+ try (FileIO io = CatalogUtil.loadFileIO(job.fileIOImpl(),
job.fileIOProperties(), null)) {
+ cleanupFiles(io, job.metadataLocation());
+ finishJob(id, heartbeat -> store.markSucceeded(id, heartbeat));
+ } catch (RuntimeException e) {
+ LOG.warn("Cleanup job {} failed transiently; will retry", id, e);
+ finishJob(id, heartbeat -> store.recordFailure(id, e.getMessage(),
maxAttempts, heartbeat));
+ } finally {
+ ownedHeartbeats.remove(id);
+ }
+ }
+
+ // markSucceeded/recordFailure CAS on the heartbeat token, so a worker whose
lease a peer
+ // reclaimed cannot overwrite the job the peer now owns. A null token means
a refresh already
+ // saw the takeover, so we skip. A failed CAS just leaves the row RUNNING to
be reclaimed and
+ // re-run (which finds the files gone and succeeds); we log it so the
reclaim is observable.
+ void finishJob(long id, LongPredicate terminalUpdate) {
+ Long heartbeat = ownedHeartbeats.get(id);
+ if (heartbeat != null && !terminalUpdate.test(heartbeat)) {
+ LOG.warn("Could not finish cleanup job {}; it will be reclaimed and
re-run", id);
+ }
+ }
+
+ void refreshHeartbeats() {
+ long now = System.currentTimeMillis();
+ for (Map.Entry<Long, Long> entry : new
ArrayList<>(ownedHeartbeats.entrySet())) {
+ long id = entry.getKey();
+ try {
+ long previousHeartbeat = entry.getValue();
+ ownedHeartbeats.put(id, now);
+ if (!store.heartbeat(id, previousHeartbeat, now)) {
+ LOG.warn("Lost ownership of cleanup job {}", id);
+ ownedHeartbeats.remove(id, now);
+ }
+ } catch (Throwable t) {
+ ownedHeartbeats.replace(id, now, entry.getValue());
+ // scheduleAtFixedRate stops a task forever if it throws, so never let
one escape: a bad job
+ // must not stop heartbeat renewal for the whole process.
+ LOG.warn("Heartbeat update failed for job {}", id, t);
+ }
+ }
+ }
+
+ private void prune() {
+ try {
+ store.deleteFinishedJobsByLegacyTimeline(System.currentTimeMillis() -
retentionMs);
+ } catch (Throwable t) {
+ // As above: don't let a throw stop the recurring prune task.
+ LOG.warn("Cleanup-row pruning failed", t);
+ }
+ }
+
+ // Streams each manifest's data files to deleteAll (one manifest's paths in
memory at a time) and
+ // collects the manifest paths into `manifests` for the caller to delete
next.
+ private void deleteDataFiles(FileIO io, TableMetadata metadata, Set<String>
manifests) {
+ for (Snapshot snapshot : metadata.snapshots()) {
+ List<ManifestFile> snapshotManifests;
+ try {
+ snapshotManifests = snapshot.allManifests(io);
+ } catch (NotFoundException manifestListGone) {
+ // Manifest lists are deleted after everything under them, so a
missing one means a prior
+ // attempt already deleted this snapshot's files. Nothing left here;
skip it.
+ LOG.debug("Manifest list for snapshot {} already gone; skipping",
snapshot.snapshotId());
+ continue;
+ }
+ for (ManifestFile manifest : snapshotManifests) {
+ if (!manifests.add(manifest.path())) {
+ continue; // shared by several snapshots; its data files were
already deleted
+ }
+ try (CloseableIterable<String> paths =
+ ManifestFiles.readPaths(manifest, io, metadata.specsById())) {
+ // deleteAll pulls this lazy iterable in batches, so only one batch
is held at a time.
+ deleteAll(io, paths);
+ } catch (NotFoundException manifestGone) {
+ // Manifests are deleted after their data files, so a missing one
has no data files left.
+ LOG.debug("Manifest {} already gone; skipping", manifest.path());
+ } catch (Exception e) {
+ throw new RuntimeException("Failed to read manifest " +
manifest.path(), e);
+ }
+ }
+ }
+ }
+
+ private static void awaitTermination(ExecutorService pool) {
+ try {
+ pool.awaitTermination(5, TimeUnit.SECONDS);
+ } catch (InterruptedException e) {
+ Thread.currentThread().interrupt();
+ }
+ }
+
+ private static void sleep(long ms) {
+ try {
+ Thread.sleep(ms);
+ } catch (InterruptedException e) {
+ Thread.currentThread().interrupt();
+ }
+ }
+
+ private static ThreadFactory daemon(String name) {
+ return runnable -> {
+ Thread thread = new Thread(runnable, name);
+ thread.setDaemon(true);
+ return thread;
+ };
+ }
+}
diff --git
a/iceberg/iceberg-rest-server/src/main/java/org/apache/gravitino/iceberg/service/cleanup/mapper/IcebergCleanupJobMapper.java
b/iceberg/iceberg-rest-server/src/main/java/org/apache/gravitino/iceberg/service/cleanup/mapper/IcebergCleanupJobMapper.java
index 12c981fb2d..4e85bf2ed0 100644
---
a/iceberg/iceberg-rest-server/src/main/java/org/apache/gravitino/iceberg/service/cleanup/mapper/IcebergCleanupJobMapper.java
+++
b/iceberg/iceberg-rest-server/src/main/java/org/apache/gravitino/iceberg/service/cleanup/mapper/IcebergCleanupJobMapper.java
@@ -53,14 +53,16 @@ public interface IcebergCleanupJobMapper {
@Param("id") long id,
@Param("state") String state,
@Param("reason") String reason,
- @Param("now") long now);
+ @Param("now") long now,
+ @Param("heartbeat") long heartbeat);
@UpdateProvider(type = IcebergCleanupJobSQLProviderFactory.class, method =
"recordFailure")
int recordFailure(
@Param("id") long id,
@Param("reason") String reason,
@Param("maxAttempts") int maxAttempts,
- @Param("now") long now);
+ @Param("now") long now,
+ @Param("heartbeat") long heartbeat);
@UpdateProvider(type = IcebergCleanupJobSQLProviderFactory.class, method =
"heartbeat")
int heartbeat(
diff --git
a/iceberg/iceberg-rest-server/src/main/java/org/apache/gravitino/iceberg/service/cleanup/mapper/IcebergCleanupJobSQLProviderFactory.java
b/iceberg/iceberg-rest-server/src/main/java/org/apache/gravitino/iceberg/service/cleanup/mapper/IcebergCleanupJobSQLProviderFactory.java
index 7c1969fc3f..733a01c295 100644
---
a/iceberg/iceberg-rest-server/src/main/java/org/apache/gravitino/iceberg/service/cleanup/mapper/IcebergCleanupJobSQLProviderFactory.java
+++
b/iceberg/iceberg-rest-server/src/main/java/org/apache/gravitino/iceberg/service/cleanup/mapper/IcebergCleanupJobSQLProviderFactory.java
@@ -75,16 +75,18 @@ public class IcebergCleanupJobSQLProviderFactory {
@Param("id") long id,
@Param("state") String state,
@Param("reason") String reason,
- @Param("now") long now) {
- return getProvider().markFinished(id, state, reason, now);
+ @Param("now") long now,
+ @Param("heartbeat") long heartbeat) {
+ return getProvider().markFinished(id, state, reason, now, heartbeat);
}
public static String recordFailure(
@Param("id") long id,
@Param("reason") String reason,
@Param("maxAttempts") int maxAttempts,
- @Param("now") long now) {
- return getProvider().recordFailure(id, reason, maxAttempts, now);
+ @Param("now") long now,
+ @Param("heartbeat") long heartbeat) {
+ return getProvider().recordFailure(id, reason, maxAttempts, now,
heartbeat);
}
public static String heartbeat(
diff --git
a/iceberg/iceberg-rest-server/src/main/java/org/apache/gravitino/iceberg/service/cleanup/mapper/provider/base/IcebergCleanupJobBaseSQLProvider.java
b/iceberg/iceberg-rest-server/src/main/java/org/apache/gravitino/iceberg/service/cleanup/mapper/provider/base/IcebergCleanupJobBaseSQLProvider.java
index 0de925bdf6..11b10876c5 100644
---
a/iceberg/iceberg-rest-server/src/main/java/org/apache/gravitino/iceberg/service/cleanup/mapper/provider/base/IcebergCleanupJobBaseSQLProvider.java
+++
b/iceberg/iceberg-rest-server/src/main/java/org/apache/gravitino/iceberg/service/cleanup/mapper/provider/base/IcebergCleanupJobBaseSQLProvider.java
@@ -93,18 +93,22 @@ public class IcebergCleanupJobBaseSQLProvider {
* @param state terminal state to set (SUCCEEDED or FAILED)
* @param reason failure reason, or {@code null} for success
* @param now state-change timestamp
+ * @param heartbeat the caller's owned heartbeat value (compare-and-swap
ownership key)
* @return the terminal-transition UPDATE statement
*/
public String markFinished(
@Param("id") long id,
@Param("state") String state,
@Param("reason") String reason,
- @Param("now") long now) {
- // Shared transition to a final state: SUCCEEDED (reason null) or FAILED
(reason set).
+ @Param("now") long now,
+ @Param("heartbeat") long heartbeat) {
+ // Shared transition to a final state: SUCCEEDED (reason null) or FAILED
(reason set). The
+ // heartbeat_at predicate checks ownership: a reclaimed worker no longer
matches it.
return "UPDATE "
+ TABLE_NAME
+ " SET state = #{state}, last_error = #{reason}, heartbeat_at = 0,"
- + " updated_at = #{now} WHERE id = #{id} AND state = 'RUNNING'";
+ + " updated_at = #{now}"
+ + " WHERE id = #{id} AND state = 'RUNNING' AND heartbeat_at =
#{heartbeat}";
}
/**
@@ -112,24 +116,28 @@ public class IcebergCleanupJobBaseSQLProvider {
* @param reason failure reason
* @param maxAttempts attempt ceiling past which the job is FAILED
* @param now state-change timestamp
+ * @param heartbeat the caller's owned heartbeat value (compare-and-swap
ownership key)
* @return the transient-failure UPDATE statement
*/
public String recordFailure(
@Param("id") long id,
@Param("reason") String reason,
@Param("maxAttempts") int maxAttempts,
- @Param("now") long now) {
+ @Param("now") long now,
+ @Param("heartbeat") long heartbeat) {
// The state CASE must observe the pre-increment attempts value, so it
MUST precede
// "attempts = attempts + 1" in the SET list. MySQL evaluates SET
assignments left to
// right and later clauses see already-updated columns; if the increment
came first the
// CASE would compare attempts + 2 and fail one attempt early. H2 and
PostgreSQL evaluate
// every right-hand side against the original row, so state-first is
correct there too,
- // keeping this statement portable across all backends.
+ // keeping this statement portable across all backends. The heartbeat_at
predicate checks
+ // ownership (see markFinished).
return "UPDATE "
+ TABLE_NAME
+ " SET state = CASE WHEN attempts + 1 >= #{maxAttempts} THEN 'FAILED'
ELSE 'PENDING' END,"
+ " attempts = attempts + 1, last_error = #{reason},"
- + " heartbeat_at = 0, updated_at = #{now} WHERE id = #{id} AND state =
'RUNNING'";
+ + " heartbeat_at = 0, updated_at = #{now}"
+ + " WHERE id = #{id} AND state = 'RUNNING' AND heartbeat_at =
#{heartbeat}";
}
/**
diff --git
a/iceberg/iceberg-rest-server/src/test/java/org/apache/gravitino/iceberg/service/cleanup/AbstractIcebergCleanupJobStoreBackendTest.java
b/iceberg/iceberg-rest-server/src/test/java/org/apache/gravitino/iceberg/service/cleanup/AbstractIcebergCleanupJobStoreBackendTest.java
index 3e58e925df..005653da64 100644
---
a/iceberg/iceberg-rest-server/src/test/java/org/apache/gravitino/iceberg/service/cleanup/AbstractIcebergCleanupJobStoreBackendTest.java
+++
b/iceberg/iceberg-rest-server/src/test/java/org/apache/gravitino/iceberg/service/cleanup/AbstractIcebergCleanupJobStoreBackendTest.java
@@ -100,36 +100,56 @@ abstract class AbstractIcebergCleanupJobStoreBackendTest
extends TestJDBCBackend
Assertions.assertTrue(store.findUnfinishedJobId(CATALOG_ID, "db",
"t").isPresent());
Assertions.assertFalse(store.takePendingJob(now, 300_000L,
10).isPresent());
- Assertions.assertTrue(store.markSucceeded(id));
+ Assertions.assertTrue(store.markSucceeded(id, now));
Assertions.assertEquals(IcebergCleanupJob.State.SUCCEEDED,
store.stateOf(id));
// A second transition no longer owns the (now terminal) row, so it
reports no update.
- Assertions.assertFalse(store.markSucceeded(id));
+ Assertions.assertFalse(store.markSucceeded(id, now));
Assertions.assertFalse(store.findUnfinishedJobId(CATALOG_ID, "db",
"t").isPresent());
Assertions.assertEquals(
1, store.deleteFinishedJobsByLegacyTimeline(System.currentTimeMillis()
+ 1));
}
@TestTemplate
- void testMarkFailed() {
+ void testTransientFailureRetriesThenFailsAtCeiling() {
long id = store.addJob(sampleJob());
- store.takePendingJob(System.currentTimeMillis(), 300_000L, 10);
- Assertions.assertTrue(store.markFailed(id, "corrupt metadata"));
+ for (int i = 0; i < 2; i++) {
+ long heartbeat = System.currentTimeMillis();
+ store.takePendingJob(heartbeat, 300_000L, 10);
+ Assertions.assertTrue(store.recordFailure(id, "boom " + i, 3,
heartbeat));
+ Assertions.assertEquals(IcebergCleanupJob.State.PENDING,
store.stateOf(id));
+ }
+ long heartbeat = System.currentTimeMillis();
+ store.takePendingJob(heartbeat, 300_000L, 10);
+ Assertions.assertTrue(store.recordFailure(id, "boom final", 3, heartbeat));
Assertions.assertEquals(IcebergCleanupJob.State.FAILED, store.stateOf(id));
}
@TestTemplate
- void testTransientFailureRetriesThenFailsAtCeiling() {
+ void testRecordFailureAtMaxAttemptsMarksFailed() {
long id = store.addJob(sampleJob());
- for (int i = 0; i < 2; i++) {
- store.takePendingJob(System.currentTimeMillis(), 300_000L, 10);
- Assertions.assertTrue(store.recordFailure(id, "boom " + i, 3));
- Assertions.assertEquals(IcebergCleanupJob.State.PENDING,
store.stateOf(id));
- }
- store.takePendingJob(System.currentTimeMillis(), 300_000L, 10);
- Assertions.assertTrue(store.recordFailure(id, "boom final", 3));
+ long heartbeat = System.currentTimeMillis();
+ store.takePendingJob(heartbeat, 300_000L, 10);
+
+ Assertions.assertTrue(store.recordFailure(id, "boom", 1, heartbeat));
Assertions.assertEquals(IcebergCleanupJob.State.FAILED, store.stateOf(id));
}
+ @TestTemplate
+ void testTerminalUpdateNeedsOwnership() {
+ long id = store.addJob(sampleJob());
+ long now = System.currentTimeMillis();
+ store.takePendingJob(now, 300_000L, 10); // writes heartbeat_at = now
+
+ // A stale heartbeat token (a reclaimed worker) cannot finish or fail the
job.
+ Assertions.assertFalse(store.markSucceeded(id, now - 1));
+ Assertions.assertFalse(store.recordFailure(id, "stale", 3, now - 1));
+ Assertions.assertEquals(IcebergCleanupJob.State.RUNNING,
store.stateOf(id));
+
+ // The owner with the current token wins.
+ Assertions.assertTrue(store.markSucceeded(id, now));
+ Assertions.assertEquals(IcebergCleanupJob.State.SUCCEEDED,
store.stateOf(id));
+ }
+
@TestTemplate
void testHeartbeatCasAndStaleTakeover() {
long id = store.addJob(sampleJob());
diff --git
a/iceberg/iceberg-rest-server/src/test/java/org/apache/gravitino/iceberg/service/cleanup/TestIcebergCleanupManager.java
b/iceberg/iceberg-rest-server/src/test/java/org/apache/gravitino/iceberg/service/cleanup/TestIcebergCleanupManager.java
new file mode 100644
index 0000000000..a525dd3fa6
--- /dev/null
+++
b/iceberg/iceberg-rest-server/src/test/java/org/apache/gravitino/iceberg/service/cleanup/TestIcebergCleanupManager.java
@@ -0,0 +1,406 @@
+/*
+ * 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.ArrayList;
+import java.util.Arrays;
+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 java.util.concurrent.atomic.AtomicInteger;
+import java.util.concurrent.atomic.AtomicLong;
+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.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.catalog.Namespace;
+import org.apache.iceberg.catalog.TableIdentifier;
+import org.apache.iceberg.exceptions.NotFoundException;
+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;
+
+/**
+ * Exercises {@link IcebergCleanupManager} against the same relational backend
matrix as the cleanup
+ * store: {@link TestJDBCBackend} initializes H2 by default, adds MySQL and
PostgreSQL when {@code
+ * dockerTest=true}, and truncates all backend tables before each invocation.
+ */
+class TestIcebergCleanupManager extends TestJDBCBackend {
+
+ private static final long CATALOG_ID = 100L;
+ private static final String DATA_FILE = "memory://db/t/data/00000-0.parquet";
+
+ private IcebergCleanupJobStore store;
+
+ // TestJDBCBackend's BackendTestExtension overwrites GravitinoEnv's
singleton "config" and
+ // "idGenerator" fields with a backend-only Mockito mock and never restores
them. Because the
+ // whole iceberg-rest-server module runs in one JVM, that mock would leak
into later test classes
+ // (e.g. credential vending), where
MetadataAuthzHelper.enableAuthorization() unboxes the
+ // unstubbed config.get(ENABLE_AUTHORIZATION) -> null and NPEs. Snapshot the
pre-test fields and
+ // restore them after this class so it leaves GravitinoEnv exactly as it
found it.
+ 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 prepareCleanupJobStore() {
+ store = new IcebergCleanupJobStore(new RandomIdGenerator());
+ }
+
+ private static IcebergConfig fastPollConfig() {
+ Map<String, String> config = new HashMap<>();
+ config.put("async-cleanup.worker-threads", "1");
+ config.put("async-cleanup.poll-interval-secs", "1");
+ return new IcebergConfig(config);
+ }
+
+ private static IcebergCleanupJob sampleJob() {
+ return new IcebergCleanupJob(
+ 0L,
+ CATALOG_ID,
+ "db",
+ "t",
+ "s3://b/db/t/metadata/0.json",
+ NoopFileIO.class.getName(),
+ ImmutableMap.of(),
+ "alice");
+ }
+
+ // Builds db.t with one appended data file (so it has a manifest list, a
manifest, and a data
+ // file) and materializes the data file in the in-memory FileIO so deletions
are observable.
+ private static BaseTable tableWithDataFile() {
+ InMemoryCatalog catalog = new InMemoryCatalog();
+ catalog.initialize("test", ImmutableMap.of());
+ catalog.createNamespace(Namespace.of("db"));
+ TableIdentifier id = TableIdentifier.of(Namespace.of("db"), "t");
+ Schema schema = new Schema(Types.NestedField.required(1, "id",
Types.IntegerType.get()));
+ Table table = catalog.createTable(id, 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(id);
+ ((InMemoryFileIO) base.io()).addFile(DATA_FILE, new byte[] {1});
+ return base;
+ }
+
+ @TestTemplate
+ void testDeleteAllBatches() {
+ CopyOnWriteArrayList<String> deleted = new CopyOnWriteArrayList<>();
+ IcebergCleanupManager svc =
+ new IcebergCleanupManager(store, new IcebergConfig(new HashMap<>()));
+ try {
+ svc.deleteAll(new RecordingFileIO(deleted), Arrays.asList("a", "b", "c",
"d", "e", "f", "g"));
+ Assertions.assertEquals(7, deleted.size());
+ } finally {
+ svc.close();
+ }
+ }
+
+ @TestTemplate
+ void testDeleteAllIgnoresMissing() {
+ IcebergCleanupManager svc =
+ new IcebergCleanupManager(store, new IcebergConfig(new HashMap<>()));
+ try {
+ Assertions.assertDoesNotThrow(
+ () -> svc.deleteAll(new MissingFileIO(),
Arrays.asList("already-gone")));
+ } finally {
+ svc.close();
+ }
+ }
+
+ @TestTemplate
+ void testDeleteAllIgnoresMissingBulk() {
+ IcebergCleanupManager svc =
+ new IcebergCleanupManager(store, new IcebergConfig(new HashMap<>()));
+ try {
+ Assertions.assertDoesNotThrow(
+ () -> svc.deleteAll(new MissingBulkFileIO(),
Arrays.asList("already-gone")));
+ } finally {
+ svc.close();
+ }
+ }
+
+ @TestTemplate
+ void testCleanupDeletesAllFiles() {
+ BaseTable base = tableWithDataFile();
+ FileIO io = base.io();
+ String metadataLocation =
base.operations().current().metadataFileLocation();
+
+ // Capture reachable files before cleanup; the manifests cannot be read
once deleted.
+ List<String> expected = new ArrayList<>();
+ expected.add(metadataLocation);
+ expected.add(DATA_FILE);
+ for (Snapshot snapshot : base.snapshots()) {
+ expected.add(snapshot.manifestListLocation());
+ for (ManifestFile manifest : snapshot.allManifests(io)) {
+ expected.add(manifest.path());
+ }
+ }
+ expected.forEach(file -> Assertions.assertTrue(((InMemoryFileIO)
io).fileExists(file), file));
+
+ IcebergCleanupManager svc =
+ new IcebergCleanupManager(store, new IcebergConfig(new HashMap<>()));
+ try {
+ svc.cleanupFiles(io, metadataLocation);
+ } finally {
+ svc.close();
+ }
+
+ expected.forEach(file -> Assertions.assertFalse(((InMemoryFileIO)
io).fileExists(file), file));
+ }
+
+ @TestTemplate
+ void testCleanupToleratesMissingMetadata() {
+ // A missing root metadata.json means the table is already gone, so
cleanup just returns.
+ IcebergCleanupManager svc =
+ new IcebergCleanupManager(store, new IcebergConfig(new HashMap<>()));
+ try {
+ Assertions.assertDoesNotThrow(
+ () -> svc.cleanupFiles(new InMemoryFileIO(),
"memory://db/t/metadata/missing.json"));
+ } finally {
+ svc.close();
+ }
+ }
+
+ @TestTemplate
+ void testWorkerSucceeds() {
+ AtomicInteger calls = new AtomicInteger();
+ IcebergCleanupManager svc =
+ new IcebergCleanupManager(store, fastPollConfig()) {
+ @Override
+ void cleanupFiles(FileIO io, String metadataLocation) {
+ calls.incrementAndGet();
+ }
+ };
+ long id = store.addJob(sampleJob());
+ svc.start();
+ try {
+ Awaitility.await()
+ .atMost(5, TimeUnit.SECONDS)
+ .until(() -> store.stateOf(id) == IcebergCleanupJob.State.SUCCEEDED);
+ Assertions.assertEquals(1, calls.get());
+ } finally {
+ svc.close();
+ }
+ }
+
+ @TestTemplate
+ void testWorkerRetriesThenFails() {
+ 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", "3");
+ IcebergCleanupManager svc =
+ new IcebergCleanupManager(store, new IcebergConfig(config)) {
+ @Override
+ void cleanupFiles(FileIO io, String metadataLocation) {
+ throw new RuntimeException("transient");
+ }
+ };
+ long id = store.addJob(sampleJob());
+ svc.start();
+ try {
+ Awaitility.await()
+ .atMost(15, TimeUnit.SECONDS)
+ .until(() -> store.stateOf(id) == IcebergCleanupJob.State.FAILED);
+ } finally {
+ svc.close();
+ }
+ }
+
+ @TestTemplate
+ void testStartAfterCloseFailsFast() {
+ IcebergCleanupManager svc =
+ new IcebergCleanupManager(store, new IcebergConfig(new HashMap<>()));
+ svc.close();
+
+ Assertions.assertThrows(IllegalStateException.class, svc::start);
+ }
+
+ @TestTemplate
+ @SuppressWarnings("unchecked")
+ void testRefreshHeartbeatsPublishesNewTokenBeforeStoreUpdate() throws
Exception {
+ BlockingHeartbeatStore blockingStore = new BlockingHeartbeatStore();
+ IcebergCleanupManager svc =
+ new IcebergCleanupManager(blockingStore, new IcebergConfig(new
HashMap<>()));
+ try {
+ Map<Long, Long> heartbeats =
+ (Map<Long, Long>) FieldUtils.readField(svc, "ownedHeartbeats", true);
+ heartbeats.put(1L, 100L);
+
+ Thread refresh = new Thread(svc::refreshHeartbeats);
+ refresh.start();
+ Assertions.assertTrue(blockingStore.heartbeatStarted.await(5,
TimeUnit.SECONDS));
+
+ AtomicLong token = new AtomicLong();
+ svc.finishJob(1L, heartbeat -> token.compareAndSet(0L, heartbeat));
+ Assertions.assertTrue(token.get() > 100L);
+
+ blockingStore.releaseHeartbeat.countDown();
+ refresh.join(5_000L);
+ Assertions.assertFalse(refresh.isAlive());
+ } finally {
+ svc.close();
+ }
+ }
+
+ @TestTemplate
+ void testIsNameOccupied() {
+ IcebergCleanupManager svc =
+ new IcebergCleanupManager(store, new IcebergConfig(new HashMap<>()));
+ try {
+ Assertions.assertFalse(svc.isNameOccupied(CATALOG_ID, "db", "t"));
+ svc.addJob(sampleJob());
+ Assertions.assertTrue(svc.isNameOccupied(CATALOG_ID, "db", "t"));
+ } finally {
+ svc.close();
+ }
+ }
+
+ static class RecordingFileIO implements SupportsBulkOperations {
+ private final CopyOnWriteArrayList<String> deleted;
+
+ RecordingFileIO(CopyOnWriteArrayList<String> deleted) {
+ this.deleted = deleted;
+ }
+
+ @Override
+ public InputFile newInputFile(String path) {
+ throw new UnsupportedOperationException();
+ }
+
+ @Override
+ public OutputFile newOutputFile(String path) {
+ throw new UnsupportedOperationException();
+ }
+
+ @Override
+ public void deleteFile(String path) {
+ deleted.add(path);
+ }
+
+ @Override
+ public void deleteFiles(Iterable<String> paths) {
+ for (String path : paths) {
+ deleted.add(path);
+ }
+ }
+ }
+
+ private static class MissingBulkFileIO extends MissingFileIO implements
SupportsBulkOperations {
+ @Override
+ public void deleteFiles(Iterable<String> paths) {
+ throw new NotFoundException("Missing files");
+ }
+ }
+
+ public static class NoopFileIO implements FileIO {
+ @Override
+ public InputFile newInputFile(String path) {
+ throw new UnsupportedOperationException();
+ }
+
+ @Override
+ public OutputFile newOutputFile(String path) {
+ throw new UnsupportedOperationException();
+ }
+
+ @Override
+ public void deleteFile(String path) {
+ throw new UnsupportedOperationException();
+ }
+ }
+
+ private static class MissingFileIO implements FileIO {
+ @Override
+ public InputFile newInputFile(String path) {
+ throw new UnsupportedOperationException();
+ }
+
+ @Override
+ public OutputFile newOutputFile(String path) {
+ throw new UnsupportedOperationException();
+ }
+
+ @Override
+ public void deleteFile(String path) {
+ throw new NotFoundException("Missing file: %s", path);
+ }
+ }
+
+ private static class BlockingHeartbeatStore extends IcebergCleanupJobStore {
+ private final CountDownLatch heartbeatStarted = new CountDownLatch(1);
+ private final CountDownLatch releaseHeartbeat = new CountDownLatch(1);
+
+ BlockingHeartbeatStore() {
+ super(new RandomIdGenerator());
+ }
+
+ @Override
+ public boolean heartbeat(long id, long lastHeartbeat, long now) {
+ heartbeatStarted.countDown();
+ try {
+ Assertions.assertTrue(releaseHeartbeat.await(5, TimeUnit.SECONDS));
+ } catch (InterruptedException e) {
+ Thread.currentThread().interrupt();
+ throw new RuntimeException(e);
+ }
+ return true;
+ }
+ }
+}
