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
commit 116c974ad0c53f6eae38a1c7ca08ccbfaeac1ddc Author: remilu <[email protected]> AuthorDate: Mon Oct 26 07:55:09 2026 +0800 [#14020] feat(tms): Cron-pool submit pipeline and scheduler pool wiring Fixes #14020. Squashed from fork #77 (B7). --- .../scheduler/TableMaintenanceSchedulerPools.java | 136 +++++++-- .../scheduler/job/cron/CronSubmitGate.java | 34 +++ .../job/cron/MaintenanceJobSubmitter.java | 73 +++++ .../job/cron/MaintenanceSparkSubmitPipeline.java | 214 ++++++++++++++ .../job/cron/RelationalCronJobScheduler.java | 4 +- .../mapper/TableMaintenanceActionMapper.java | 5 + .../TableMaintenanceActionSQLProviderFactory.java | 10 + .../TableMaintenanceActionBaseSQLProvider.java | 16 + .../storage/service/OccupancyClaim.java | 3 +- .../service/TableMaintenanceActionMetaService.java | 22 ++ .../rest/feature/TableMaintenanceRESTFeature.java | 2 +- .../TestTableMaintenanceSchedulerPools.java | 7 +- .../cron/TestMaintenanceSparkSubmitPipeline.java | 321 +++++++++++++++++++++ .../TestTableMaintenanceActionMetaService.java | 30 ++ 14 files changed, 848 insertions(+), 29 deletions(-) diff --git a/plugins/tms/src/main/java/org/apache/gravitino/maintenance/scheduler/TableMaintenanceSchedulerPools.java b/plugins/tms/src/main/java/org/apache/gravitino/maintenance/scheduler/TableMaintenanceSchedulerPools.java index 88582781e1..fc1140e922 100644 --- a/plugins/tms/src/main/java/org/apache/gravitino/maintenance/scheduler/TableMaintenanceSchedulerPools.java +++ b/plugins/tms/src/main/java/org/apache/gravitino/maintenance/scheduler/TableMaintenanceSchedulerPools.java @@ -24,13 +24,29 @@ import com.github.kagkarlsson.scheduler.task.Task; import com.github.kagkarlsson.scheduler.task.helper.Tasks; import com.google.common.base.Preconditions; import java.time.Duration; +import java.time.Instant; import java.util.ArrayList; import java.util.List; +import java.util.Optional; +import java.util.concurrent.atomic.AtomicReference; +import java.util.function.BiConsumer; +import java.util.function.Supplier; import javax.sql.DataSource; +import org.apache.gravitino.Config; import org.apache.gravitino.json.JsonUtils; import org.apache.gravitino.maintenance.TableMaintenanceConfig; +import org.apache.gravitino.maintenance.occupancy.OccupancyGate; +import org.apache.gravitino.maintenance.scheduler.job.cron.MaintenanceJobConfBuilder; +import org.apache.gravitino.maintenance.scheduler.job.cron.MaintenanceJobSubmitter; +import org.apache.gravitino.maintenance.scheduler.job.cron.MaintenanceSparkSubmitPipeline; +import org.apache.gravitino.maintenance.scheduler.job.cron.PendingCronExpandOccupancyGate; +import org.apache.gravitino.maintenance.scheduler.job.cron.RelationalCronJobScheduler; +import org.apache.gravitino.maintenance.scheduler.policy.expand.NearestPolicyTableEnumerator; +import org.apache.gravitino.maintenance.scheduler.policy.expand.PolicyExpandPipeline; +import org.apache.gravitino.maintenance.storage.service.TableMaintenanceActionMetaService; import org.slf4j.Logger; import org.slf4j.LoggerFactory; +import org.apache.gravitino.maintenance.scheduler.policy.expand.ExpandOccupancyGate; /** * Starts and stops three isolated db-scheduler pools (§5.2, §7.2), each with its own task table @@ -39,8 +55,9 @@ import org.slf4j.LoggerFactory; * <p>db-scheduler requires a dedicated {@code tableName} per Scheduler instance for multi-pool * isolation; sharing one physical table across pools is unsupported (see design §5.2). * - * <p>Cron enqueue uses {@link #cronJobTask()} with Jackson {@code task_data}; pick submit lands in - * a follow-up. + * <p>The expand pool fans a due {@code policy-expand} lease into {@code cron-job} rows with an + * enqueue-time {@link MaintenanceJobSpec}. The cron pool consumes that spec, claims a soft slot, + * and submits Spark. The commit pool still only removes its one-shot row. */ public class TableMaintenanceSchedulerPools implements AutoCloseable { @@ -64,6 +81,21 @@ public class TableMaintenanceSchedulerPools implements AutoCloseable { /** One-shot on-commit chained Spark submit ({@code task_name}). */ public static final String COMMIT_JOB = "commit-job"; + /** + * Expand pipeline supplier used by {@link #policyExpandTask()}. {@link #start} installs the + * relational implementation. + */ + private static final AtomicReference<Supplier<PolicyExpandPipeline>> EXPAND_PIPELINE = + new AtomicReference<>( + () -> { + throw new IllegalStateException("policy-expand pipeline is not configured"); + }); + + /** Cron submit handler used by {@link #cronJobTask()}; bound in {@link #start}. */ + static final AtomicReference<BiConsumer<String, MaintenanceJobSpec>> CRON_SUBMIT = + new AtomicReference<>(unconfiguredSubmit("cron-job")); + + private final List<Scheduler> schedulers = new ArrayList<>(); /** @@ -74,27 +106,43 @@ public class TableMaintenanceSchedulerPools implements AutoCloseable { * * @param dataSource relational JDBC pool shared with the entity store * @param config maintenance scheduler settings + * @param gravitinoConfig server config for {@code minIntervalMs} fallbacks and jobConf overlay */ - public void start(DataSource dataSource, TableMaintenanceConfig config) { + public void start(DataSource dataSource, TableMaintenanceConfig config, Config gravitinoConfig) { Preconditions.checkState(schedulers.isEmpty(), "scheduler pools already started"); Preconditions.checkNotNull(dataSource, "dataSource must not be null"); + Preconditions.checkNotNull(gravitinoConfig, "gravitinoConfig must not be null"); + + OccupancyGate occupancyGate = + new OccupancyGate(TableMaintenanceActionMetaService.getInstance(), config.claimTimeoutMs()); + ExpandOccupancyGate expandGate = + new PendingCronExpandOccupancyGate(occupancyGate::shouldSkip, dataSource); + PolicyExpandPipeline pipeline = + new PolicyExpandPipeline( + new NearestPolicyTableEnumerator(), + new RelationalCronJobScheduler(dataSource, gravitinoConfig), + expandGate, + gravitinoConfig, + config.expandEnqueueBatchSize()); + EXPAND_PIPELINE.set(() -> pipeline); + + MaintenanceSparkSubmitPipeline cronSubmit = + new MaintenanceSparkSubmitPipeline( + occupancyGate::isInFlight, + TableMaintenanceActionMetaService.getInstance(), + MaintenanceJobSubmitter.fromEnv(), + new MaintenanceJobConfBuilder(gravitinoConfig), + config.claimTimeoutMs()); + CRON_SUBMIT.set(cronSubmit::submit); List<Scheduler> built = List.of( buildScheduler( - dataSource, - config, - config.expandThreads(), - EXPAND_TASKS_TABLE, - noopTask(POLICY_EXPAND)), + dataSource, config, config.expandThreads(), EXPAND_TASKS_TABLE, policyExpandTask()), buildScheduler( dataSource, config, config.cronThreads(), CRON_TASKS_TABLE, cronJobTask()), buildScheduler( - dataSource, - config, - config.commitThreads(), - COMMIT_TASKS_TABLE, - noopTask(COMMIT_JOB))); + dataSource, config, config.commitThreads(), COMMIT_TASKS_TABLE, commitJobTask())); try { for (Scheduler scheduler : built) { @@ -118,6 +166,11 @@ public class TableMaintenanceSchedulerPools implements AutoCloseable { } } schedulers.clear(); + EXPAND_PIPELINE.set( + () -> { + throw new IllegalStateException("policy-expand pipeline is not configured"); + }); + CRON_SUBMIT.set(unconfiguredSubmit("cron-job")); LOG.info("Stopped table maintenance scheduler pools"); } @@ -145,24 +198,61 @@ public class TableMaintenanceSchedulerPools implements AutoCloseable { } /** - * Cron-pool task stub: removes the one-shot row after dequeue. Real submit lands in a follow-up. + * Expand-pool task definition used by the running expand scheduler. + * + * <p>On each due pick the handler expands the policy into {@code cron-job} rows and reschedules + * (or removes) the expand lease. + * + * @return the {@code policy-expand} task + */ + public static Task<Void> policyExpandTask() { + return Tasks.custom(POLICY_EXPAND, Void.class) + .execute( + (taskInstance, executionContext) -> { + long policyId = Long.parseLong(taskInstance.getId()); + Optional<Instant> nextDue = EXPAND_PIPELINE.get().get().expand(policyId); + return (executionComplete, executionOperations) -> { + if (nextDue.isEmpty()) { + executionOperations.remove(); + } else { + executionOperations.reschedule(executionComplete, nextDue.get()); + } + }; + }); + } + + /** + * Cron-pool task definition. Consumes the enqueue-time spec, claims a soft slot, submits Spark, + * then removes the one-shot row so a later expand pick can schedule again. * * @return the {@code cron-job} task */ public static Task<MaintenanceJobSpec> cronJobTask() { return Tasks.custom(CRON_JOB, MaintenanceJobSpec.class) .execute( - (taskInstance, executionContext) -> - (executionComplete, executionOperations) -> executionOperations.remove()); + (taskInstance, executionContext) -> { + try { + CRON_SUBMIT.get().accept(taskInstance.getId(), taskInstance.getData()); + } catch (RuntimeException e) { + LOG.warn( + "cron-job submit failed for instance {}; removing one-shot row", + taskInstance.getId(), + e); + } + return (executionComplete, executionOperations) -> executionOperations.remove(); + }); } - private static Task<Void> noopTask(String taskName) { - return Tasks.custom(taskName, Void.class) + + /** + * Commit-pool task definition. Removes the one-shot row. Chained submit lands in a follow-up. + * + * @return the {@code commit-job} task + */ + public static Task<MaintenanceJobSpec> commitJobTask() { + return Tasks.custom(COMMIT_JOB, MaintenanceJobSpec.class) .execute( - (taskInstance, executionContext) -> { - LOG.debug( - "TMS scheduler callback for {} instance {}", taskName, taskInstance.getId()); - return (executionComplete, executionOperations) -> {}; - }); + (taskInstance, executionContext) -> + (executionComplete, executionOperations) -> executionOperations.remove()); } } diff --git a/plugins/tms/src/main/java/org/apache/gravitino/maintenance/scheduler/job/cron/CronSubmitGate.java b/plugins/tms/src/main/java/org/apache/gravitino/maintenance/scheduler/job/cron/CronSubmitGate.java new file mode 100644 index 0000000000..e2b42a7d75 --- /dev/null +++ b/plugins/tms/src/main/java/org/apache/gravitino/maintenance/scheduler/job/cron/CronSubmitGate.java @@ -0,0 +1,34 @@ +/* + * 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.maintenance.scheduler.job.cron; + +/** + * In-flight check for cron-job pick. {@code minIntervalMs} was already applied at expand enqueue; + * pick only blocks on a live soft slot. + */ +@FunctionalInterface +public interface CronSubmitGate { + + /** + * @param tableId table entity id + * @param policyId policy entity id + * @return true when that policy still holds a soft slot for the table + */ + boolean isInFlight(long tableId, long policyId); +} diff --git a/plugins/tms/src/main/java/org/apache/gravitino/maintenance/scheduler/job/cron/MaintenanceJobSubmitter.java b/plugins/tms/src/main/java/org/apache/gravitino/maintenance/scheduler/job/cron/MaintenanceJobSubmitter.java new file mode 100644 index 0000000000..54525187cb --- /dev/null +++ b/plugins/tms/src/main/java/org/apache/gravitino/maintenance/scheduler/job/cron/MaintenanceJobSubmitter.java @@ -0,0 +1,73 @@ +/* + * 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.maintenance.scheduler.job.cron; + +import java.util.Map; +import org.apache.gravitino.GravitinoEnv; +import org.apache.gravitino.job.JobHandle; +import org.apache.gravitino.meta.JobEntity; + +/** + * Submits a Gravitino Job for a TMS cron-job callback (§5.2.2). + * + * <p>Production wiring uses {@link GravitinoEnv#jobOperationDispatcher()}; tests inject a fake. + */ +public interface MaintenanceJobSubmitter { + + /** + * Runs a job template and returns the persisted job-run id for occupancy bind. + * + * @param metalake metalake name + * @param jobTemplateName built-in or registered job template name + * @param jobConf runtime job configuration (policy jobOptions + auth overlay) + * @return {@link JobEntity#id()} + */ + long runJob(String metalake, String jobTemplateName, Map<String, String> jobConf); + + /** + * Best-effort cancel after a successful {@link #runJob} when occupancy bind fails. + * + * @param metalake metalake name + * @param jobId numeric Gravitino job run id + */ + void cancelJob(String metalake, long jobId); + + /** + * @return submitter that delegates to {@link GravitinoEnv#jobOperationDispatcher()} + */ + static MaintenanceJobSubmitter fromEnv() { + return new MaintenanceJobSubmitter() { + @Override + public long runJob(String metalake, String jobTemplateName, Map<String, String> jobConf) { + JobEntity job = + GravitinoEnv.getInstance() + .jobOperationDispatcher() + .runJob(metalake, jobTemplateName, jobConf); + return job.id(); + } + + @Override + public void cancelJob(String metalake, long jobId) { + GravitinoEnv.getInstance() + .jobOperationDispatcher() + .cancelJob(metalake, JobHandle.JOB_ID_PREFIX + jobId); + } + }; + } +} diff --git a/plugins/tms/src/main/java/org/apache/gravitino/maintenance/scheduler/job/cron/MaintenanceSparkSubmitPipeline.java b/plugins/tms/src/main/java/org/apache/gravitino/maintenance/scheduler/job/cron/MaintenanceSparkSubmitPipeline.java new file mode 100644 index 0000000000..10eabbb8b9 --- /dev/null +++ b/plugins/tms/src/main/java/org/apache/gravitino/maintenance/scheduler/job/cron/MaintenanceSparkSubmitPipeline.java @@ -0,0 +1,214 @@ +/* + * 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.maintenance.scheduler.job.cron; + +import com.google.common.base.Preconditions; +import java.util.Map; +import java.util.Optional; +import javax.annotation.Nullable; +import org.apache.gravitino.maintenance.scheduler.MaintenanceJobSpec; +import org.apache.gravitino.maintenance.scheduler.MaintenanceJobSpec.Operation; +import org.apache.gravitino.maintenance.scheduler.MaintenanceJobSpec.TableTarget; +import org.apache.gravitino.maintenance.scheduler.MaintenanceJobSpecFactory; +import org.apache.gravitino.maintenance.storage.service.OccupancyClaim; +import org.apache.gravitino.maintenance.storage.service.TableMaintenanceActionMetaService; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +/** + * Cron-pool control-plane submit (§5.2.2): consume enqueue-time {@link MaintenanceJobSpec} → + * in-flight gate → soft-slot claim → {@code runJob} → bind job id. + * + * <p>Single-table {@code table:...} and snapshot-expiry {@code batch:...} rows share the same spec + * shape; batch pick overlays per-table keys and submits one Spark job per table. If any spec table + * is in-flight, the whole pick stops without partial submits (one-shot row is still removed by the + * caller). + * + * <p>Does not sample metrics or evaluate trigger-expr (both run inside Spark from enqueue-time + * {@code rules} in jobConf). Policy / jobConf are not recomputed here. + */ +public final class MaintenanceSparkSubmitPipeline { + + private static final Logger LOG = LoggerFactory.getLogger(MaintenanceSparkSubmitPipeline.class); + + private final CronSubmitGate occupancyGate; + private final TableMaintenanceActionMetaService actionMetaService; + private final MaintenanceJobSubmitter jobSubmitter; + private final MaintenanceJobConfBuilder jobConfBuilder; + private final long claimTimeoutMs; + + /** + * @param occupancyGate in-flight gate (minInterval already applied at enqueue) + * @param actionMetaService soft-slot claim / bind + * @param jobSubmitter Gravitino Job submitter + * @param jobConfBuilder resolves enqueue-time conf-key refs before {@code runJob} + * @param claimTimeoutMs unbound claim TTL + */ + public MaintenanceSparkSubmitPipeline( + CronSubmitGate occupancyGate, + TableMaintenanceActionMetaService actionMetaService, + MaintenanceJobSubmitter jobSubmitter, + MaintenanceJobConfBuilder jobConfBuilder, + long claimTimeoutMs) { + this.occupancyGate = Preconditions.checkNotNull(occupancyGate); + this.actionMetaService = Preconditions.checkNotNull(actionMetaService); + this.jobSubmitter = Preconditions.checkNotNull(jobSubmitter); + this.jobConfBuilder = Preconditions.checkNotNull(jobConfBuilder); + Preconditions.checkArgument(claimTimeoutMs > 0, "claimTimeoutMs must be positive"); + this.claimTimeoutMs = claimTimeoutMs; + } + + /** + * Runs one cron-job pick for {@code instanceId}. + * + * @param instanceId db-scheduler cron-job instance id + * @param spec enqueue-time payload; required + */ + public void submit(String instanceId, @Nullable MaintenanceJobSpec spec) { + if (spec == null || !spec.isValid()) { + LOG.warn("Skipping cron-job submit for {} with missing/invalid task_data spec", instanceId); + return; + } + + Optional<CronJobIds.TablePolicyIds> tableIds = CronJobIds.parseTable(instanceId); + Optional<CronJobIds.BatchPolicyIds> batchIds = CronJobIds.parseBatch(instanceId); + if (tableIds.isEmpty() && batchIds.isEmpty()) { + LOG.warn("Skipping cron-job submit for unparseable instance id {}", instanceId); + return; + } + + Operation step = spec.getOperations().get(0); + long policyId = step.getPolicyId(); + long expectedPolicyId = + tableIds.isPresent() ? tableIds.get().policyId() : batchIds.get().policyId(); + if (policyId != expectedPolicyId) { + LOG.warn( + "Skipping cron-job submit for {}: spec policyId {} != instance policyId {}", + instanceId, + policyId, + expectedPolicyId); + return; + } + if (tableIds.isPresent()) { + long expectedTableId = tableIds.get().tableId(); + if (spec.getTables().size() != 1 || spec.getTables().get(0).getTableId() != expectedTableId) { + LOG.warn( + "Skipping cron-job submit for {}: spec tables {} do not match instance tableId {}", + instanceId, + spec.tableIds(), + expectedTableId); + return; + } + } + + // minInterval was gated at expand enqueue; pick only blocks on in-flight. Any in-flight + // member aborts the whole pick so a batch row is not partially submitted then deleted. + for (TableTarget target : spec.getTables()) { + if (occupancyGate.isInFlight(target.getTableId(), policyId)) { + LOG.info( + "Skipping cron-job {} entirely (table {} in-flight for policy {})", + instanceId, + target.getTableId(), + policyId); + return; + } + } + for (TableTarget target : spec.getTables()) { + submitOne(spec, target, policyId); + } + } + + private void submitOne(MaintenanceJobSpec spec, TableTarget target, long policyId) { + Optional<OccupancyClaim> claim = + actionMetaService.tryAcquireInFlight(target.getTableId(), policyId, claimTimeoutMs); + if (claim.isEmpty()) { + LOG.info( + "Skipping cron-job submit for table {} policy {} (lost soft-slot race)", + target.getTableId(), + policyId); + return; + } + + Map<String, String> jobConf; + try { + jobConf = + jobConfBuilder.resolveForSubmit( + MaintenanceJobSpecFactory.jobConfForTable(spec.getJobConf(), target)); + } catch (RuntimeException e) { + releaseClaimQuietly(claim.get()); + LOG.warn( + "Skipping cron-job submit for table {} policy {}: invalid spec table target: {}", + target.getTableId(), + policyId, + e.toString()); + return; + } + runAndBind(spec.metalake(), jobConf, claim.get(), target.getTableId(), policyId); + } + + private void runAndBind( + String metalake, + Map<String, String> jobConf, + OccupancyClaim claim, + long tableId, + long policyId) { + Long jobId = null; + try { + jobId = + jobSubmitter.runJob( + metalake, MaintenanceJobTemplates.TMS_ICEBERG_MAINTENANCE_TEMPLATE, jobConf); + actionMetaService.bindJobId(claim, jobId); + LOG.info("Submitted cron-job for table {} policy {} as job {}", tableId, policyId, jobId); + } catch (RuntimeException e) { + if (jobId != null) { + cancelJobQuietly(metalake, jobId); + } + releaseClaimQuietly(claim); + LOG.warn( + "Failed cron-job submit for table {} policy {}: {}", tableId, policyId, e.toString()); + } + } + + private void cancelJobQuietly(String metalake, long jobId) { + try { + jobSubmitter.cancelJob(metalake, jobId); + } catch (RuntimeException cancelError) { + LOG.warn( + "Failed to cancel orphaned cron Spark job {} under metalake {} after bind failure", + jobId, + metalake, + cancelError); + } + } + + private void releaseClaimQuietly(OccupancyClaim claim) { + try { + // Fence on (id, created_at) + unbound so a concurrent reclaim's live claim is not deleted. + if (!actionMetaService.deleteUnboundClaim(claim)) { + LOG.debug( + "Occupancy claim {} created_at={} was not released (reclaimed or already bound)", + claim.id(), + claim.createdAt()); + } + } catch (RuntimeException releaseError) { + LOG.warn( + "Failed to release occupancy claim {} after submit failure", claim.id(), releaseError); + } + } +} diff --git a/plugins/tms/src/main/java/org/apache/gravitino/maintenance/scheduler/job/cron/RelationalCronJobScheduler.java b/plugins/tms/src/main/java/org/apache/gravitino/maintenance/scheduler/job/cron/RelationalCronJobScheduler.java index 1fd173ec0d..863025715c 100644 --- a/plugins/tms/src/main/java/org/apache/gravitino/maintenance/scheduler/job/cron/RelationalCronJobScheduler.java +++ b/plugins/tms/src/main/java/org/apache/gravitino/maintenance/scheduler/job/cron/RelationalCronJobScheduler.java @@ -311,9 +311,7 @@ public final class RelationalCronJobScheduler implements CronJobScheduler { } MaintenanceJobSpec merged = new MaintenanceJobSpec( - new ArrayList<>(byId.values()), - open.spec.getOperations(), - open.spec.getJobConf()); + new ArrayList<>(byId.values()), open.spec.getOperations(), open.spec.getJobConf()); try { return client.reschedule(open.instanceId, open.executionTime, merged); } catch (RuntimeException e) { diff --git a/plugins/tms/src/main/java/org/apache/gravitino/maintenance/storage/mapper/TableMaintenanceActionMapper.java b/plugins/tms/src/main/java/org/apache/gravitino/maintenance/storage/mapper/TableMaintenanceActionMapper.java index c7313611ee..0453d0cd0a 100644 --- a/plugins/tms/src/main/java/org/apache/gravitino/maintenance/storage/mapper/TableMaintenanceActionMapper.java +++ b/plugins/tms/src/main/java/org/apache/gravitino/maintenance/storage/mapper/TableMaintenanceActionMapper.java @@ -122,6 +122,11 @@ public interface TableMaintenanceActionMapper { @DeleteProvider(type = TableMaintenanceActionSQLProviderFactory.class, method = "deleteById") int deleteById(@Param("id") long id); + @DeleteProvider( + type = TableMaintenanceActionSQLProviderFactory.class, + method = "deleteUnboundClaim") + int deleteUnboundClaim(@Param("id") long id, @Param("expectedCreatedAt") long expectedCreatedAt); + @DeleteProvider(type = TableMaintenanceActionSQLProviderFactory.class, method = "deleteByJobId") int deleteByJobId(@Param("jobId") long jobId); diff --git a/plugins/tms/src/main/java/org/apache/gravitino/maintenance/storage/mapper/TableMaintenanceActionSQLProviderFactory.java b/plugins/tms/src/main/java/org/apache/gravitino/maintenance/storage/mapper/TableMaintenanceActionSQLProviderFactory.java index 784da61e70..f4a0aaf3fe 100644 --- a/plugins/tms/src/main/java/org/apache/gravitino/maintenance/storage/mapper/TableMaintenanceActionSQLProviderFactory.java +++ b/plugins/tms/src/main/java/org/apache/gravitino/maintenance/storage/mapper/TableMaintenanceActionSQLProviderFactory.java @@ -207,6 +207,16 @@ public class TableMaintenanceActionSQLProviderFactory { return getProvider().deleteById(id); } + /** + * @param id action id + * @param expectedCreatedAt claim lease start + * @return DELETE statement + */ + public static String deleteUnboundClaim( + @Param("id") long id, @Param("expectedCreatedAt") long expectedCreatedAt) { + return getProvider().deleteUnboundClaim(id, expectedCreatedAt); + } + /** * @param jobId job run id * @return DELETE statement diff --git a/plugins/tms/src/main/java/org/apache/gravitino/maintenance/storage/mapper/provider/base/TableMaintenanceActionBaseSQLProvider.java b/plugins/tms/src/main/java/org/apache/gravitino/maintenance/storage/mapper/provider/base/TableMaintenanceActionBaseSQLProvider.java index a18f685fbd..e29ca6c454 100644 --- a/plugins/tms/src/main/java/org/apache/gravitino/maintenance/storage/mapper/provider/base/TableMaintenanceActionBaseSQLProvider.java +++ b/plugins/tms/src/main/java/org/apache/gravitino/maintenance/storage/mapper/provider/base/TableMaintenanceActionBaseSQLProvider.java @@ -258,6 +258,22 @@ public class TableMaintenanceActionBaseSQLProvider { return "DELETE FROM " + TABLE_NAME + " WHERE id = #{id}"; } + /** + * Deletes an unbound claim only when the lease {@code created_at} still matches. A concurrent + * reclaim that CAS-updated {@code created_at} leaves the new owner's row intact. + * + * @param id action row id + * @param expectedCreatedAt claim lease start observed at acquire time + * @return DELETE statement + */ + public String deleteUnboundClaim( + @Param("id") long id, @Param("expectedCreatedAt") long expectedCreatedAt) { + return "DELETE FROM " + + TABLE_NAME + + " WHERE id = #{id} AND job_id IS NULL AND finished_at = 0" + + " AND created_at = #{expectedCreatedAt}"; + } + /** * @param jobId Gravitino job run id * @return DELETE statement diff --git a/plugins/tms/src/main/java/org/apache/gravitino/maintenance/storage/service/OccupancyClaim.java b/plugins/tms/src/main/java/org/apache/gravitino/maintenance/storage/service/OccupancyClaim.java index adc4f5bcdd..5f927329f2 100644 --- a/plugins/tms/src/main/java/org/apache/gravitino/maintenance/storage/service/OccupancyClaim.java +++ b/plugins/tms/src/main/java/org/apache/gravitino/maintenance/storage/service/OccupancyClaim.java @@ -24,7 +24,8 @@ import java.util.Objects; * Immutable occupancy claim returned by acquire. * * <p>Callers must pass {@link #createdAt()} to {@link TableMaintenanceActionMetaService#bindJobId} - * without re-reading {@code created_at} from the row. + * and {@link TableMaintenanceActionMetaService#deleteUnboundClaim} without re-reading {@code + * created_at} from the row. * * <p>{@link #id()} is the durable claim token for OccupancyGate follow-ups: make {@code runJob} * idempotent on this id (or outbox / reconcile-before-reclaim). Binding {@code job_id} alone cannot diff --git a/plugins/tms/src/main/java/org/apache/gravitino/maintenance/storage/service/TableMaintenanceActionMetaService.java b/plugins/tms/src/main/java/org/apache/gravitino/maintenance/storage/service/TableMaintenanceActionMetaService.java index 454243bb10..f73ebfd59c 100644 --- a/plugins/tms/src/main/java/org/apache/gravitino/maintenance/storage/service/TableMaintenanceActionMetaService.java +++ b/plugins/tms/src/main/java/org/apache/gravitino/maintenance/storage/service/TableMaintenanceActionMetaService.java @@ -429,6 +429,28 @@ public class TableMaintenanceActionMetaService { SessionUtils.doWithCommit(TableMaintenanceActionMapper.class, mapper -> mapper.deleteById(id)); } + /** + * Releases an unbound occupancy claim when submit / bind fails. + * + * <p>Deletes only when {@code (id, created_at)} still match and the row is unbound. A concurrent + * reclaim that CAS-updated {@code created_at} leaves the new owner's live claim intact (returns + * {@code false}). + * + * @param claim claim returned by {@link #tryAcquireInFlight} / {@link #insertInFlight} + * @return {@code true} when this claim lease was deleted + */ + @Monitored( + metricsSource = GRAVITINO_RELATIONAL_STORE_METRIC_NAME, + baseMetricName = "deleteUnboundTableMaintenanceActionClaim") + public boolean deleteUnboundClaim(OccupancyClaim claim) { + Preconditions.checkNotNull(claim, "claim must not be null"); + int deleted = + SessionUtils.doWithCommitAndFetchResult( + TableMaintenanceActionMapper.class, + mapper -> mapper.deleteUnboundClaim(claim.id(), claim.createdAt())); + return deleted == 1; + } + /** * Deletes the {@code table_maintenance_action} row keyed by job run id. Does not delete metrics. * diff --git a/plugins/tms/src/main/java/org/apache/gravitino/maintenance/web/rest/feature/TableMaintenanceRESTFeature.java b/plugins/tms/src/main/java/org/apache/gravitino/maintenance/web/rest/feature/TableMaintenanceRESTFeature.java index 1b9473553b..0930ed3d61 100644 --- a/plugins/tms/src/main/java/org/apache/gravitino/maintenance/web/rest/feature/TableMaintenanceRESTFeature.java +++ b/plugins/tms/src/main/java/org/apache/gravitino/maintenance/web/rest/feature/TableMaintenanceRESTFeature.java @@ -86,7 +86,7 @@ public class TableMaintenanceRESTFeature implements Feature { DataSource dataSource = resolveEntityStoreDataSource(); TableMaintenanceSchedulerPools pools = new TableMaintenanceSchedulerPools(); - pools.start(dataSource, TableMaintenanceConfig.from(config)); + pools.start(dataSource, TableMaintenanceConfig.from(config), config); if (!SCHEDULER_POOLS.compareAndSet(null, pools)) { pools.stop(); LOG.warn("Table maintenance scheduler pools were started concurrently; discarded duplicate"); diff --git a/plugins/tms/src/test/java/org/apache/gravitino/maintenance/scheduler/TestTableMaintenanceSchedulerPools.java b/plugins/tms/src/test/java/org/apache/gravitino/maintenance/scheduler/TestTableMaintenanceSchedulerPools.java index f654612bbb..ce90bf342b 100644 --- a/plugins/tms/src/test/java/org/apache/gravitino/maintenance/scheduler/TestTableMaintenanceSchedulerPools.java +++ b/plugins/tms/src/test/java/org/apache/gravitino/maintenance/scheduler/TestTableMaintenanceSchedulerPools.java @@ -20,6 +20,7 @@ package org.apache.gravitino.maintenance.scheduler; import javax.sql.DataSource; +import org.apache.gravitino.Config; import org.apache.gravitino.maintenance.TableMaintenanceConfig; import org.apache.gravitino.storage.relational.TestJDBCBackend; import org.apache.gravitino.storage.relational.session.SqlSessionFactoryHelper; @@ -47,9 +48,13 @@ class TestTableMaintenanceSchedulerPools extends TestJDBCBackend { .getEnvironment() .getDataSource(); pools = new TableMaintenanceSchedulerPools(); + Config gravitinoConfig = new Config(false) {}; Assertions.assertDoesNotThrow( () -> - pools.start(dataSource, TableMaintenanceConfig.forTest(1, 1, 1, 500L, 1_000L, 4, 10))); + pools.start( + dataSource, + TableMaintenanceConfig.forTest(1, 1, 1, 500L, 1_000L, 4, 10), + gravitinoConfig)); Assertions.assertDoesNotThrow(pools::stop); } } diff --git a/plugins/tms/src/test/java/org/apache/gravitino/maintenance/scheduler/job/cron/TestMaintenanceSparkSubmitPipeline.java b/plugins/tms/src/test/java/org/apache/gravitino/maintenance/scheduler/job/cron/TestMaintenanceSparkSubmitPipeline.java new file mode 100644 index 0000000000..ff9867772e --- /dev/null +++ b/plugins/tms/src/test/java/org/apache/gravitino/maintenance/scheduler/job/cron/TestMaintenanceSparkSubmitPipeline.java @@ -0,0 +1,321 @@ +/* + * 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.maintenance.scheduler.job.cron; + +import static org.mockito.ArgumentMatchers.any; +import static org.mockito.ArgumentMatchers.anyLong; +import static org.mockito.ArgumentMatchers.anyMap; +import static org.mockito.ArgumentMatchers.anyString; +import static org.mockito.ArgumentMatchers.eq; +import static org.mockito.Mockito.doThrow; +import static org.mockito.Mockito.mock; +import static org.mockito.Mockito.never; +import static org.mockito.Mockito.times; +import static org.mockito.Mockito.verify; +import static org.mockito.Mockito.when; + +import java.util.LinkedHashMap; +import java.util.List; +import java.util.Map; +import java.util.Optional; +import org.apache.gravitino.Config; +import org.apache.gravitino.maintenance.scheduler.MaintenanceJobSpec; +import org.apache.gravitino.maintenance.scheduler.MaintenanceJobSpec.Operation; +import org.apache.gravitino.maintenance.scheduler.MaintenanceJobSpec.TableTarget; +import org.apache.gravitino.maintenance.storage.service.OccupancyClaim; +import org.apache.gravitino.maintenance.storage.service.TableMaintenanceActionMetaService; +import org.junit.jupiter.api.Assertions; +import org.junit.jupiter.api.Test; +import org.mockito.ArgumentCaptor; + +class TestMaintenanceSparkSubmitPipeline { + + private static final long TABLE_ID = 101L; + private static final long POLICY_ID = 202L; + private static final String INSTANCE = CronJobIds.forTable(TABLE_ID, POLICY_ID); + private static final long CLAIM_TIMEOUT_MS = 60_000L; + + @Test + void testUnparseableInstanceSkipsSubmit() { + MaintenanceJobSubmitter submitter = mock(MaintenanceJobSubmitter.class); + TableMaintenanceActionMetaService actions = mock(TableMaintenanceActionMetaService.class); + pipeline(mockGate(false), actions, submitter).submit("not-a-cron-id", spec()); + verify(submitter, never()).runJob(anyString(), anyString(), anyMap()); + verify(actions, never()).tryAcquireInFlight(anyLong(), anyLong(), anyLong()); + } + + @Test + void testNullSpecSkipsSubmit() { + MaintenanceJobSubmitter submitter = mock(MaintenanceJobSubmitter.class); + TableMaintenanceActionMetaService actions = mock(TableMaintenanceActionMetaService.class); + pipeline(mockGate(false), actions, submitter).submit(INSTANCE, null); + verify(submitter, never()).runJob(anyString(), anyString(), anyMap()); + } + + @Test + void testInFlightDoesNotClaimOrSubmit() { + MaintenanceJobSubmitter submitter = mock(MaintenanceJobSubmitter.class); + TableMaintenanceActionMetaService actions = mock(TableMaintenanceActionMetaService.class); + pipeline(mockGate(true), actions, submitter).submit(INSTANCE, spec()); + verify(actions, never()).tryAcquireInFlight(anyLong(), anyLong(), anyLong()); + verify(submitter, never()).runJob(anyString(), anyString(), anyMap()); + } + + @Test + void testBatchInFlightAbortsEntirePick() { + MaintenanceJobSubmitter submitter = mock(MaintenanceJobSubmitter.class); + TableMaintenanceActionMetaService actions = mock(TableMaintenanceActionMetaService.class); + CronSubmitGate gate = (tableId, policyId) -> tableId == 11L; + + Map<String, String> shared = new LinkedHashMap<>(); + shared.put(MaintenanceJobTemplates.MAINTENANCE_TYPES_KEY, "expire_snapshots"); + shared.put("updater_options", "{\"gravitino_uri\":\"http://localhost:8090\"}";); + MaintenanceJobSpec batch = + new MaintenanceJobSpec( + List.of( + new TableTarget(10L, List.of("metalake", "iceberg", "db", "t1")), + new TableTarget(11L, List.of("metalake", "iceberg", "db", "t2"))), + List.of(new Operation(POLICY_ID, "expire_snapshots")), + shared); + + pipeline(gate, actions, submitter).submit(CronJobIds.forBatch(0L, POLICY_ID), batch); + + verify(actions, never()).tryAcquireInFlight(anyLong(), anyLong(), anyLong()); + verify(submitter, never()).runJob(anyString(), anyString(), anyMap()); + } + + @Test + void testLostClaimRaceSkipsSubmit() { + MaintenanceJobSubmitter submitter = mock(MaintenanceJobSubmitter.class); + TableMaintenanceActionMetaService actions = mock(TableMaintenanceActionMetaService.class); + when(actions.tryAcquireInFlight(TABLE_ID, POLICY_ID, CLAIM_TIMEOUT_MS)) + .thenReturn(Optional.empty()); + + pipeline(mockGate(false), actions, submitter).submit(INSTANCE, spec()); + + verify(submitter, never()).runJob(anyString(), anyString(), anyMap()); + } + + @Test + void testSuccessfulSubmitBindsJobId() { + MaintenanceJobSubmitter submitter = mock(MaintenanceJobSubmitter.class); + when(submitter.runJob( + eq("metalake"), eq(MaintenanceJobTemplates.TMS_ICEBERG_MAINTENANCE_TEMPLATE), anyMap())) + .thenReturn(999L); + TableMaintenanceActionMetaService actions = mock(TableMaintenanceActionMetaService.class); + OccupancyClaim claim = new OccupancyClaim(7L, 1_000L); + when(actions.tryAcquireInFlight(TABLE_ID, POLICY_ID, CLAIM_TIMEOUT_MS)) + .thenReturn(Optional.of(claim)); + + pipeline(mockGate(false), actions, submitter).submit(INSTANCE, spec()); + + verify(actions).bindJobId(claim, 999L); + verify(actions, never()).deleteUnboundClaim(any()); + } + + @Test + void testSubmitFailureReleasesClaim() { + MaintenanceJobSubmitter submitter = mock(MaintenanceJobSubmitter.class); + when(submitter.runJob(anyString(), anyString(), anyMap())) + .thenThrow(new RuntimeException("submit failed")); + TableMaintenanceActionMetaService actions = mock(TableMaintenanceActionMetaService.class); + OccupancyClaim claim = new OccupancyClaim(8L, 2_000L); + when(actions.tryAcquireInFlight(TABLE_ID, POLICY_ID, CLAIM_TIMEOUT_MS)) + .thenReturn(Optional.of(claim)); + when(actions.deleteUnboundClaim(claim)).thenReturn(true); + + Assertions.assertDoesNotThrow( + () -> pipeline(mockGate(false), actions, submitter).submit(INSTANCE, spec())); + + verify(actions).deleteUnboundClaim(claim); + verify(actions, never()).bindJobId(any(), anyLong()); + verify(submitter, never()).cancelJob(anyString(), anyLong()); + } + + @Test + void testBindFailureCancelsJobAndReleasesClaim() { + MaintenanceJobSubmitter submitter = mock(MaintenanceJobSubmitter.class); + when(submitter.runJob( + eq("metalake"), eq(MaintenanceJobTemplates.TMS_ICEBERG_MAINTENANCE_TEMPLATE), anyMap())) + .thenReturn(777L); + TableMaintenanceActionMetaService actions = mock(TableMaintenanceActionMetaService.class); + OccupancyClaim claim = new OccupancyClaim(9L, 3_000L); + when(actions.tryAcquireInFlight(TABLE_ID, POLICY_ID, CLAIM_TIMEOUT_MS)) + .thenReturn(Optional.of(claim)); + doThrow(new RuntimeException("bind failed")).when(actions).bindJobId(claim, 777L); + when(actions.deleteUnboundClaim(claim)).thenReturn(true); + + Assertions.assertDoesNotThrow( + () -> pipeline(mockGate(false), actions, submitter).submit(INSTANCE, spec())); + + verify(submitter).cancelJob("metalake", 777L); + verify(actions).deleteUnboundClaim(claim); + } + + @Test + void testBindFailureSkipsReleaseWhenClaimAlreadyReclaimed() { + MaintenanceJobSubmitter submitter = mock(MaintenanceJobSubmitter.class); + when(submitter.runJob( + eq("metalake"), eq(MaintenanceJobTemplates.TMS_ICEBERG_MAINTENANCE_TEMPLATE), anyMap())) + .thenReturn(778L); + TableMaintenanceActionMetaService actions = mock(TableMaintenanceActionMetaService.class); + OccupancyClaim staleClaim = new OccupancyClaim(10L, 4_000L); + when(actions.tryAcquireInFlight(TABLE_ID, POLICY_ID, CLAIM_TIMEOUT_MS)) + .thenReturn(Optional.of(staleClaim)); + doThrow(new RuntimeException("bind lost lease")).when(actions).bindJobId(staleClaim, 778L); + // Concurrent reclaim already moved created_at — fenced delete matches zero rows. + when(actions.deleteUnboundClaim(staleClaim)).thenReturn(false); + + Assertions.assertDoesNotThrow( + () -> pipeline(mockGate(false), actions, submitter).submit(INSTANCE, spec())); + + verify(submitter).cancelJob("metalake", 778L); + verify(actions).deleteUnboundClaim(staleClaim); + verify(actions, never()).deleteById(anyLong()); + } + + @Test + void testBatchSubmitOneSparkJobPerTable() { + MaintenanceJobSubmitter submitter = mock(MaintenanceJobSubmitter.class); + when(submitter.runJob( + eq("metalake"), eq(MaintenanceJobTemplates.TMS_ICEBERG_MAINTENANCE_TEMPLATE), anyMap())) + .thenReturn(555L, 556L); + TableMaintenanceActionMetaService actions = mock(TableMaintenanceActionMetaService.class); + OccupancyClaim claimA = new OccupancyClaim(1L, 10L); + OccupancyClaim claimB = new OccupancyClaim(2L, 20L); + when(actions.tryAcquireInFlight(10L, POLICY_ID, CLAIM_TIMEOUT_MS)) + .thenReturn(Optional.of(claimA)); + when(actions.tryAcquireInFlight(11L, POLICY_ID, CLAIM_TIMEOUT_MS)) + .thenReturn(Optional.of(claimB)); + + Map<String, String> shared = new LinkedHashMap<>(); + shared.put(MaintenanceJobTemplates.MAINTENANCE_TYPES_KEY, "expire_snapshots"); + shared.put("updater_options", "{\"gravitino_uri\":\"http://localhost:8090\"}";); + + MaintenanceJobSpec batch = + new MaintenanceJobSpec( + List.of( + new TableTarget(10L, List.of("metalake", "iceberg", "db", "t1")), + new TableTarget(11L, List.of("metalake", "iceberg", "db", "t2"))), + List.of(new Operation(POLICY_ID, "expire_snapshots")), + shared); + + pipeline(mockGate(false), actions, submitter).submit(CronJobIds.forBatch(0L, POLICY_ID), batch); + + @SuppressWarnings("unchecked") + ArgumentCaptor<Map<String, String>> confCaptor = ArgumentCaptor.forClass(Map.class); + verify(submitter, times(2)) + .runJob( + eq("metalake"), + eq(MaintenanceJobTemplates.TMS_ICEBERG_MAINTENANCE_TEMPLATE), + confCaptor.capture()); + List<Map<String, String>> confs = confCaptor.getAllValues(); + Assertions.assertEquals("iceberg", confs.get(0).get("catalog_name")); + Assertions.assertNull(confs.get(0).get("table_identifier")); + Assertions.assertTrue(confs.get(0).get("tables").contains("\"table_identifier\":\"db.t1\"")); + Assertions.assertTrue(confs.get(0).get("tables").contains("\"table_id\":10")); + Assertions.assertFalse(confs.get(0).containsKey("tms_table_id")); + Assertions.assertEquals("expire_snapshots", confs.get(0).get("maintenance_types")); + Assertions.assertEquals("iceberg", confs.get(1).get("catalog_name")); + Assertions.assertNull(confs.get(1).get("table_identifier")); + Assertions.assertTrue(confs.get(1).get("tables").contains("\"table_identifier\":\"db.t2\"")); + Assertions.assertTrue(confs.get(1).get("tables").contains("\"table_id\":11")); + Assertions.assertFalse(confs.get(1).containsKey("tms_table_id")); + Assertions.assertFalse(confs.get(0).containsKey("tms_policy_id")); + verify(actions).bindJobId(claimA, 555L); + verify(actions).bindJobId(claimB, 556L); + } + + @Test + void testConsumesSpecJobConfWithoutRebuild() { + MaintenanceJobSubmitter submitter = mock(MaintenanceJobSubmitter.class); + when(submitter.runJob(anyString(), anyString(), anyMap())).thenReturn(1L); + TableMaintenanceActionMetaService actions = mock(TableMaintenanceActionMetaService.class); + when(actions.tryAcquireInFlight(TABLE_ID, POLICY_ID, CLAIM_TIMEOUT_MS)) + .thenReturn(Optional.of(new OccupancyClaim(1L, 1L))); + + Map<String, String> jobConf = new LinkedHashMap<>(); + jobConf.put(MaintenanceJobTemplates.MAINTENANCE_TYPES_KEY, "rewrite_data_files"); + jobConf.put("strategy", "binpack"); + jobConf.put( + "updater_options", + "{\"gravitino_uri\":\"http://localhost:8090\",\"metalake\":\"metalake\"}";); + jobConf.put("spark_conf", "{\"spark.sql.catalog.iceberg.rest.auth.type\":\"basic\"}"); + + pipeline(mockGate(false), actions, submitter).submit(INSTANCE, spec(jobConf)); + + @SuppressWarnings("unchecked") + ArgumentCaptor<Map<String, String>> confCaptor = ArgumentCaptor.forClass(Map.class); + verify(submitter) + .runJob( + eq("metalake"), + eq(MaintenanceJobTemplates.TMS_ICEBERG_MAINTENANCE_TEMPLATE), + confCaptor.capture()); + Map<String, String> conf = confCaptor.getValue(); + Assertions.assertEquals("iceberg", conf.get("catalog_name")); + Assertions.assertNull(conf.get("table_identifier")); + Assertions.assertTrue(conf.get("tables").contains("\"table_identifier\":\"db.t\"")); + Assertions.assertTrue(conf.get("tables").contains("\"table_id\":" + TABLE_ID)); + Assertions.assertFalse(conf.containsKey("tms_table_id")); + Assertions.assertEquals("rewrite_data_files", conf.get("maintenance_types")); + Assertions.assertFalse(conf.containsKey("tms_policy_id")); + Assertions.assertEquals("binpack", conf.get("strategy")); + Assertions.assertTrue(conf.get("updater_options").contains("gravitino_uri")); + Assertions.assertTrue(conf.get("spark_conf").contains("basic")); + } + + @Test + void testPolicyIdMismatchSkipsSubmit() { + MaintenanceJobSubmitter submitter = mock(MaintenanceJobSubmitter.class); + TableMaintenanceActionMetaService actions = mock(TableMaintenanceActionMetaService.class); + pipeline(mockGate(false), actions, submitter) + .submit(CronJobIds.forTable(TABLE_ID, POLICY_ID + 1), spec()); + verify(submitter, never()).runJob(anyString(), anyString(), anyMap()); + } + + private static MaintenanceSparkSubmitPipeline pipeline( + CronSubmitGate gate, + TableMaintenanceActionMetaService actions, + MaintenanceJobSubmitter submitter) { + return new MaintenanceSparkSubmitPipeline( + gate, + actions, + submitter, + new MaintenanceJobConfBuilder(new Config(false) {}), + CLAIM_TIMEOUT_MS); + } + + private static CronSubmitGate mockGate(boolean inFlight) { + return (tableId, policyId) -> inFlight; + } + + private static MaintenanceJobSpec spec() { + Map<String, String> jobConf = new LinkedHashMap<>(); + jobConf.put(MaintenanceJobTemplates.MAINTENANCE_TYPES_KEY, "rewrite_data_files"); + jobConf.put("updater_options", "{\"gravitino_uri\":\"http://localhost:8090\"}";); + return spec(jobConf); + } + + private static MaintenanceJobSpec spec(Map<String, String> jobConf) { + return new MaintenanceJobSpec( + List.of(new TableTarget(TABLE_ID, List.of("metalake", "iceberg", "db", "t"))), + List.of(new Operation(POLICY_ID, "rewrite_data_files")), + jobConf); + } +} diff --git a/plugins/tms/src/test/java/org/apache/gravitino/maintenance/storage/service/TestTableMaintenanceActionMetaService.java b/plugins/tms/src/test/java/org/apache/gravitino/maintenance/storage/service/TestTableMaintenanceActionMetaService.java index ec54a63786..c5c694ae7b 100644 --- a/plugins/tms/src/test/java/org/apache/gravitino/maintenance/storage/service/TestTableMaintenanceActionMetaService.java +++ b/plugins/tms/src/test/java/org/apache/gravitino/maintenance/storage/service/TestTableMaintenanceActionMetaService.java @@ -259,4 +259,34 @@ class TestTableMaintenanceActionMetaService extends TestJDBCBackend { actions.bindJobId(reclaimed.get(), 1002L); Assertions.assertEquals(1002L, actions.getById(reclaimed.get().id()).jobId()); } + + @TestTemplate + void testDeleteUnboundClaimFencesAgainstReclaim() throws Exception { + TableMaintenanceActionMetaService actions = TableMaintenanceActionMetaService.getInstance(); + long tableId = 12L; + long policyId = 22L; + long timeoutMs = 200L; + + Optional<OccupancyClaim> first = actions.tryAcquireInFlight(tableId, policyId, timeoutMs); + Assertions.assertTrue(first.isPresent()); + Assertions.assertTrue(actions.deleteUnboundClaim(first.get())); + Assertions.assertNull(actions.getById(first.get().id())); + + Optional<OccupancyClaim> second = actions.tryAcquireInFlight(tableId, policyId, timeoutMs); + Assertions.assertTrue(second.isPresent()); + Thread.sleep(timeoutMs + 50L); + Optional<OccupancyClaim> reclaimed = actions.tryAcquireInFlight(tableId, policyId, timeoutMs); + Assertions.assertTrue(reclaimed.isPresent()); + Assertions.assertEquals(second.get().id(), reclaimed.get().id()); + Assertions.assertTrue(reclaimed.get().createdAt() >= second.get().createdAt()); + + // Stale submitter cleanup must not delete the reclaimed live claim. + Assertions.assertFalse(actions.deleteUnboundClaim(second.get())); + Assertions.assertNotNull(actions.getById(reclaimed.get().id())); + Assertions.assertNull(actions.getById(reclaimed.get().id()).jobId()); + Assertions.assertEquals(0L, actions.getById(reclaimed.get().id()).finishedAt()); + + Assertions.assertTrue(actions.deleteUnboundClaim(reclaimed.get())); + Assertions.assertNull(actions.getById(reclaimed.get().id())); + } }
