roryqi commented on code in PR #11299: URL: https://github.com/apache/gravitino/pull/11299#discussion_r3331669488
########## design-docs/async-iceberg-rest-hard-deletion.md: ########## @@ -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.enqueue( + new IcebergCleanupJob( + 0L, // id assigned from IdGenerator at enqueue (§5.4) Review Comment: Removed the `(§5.4)` flags; the inline comments now describe the fields directly (`id assigned by IdGenerator at enqueue`, `resolved catalog entity id`). ########## iceberg/iceberg-rest-server/src/main/java/org/apache/gravitino/iceberg/service/cleanup/IcebergCleanupManager.java: ########## @@ -0,0 +1,319 @@ +/* + * 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.Iterators; +import java.util.ArrayList; +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 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; + private final Map<Long, Long> ownedHeartbeats = new ConcurrentHashMap<>(); + + private volatile boolean running; + 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; + 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()); + } + + /** + * Enqueues a cleanup job. + * + * @param job job to persist + * @return generated id + */ + public long enqueue(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 (running) { + return; + } + + running = true; + workers = Executors.newFixedThreadPool(workerThreads, daemon("iceberg-cleanup-worker")); + scheduler = Executors.newScheduledThreadPool(1, daemon("iceberg-cleanup-scheduler")); Review Comment: Done. Replaced the implicit unbounded queue with an explicit `ThreadPoolExecutor` backed by a bounded `ArrayBlockingQueue`. We submit exactly `workerThreads` loops, so it's bounded only to avoid `newFixedThreadPool`'s unbounded `LinkedBlockingQueue`. ########## iceberg/iceberg-rest-server/src/main/java/org/apache/gravitino/iceberg/service/cleanup/IcebergCleanupManager.java: ########## @@ -0,0 +1,356 @@ +/* + * 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.Iterators; +import java.util.ArrayList; +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 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; + private final Map<Long, Long> ownedHeartbeats = new ConcurrentHashMap<>(); + + private volatile boolean running; + 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; + 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 (running) { + return; + } + + running = true; + // Exactly workerThreads long-running loops are submitted below, so the bounded queue is never + // actually used; it is declared explicitly to avoid Executors.newFixedThreadPool's unbounded + // LinkedBlockingQueue. + workers = + new ThreadPoolExecutor( + workerThreads, + workerThreads, + 0L, + TimeUnit.MILLISECONDS, + new ArrayBlockingQueue<>(workerThreads), + daemon("iceberg-cleanup-worker")); + scheduler = Executors.newScheduledThreadPool(1, daemon("iceberg-cleanup-scheduler")); + for (int i = 0; i < workerThreads; i++) { + workers.submit(this::workerLoop); + } + + long heartbeatIntervalMs = Math.max(1L, heartbeatTimeoutMs / 3L); + scheduler.scheduleAtFixedRate( + this::refreshHeartbeats, heartbeatIntervalMs, heartbeatIntervalMs, TimeUnit.MILLISECONDS); + scheduler.scheduleAtFixedRate(this::prune, 1L, 1L, TimeUnit.HOURS); + } + + @Override + public void close() { + running = false; + if (scheduler != null) { + scheduler.shutdownNow(); + } + if (workers != null) { + workers.shutdownNow(); + try { + workers.awaitTermination(5, TimeUnit.SECONDS); + } catch (InterruptedException e) { + Thread.currentThread().interrupt(); + } + } + deleteExecutor.shutdownNow(); + // 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) { + // The root metadata.json pointer is the ONLY file whose absence means "the table is already + // gone", so this is the single place a NotFoundException is treated as success: a prior + // attempt finished deleting the table before its row was marked, or the table was never + // fully written. Downstream, enumeration (reachableFiles) and bulk deletion (deleteAll) + // tolerate already-deleted files locally instead of letting a NotFoundException escape, so + // it never reaches here from anywhere but this read. There is nothing left to clean up; + // return so runJob marks the job SUCCEEDED. + LOG.info( + "Cleanup metadata {} already absent; treating as completed", + metadataLocation, + metadataAlreadyGone); + return; + } + deleteAll(io, reachableFiles(io, metadata)); + } + + void deleteAll(FileIO io, Iterable<String> files) { + 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 (hasCause(e, NotFoundException.class)) { + LOG.debug("Ignoring already-deleted file during async cleanup", e); + continue; + } + throw new RuntimeException("Bulk delete batch failed", e); + } + } + } + + private static boolean hasCause(Throwable throwable, Class<? extends Throwable> type) { + Throwable current = throwable; + while (current != null) { + if (type.isInstance(current)) { + return true; + } + current = current.getCause(); + } + return false; + } + + private void workerLoop() { + while (running) { + try { + long now = System.currentTimeMillis(); + Optional<IcebergCleanupJob> job = + store.takePendingJob(now, heartbeatTimeoutMs, candidateWindow); + if (!job.isPresent()) { + sleep(pollIntervalMs); + continue; + } + + ownedHeartbeats.put(job.get().id(), now); + runJob(job.get()); + } catch (Throwable t) { + // A worker task is submitted exactly once in start(), so if this loop ever exits the + // thread is gone for good and the pool shrinks permanently. Catch every Throwable -- + // including Errors and anything takePendingJob/runJob may surface -- so a fault only backs + // the loop off instead of killing the worker. Re-assert the interrupt flag so shutdown + // still unwinds promptly. + 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) { + try { + FileIO io = CatalogUtil.loadFileIO(job.fileIOImpl(), job.fileIOProperties(), null); + // cleanupFiles swallows the only legitimate already-gone signal (a missing metadata.json) + // internally, so reaching this line means the table's files were enumerated and deleted. + cleanupFiles(io, job.metadataLocation()); + store.markSucceeded(job.id()); + } catch (RuntimeException e) { + LOG.warn("Cleanup job {} failed transiently; will retry", job.id(), e); + store.recordFailure(job.id(), e.getMessage(), maxAttempts); + } finally { + ownedHeartbeats.remove(job.id()); + } + } + + private void refreshHeartbeats() { + long now = System.currentTimeMillis(); + List<Map.Entry<Long, Long>> heartbeats = new ArrayList<>(ownedHeartbeats.entrySet()); + for (Map.Entry<Long, Long> entry : heartbeats) { + try { + if (store.heartbeat(entry.getKey(), entry.getValue(), now)) { + ownedHeartbeats.put(entry.getKey(), now); + } else { + LOG.warn("Lost ownership of cleanup job {}", entry.getKey()); + ownedHeartbeats.remove(entry.getKey()); + } + } catch (Throwable t) { + // scheduleAtFixedRate suppresses all future runs if a task ever throws, which would stop + // heartbeat renewal for the whole process. Swallow per-job faults (including Errors) so one + // bad job neither halts this pass nor the recurring task. + LOG.warn("Heartbeat update failed for job {}", entry.getKey(), t); + } + } + } + + private void prune() { + try { + store.deleteFinishedJobsByLegacyTimeline(System.currentTimeMillis() - retentionMs); + } catch (Throwable t) { + // As above: never let a Throwable escape a scheduleAtFixedRate task, or pruning silently + // stops for the life of the process and finished rows accumulate without bound. + LOG.warn("Cleanup-row pruning failed", t); + } + } + + private Iterable<String> reachableFiles(FileIO io, TableMetadata metadata) { + Table table = new BaseTable(new StaticTableOperations(metadata, io), "async-cleanup"); + Set<String> files = new LinkedHashSet<>(); + files.addAll(ReachableFileUtil.metadataFileLocations(table, true)); + files.addAll(ReachableFileUtil.manifestListLocations(table)); + files.addAll(ReachableFileUtil.statisticsFilesLocations(table)); + + for (Snapshot snapshot : metadata.snapshots()) { + for (ManifestFile manifest : snapshot.allManifests(io)) { + files.add(manifest.path()); + try (CloseableIterable<String> paths = + ManifestFiles.readPaths(manifest, io, metadata.specsById())) { + for (String path : paths) { + files.add(path); + } + } catch (NotFoundException alreadyDeleted) { + // A concurrent worker (e.g. one that reclaimed this job after a heartbeat timeout) may + // already have deleted this manifest; its data files are gone with it. Skip it and keep + // enumerating the rest rather than surfacing the NotFoundException -- only the root + // metadata.json read in cleanupFiles treats a NotFoundException as "table already gone". + // This mirrors deleteAll tolerating already-deleted files and keeps cleanup idempotent + // under double processing. + LOG.debug( + "Manifest {} already deleted during async cleanup; skipping", + manifest.path(), + alreadyDeleted); + } catch (Exception e) { + throw new RuntimeException("Failed to read manifest " + manifest.path(), e); + } + } + } + return files; + } Review Comment: Fixed. Data files are now streamed one manifest at a time (`deleteDataFiles`): each manifest's `ManifestFiles.readPaths` iterable is handed straight to `deleteAll`, which partitions it lazily, so only one manifest's paths are resident at once. Only the much smaller manifest/manifest-list/metadata path lists are materialized. ########## iceberg/iceberg-rest-server/src/main/java/org/apache/gravitino/iceberg/service/cleanup/IcebergCleanupManager.java: ########## @@ -0,0 +1,356 @@ +/* + * 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.Iterators; +import java.util.ArrayList; +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 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; + private final Map<Long, Long> ownedHeartbeats = new ConcurrentHashMap<>(); + + private volatile boolean running; + 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; + 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 (running) { + return; + } + + running = true; + // Exactly workerThreads long-running loops are submitted below, so the bounded queue is never + // actually used; it is declared explicitly to avoid Executors.newFixedThreadPool's unbounded + // LinkedBlockingQueue. + workers = + new ThreadPoolExecutor( + workerThreads, + workerThreads, + 0L, + TimeUnit.MILLISECONDS, + new ArrayBlockingQueue<>(workerThreads), + daemon("iceberg-cleanup-worker")); + scheduler = Executors.newScheduledThreadPool(1, daemon("iceberg-cleanup-scheduler")); + for (int i = 0; i < workerThreads; i++) { + workers.submit(this::workerLoop); + } + + long heartbeatIntervalMs = Math.max(1L, heartbeatTimeoutMs / 3L); + scheduler.scheduleAtFixedRate( + this::refreshHeartbeats, heartbeatIntervalMs, heartbeatIntervalMs, TimeUnit.MILLISECONDS); + scheduler.scheduleAtFixedRate(this::prune, 1L, 1L, TimeUnit.HOURS); + } + + @Override + public void close() { + running = false; + if (scheduler != null) { + scheduler.shutdownNow(); + } + if (workers != null) { + workers.shutdownNow(); + try { + workers.awaitTermination(5, TimeUnit.SECONDS); + } catch (InterruptedException e) { + Thread.currentThread().interrupt(); + } + } + deleteExecutor.shutdownNow(); + // 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) { + // The root metadata.json pointer is the ONLY file whose absence means "the table is already + // gone", so this is the single place a NotFoundException is treated as success: a prior + // attempt finished deleting the table before its row was marked, or the table was never + // fully written. Downstream, enumeration (reachableFiles) and bulk deletion (deleteAll) + // tolerate already-deleted files locally instead of letting a NotFoundException escape, so + // it never reaches here from anywhere but this read. There is nothing left to clean up; + // return so runJob marks the job SUCCEEDED. + LOG.info( + "Cleanup metadata {} already absent; treating as completed", + metadataLocation, + metadataAlreadyGone); + return; + } + deleteAll(io, reachableFiles(io, metadata)); + } + + void deleteAll(FileIO io, Iterable<String> files) { + 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 (hasCause(e, NotFoundException.class)) { + LOG.debug("Ignoring already-deleted file during async cleanup", e); + continue; + } + throw new RuntimeException("Bulk delete batch failed", e); + } + } + } + + private static boolean hasCause(Throwable throwable, Class<? extends Throwable> type) { + Throwable current = throwable; + while (current != null) { + if (type.isInstance(current)) { + return true; + } + current = current.getCause(); + } + return false; + } + + private void workerLoop() { + while (running) { + try { + long now = System.currentTimeMillis(); + Optional<IcebergCleanupJob> job = + store.takePendingJob(now, heartbeatTimeoutMs, candidateWindow); + if (!job.isPresent()) { + sleep(pollIntervalMs); + continue; + } + + ownedHeartbeats.put(job.get().id(), now); + runJob(job.get()); + } catch (Throwable t) { + // A worker task is submitted exactly once in start(), so if this loop ever exits the + // thread is gone for good and the pool shrinks permanently. Catch every Throwable -- + // including Errors and anything takePendingJob/runJob may surface -- so a fault only backs + // the loop off instead of killing the worker. Re-assert the interrupt flag so shutdown + // still unwinds promptly. + 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) { + try { + FileIO io = CatalogUtil.loadFileIO(job.fileIOImpl(), job.fileIOProperties(), null); + // cleanupFiles swallows the only legitimate already-gone signal (a missing metadata.json) + // internally, so reaching this line means the table's files were enumerated and deleted. + cleanupFiles(io, job.metadataLocation()); + store.markSucceeded(job.id()); + } catch (RuntimeException e) { + LOG.warn("Cleanup job {} failed transiently; will retry", job.id(), e); + store.recordFailure(job.id(), e.getMessage(), maxAttempts); + } finally { + ownedHeartbeats.remove(job.id()); + } + } + + private void refreshHeartbeats() { + long now = System.currentTimeMillis(); + List<Map.Entry<Long, Long>> heartbeats = new ArrayList<>(ownedHeartbeats.entrySet()); + for (Map.Entry<Long, Long> entry : heartbeats) { + try { + if (store.heartbeat(entry.getKey(), entry.getValue(), now)) { + ownedHeartbeats.put(entry.getKey(), now); + } else { + LOG.warn("Lost ownership of cleanup job {}", entry.getKey()); + ownedHeartbeats.remove(entry.getKey()); + } + } catch (Throwable t) { + // scheduleAtFixedRate suppresses all future runs if a task ever throws, which would stop + // heartbeat renewal for the whole process. Swallow per-job faults (including Errors) so one + // bad job neither halts this pass nor the recurring task. + LOG.warn("Heartbeat update failed for job {}", entry.getKey(), t); + } + } + } Review Comment: Addressed. The worker now checks ownership before each manifest and between deletion levels (`requireOwnership`) and stops early once the lease is lost. I also added a heartbeat-token CAS to `markSucceeded`/`recordFailure` (`WHERE ... AND heartbeat_at = ?`) so a reclaimed worker can't change a job a peer now owns. ########## iceberg/iceberg-rest-server/src/main/java/org/apache/gravitino/iceberg/service/cleanup/IcebergCleanupManager.java: ########## @@ -0,0 +1,356 @@ +/* + * 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.Iterators; +import java.util.ArrayList; +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 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; + private final Map<Long, Long> ownedHeartbeats = new ConcurrentHashMap<>(); + + private volatile boolean running; + 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; + 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 (running) { + return; + } + + running = true; + // Exactly workerThreads long-running loops are submitted below, so the bounded queue is never + // actually used; it is declared explicitly to avoid Executors.newFixedThreadPool's unbounded + // LinkedBlockingQueue. + workers = + new ThreadPoolExecutor( + workerThreads, + workerThreads, + 0L, + TimeUnit.MILLISECONDS, + new ArrayBlockingQueue<>(workerThreads), + daemon("iceberg-cleanup-worker")); + scheduler = Executors.newScheduledThreadPool(1, daemon("iceberg-cleanup-scheduler")); + for (int i = 0; i < workerThreads; i++) { + workers.submit(this::workerLoop); + } + + long heartbeatIntervalMs = Math.max(1L, heartbeatTimeoutMs / 3L); + scheduler.scheduleAtFixedRate( + this::refreshHeartbeats, heartbeatIntervalMs, heartbeatIntervalMs, TimeUnit.MILLISECONDS); + scheduler.scheduleAtFixedRate(this::prune, 1L, 1L, TimeUnit.HOURS); + } Review Comment: Done. `running` is now an `AtomicBoolean` and `start()` guards with `compareAndSet(false, true)`. ########## design-docs/async-iceberg-rest-hard-deletion.md: ########## @@ -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, + `id` BIGINT(20) UNSIGNED NOT NULL, + `catalog_id` BIGINT(20) UNSIGNED NOT NULL COMMENT 'globally unique id of the owning catalog, stable across catalog rename', `namespace` VARCHAR(512) NOT NULL, `table_name` VARCHAR(256) NOT NULL, - `metadata_location` VARCHAR(1024) NOT NULL, + `metadata_location` MEDIUMTEXT NOT NULL, `file_io_impl` VARCHAR(256) NOT NULL, `file_io_props` MEDIUMTEXT NOT NULL COMMENT 'JSON', `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', + `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'; Review Comment: Aligned the doc's CREATE TABLE with scripts/mysql/schema-1.3.0-mysql.sql -- added the per-column COMMENTs so they match. ########## iceberg/iceberg-rest-server/src/main/java/org/apache/gravitino/iceberg/service/cleanup/IcebergCleanupManager.java: ########## @@ -0,0 +1,356 @@ +/* + * 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.Iterators; +import java.util.ArrayList; +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 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; + private final Map<Long, Long> ownedHeartbeats = new ConcurrentHashMap<>(); + + private volatile boolean running; + 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; + 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 (running) { + return; + } + + running = true; + // Exactly workerThreads long-running loops are submitted below, so the bounded queue is never + // actually used; it is declared explicitly to avoid Executors.newFixedThreadPool's unbounded + // LinkedBlockingQueue. + workers = + new ThreadPoolExecutor( + workerThreads, + workerThreads, + 0L, + TimeUnit.MILLISECONDS, + new ArrayBlockingQueue<>(workerThreads), + daemon("iceberg-cleanup-worker")); + scheduler = Executors.newScheduledThreadPool(1, daemon("iceberg-cleanup-scheduler")); Review Comment: Renamed to `iceberg-cleanup-heartbeat-prune`, with a comment noting the single scheduler thread runs both heartbeat renewal and pruning. ########## iceberg/iceberg-rest-server/src/main/java/org/apache/gravitino/iceberg/service/cleanup/IcebergCleanupManager.java: ########## @@ -0,0 +1,356 @@ +/* + * 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.Iterators; +import java.util.ArrayList; +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 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; + private final Map<Long, Long> ownedHeartbeats = new ConcurrentHashMap<>(); + + private volatile boolean running; + 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; + 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 (running) { + return; + } + + running = true; + // Exactly workerThreads long-running loops are submitted below, so the bounded queue is never + // actually used; it is declared explicitly to avoid Executors.newFixedThreadPool's unbounded + // LinkedBlockingQueue. + workers = + new ThreadPoolExecutor( + workerThreads, + workerThreads, + 0L, + TimeUnit.MILLISECONDS, + new ArrayBlockingQueue<>(workerThreads), + daemon("iceberg-cleanup-worker")); + scheduler = Executors.newScheduledThreadPool(1, daemon("iceberg-cleanup-scheduler")); + for (int i = 0; i < workerThreads; i++) { + workers.submit(this::workerLoop); + } Review Comment: Moved the worker-submission loop directly under the `workers = new ThreadPoolExecutor(...)` definition, before the scheduler setup. -- 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]
