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 f00547368050ba0232d5963f02e8da62f8539345 Author: remilu <[email protected]> AuthorDate: Mon Oct 26 07:55:09 2026 +0800 [#14215] feat(tms): Policy-expand pipeline into cron-job enqueue API Fixes #14215. Squashed from fork #73 (B3). --- .../scheduler/job/cron/CronJobScheduler.java | 46 +++++ .../{ => policy/expand}/ExpandOccupancyGate.java | 2 +- .../{ => policy}/expand/ExpandTableCandidate.java | 2 +- .../expand/NearestMaintenancePolicyResolver.java | 2 +- .../expand/NearestPolicyTableEnumerator.java | 4 +- .../scheduler/policy/expand/PolicyExpandCrons.java | 69 +++++++ .../policy/expand/PolicyExpandPipeline.java | 197 ++++++++++++++++++ .../{ => policy}/expand/PolicyTableEnumerator.java | 4 +- .../TestNearestMaintenancePolicyResolver.java | 2 +- .../expand/TestNearestPolicyTableEnumerator.java | 4 +- .../policy/expand/TestPolicyExpandCrons.java | 58 ++++++ .../policy/expand/TestPolicyExpandPipeline.java | 221 +++++++++++++++++++++ 12 files changed, 601 insertions(+), 10 deletions(-) diff --git a/plugins/tms/src/main/java/org/apache/gravitino/maintenance/scheduler/job/cron/CronJobScheduler.java b/plugins/tms/src/main/java/org/apache/gravitino/maintenance/scheduler/job/cron/CronJobScheduler.java new file mode 100644 index 0000000000..f16cc3febd --- /dev/null +++ b/plugins/tms/src/main/java/org/apache/gravitino/maintenance/scheduler/job/cron/CronJobScheduler.java @@ -0,0 +1,46 @@ +/* + * 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.List; +import org.apache.gravitino.maintenance.scheduler.policy.expand.ExpandTableCandidate; +import org.apache.gravitino.meta.PolicyEntity; + +/** + * Inserts one-shot {@code cron-job} rows into the cron-pool db-scheduler table. + * + * <p>Existing instance ids should be left untouched so a repeated expand pick does not reset an + * already-queued submit. Snapshot-expiry winners attached at schema/catalog are packed into {@code + * batch:{batch_id}:{policy_id}} rows; incomplete batches are filled via {@code + * SchedulerClient#getScheduledExecution} / {@code reschedule} before a new {@code batch_id} is + * allocated (§5.2.2). + */ +@FunctionalInterface +public interface CronJobScheduler { + + /** + * Enqueues due-now cron jobs for the given expand candidates. + * + * @param policy effective maintenance policy + * @param candidates enqueueable tables with winner levels + * @param batchCapacity max tables per snapshot-expiry batch row (and max candidates considered) + * @return number of newly inserted cron-job rows (merges into existing batches do not count) + */ + int enqueueDueNow(PolicyEntity policy, List<ExpandTableCandidate> candidates, int batchCapacity); +} diff --git a/plugins/tms/src/main/java/org/apache/gravitino/maintenance/scheduler/ExpandOccupancyGate.java b/plugins/tms/src/main/java/org/apache/gravitino/maintenance/scheduler/policy/expand/ExpandOccupancyGate.java similarity index 95% rename from plugins/tms/src/main/java/org/apache/gravitino/maintenance/scheduler/ExpandOccupancyGate.java rename to plugins/tms/src/main/java/org/apache/gravitino/maintenance/scheduler/policy/expand/ExpandOccupancyGate.java index 6d36baf0f5..cb58553037 100644 --- a/plugins/tms/src/main/java/org/apache/gravitino/maintenance/scheduler/ExpandOccupancyGate.java +++ b/plugins/tms/src/main/java/org/apache/gravitino/maintenance/scheduler/policy/expand/ExpandOccupancyGate.java @@ -16,7 +16,7 @@ * specific language governing permissions and limitations * under the License. */ -package org.apache.gravitino.maintenance.scheduler; +package org.apache.gravitino.maintenance.scheduler.policy.expand; /** * Gate used by expand before inserting a {@code cron-job} row. diff --git a/plugins/tms/src/main/java/org/apache/gravitino/maintenance/scheduler/expand/ExpandTableCandidate.java b/plugins/tms/src/main/java/org/apache/gravitino/maintenance/scheduler/policy/expand/ExpandTableCandidate.java similarity index 97% rename from plugins/tms/src/main/java/org/apache/gravitino/maintenance/scheduler/expand/ExpandTableCandidate.java rename to plugins/tms/src/main/java/org/apache/gravitino/maintenance/scheduler/policy/expand/ExpandTableCandidate.java index 3311aafb8c..e44a7e5f7a 100644 --- a/plugins/tms/src/main/java/org/apache/gravitino/maintenance/scheduler/expand/ExpandTableCandidate.java +++ b/plugins/tms/src/main/java/org/apache/gravitino/maintenance/scheduler/policy/expand/ExpandTableCandidate.java @@ -16,7 +16,7 @@ * specific language governing permissions and limitations * under the License. */ -package org.apache.gravitino.maintenance.scheduler.expand; +package org.apache.gravitino.maintenance.scheduler.policy.expand; import com.google.common.base.Preconditions; import java.util.Objects; diff --git a/plugins/tms/src/main/java/org/apache/gravitino/maintenance/scheduler/expand/NearestMaintenancePolicyResolver.java b/plugins/tms/src/main/java/org/apache/gravitino/maintenance/scheduler/policy/expand/NearestMaintenancePolicyResolver.java similarity index 98% rename from plugins/tms/src/main/java/org/apache/gravitino/maintenance/scheduler/expand/NearestMaintenancePolicyResolver.java rename to plugins/tms/src/main/java/org/apache/gravitino/maintenance/scheduler/policy/expand/NearestMaintenancePolicyResolver.java index 34fa73a2c1..56f53efd6e 100644 --- a/plugins/tms/src/main/java/org/apache/gravitino/maintenance/scheduler/expand/NearestMaintenancePolicyResolver.java +++ b/plugins/tms/src/main/java/org/apache/gravitino/maintenance/scheduler/policy/expand/NearestMaintenancePolicyResolver.java @@ -16,7 +16,7 @@ * specific language governing permissions and limitations * under the License. */ -package org.apache.gravitino.maintenance.scheduler.expand; +package org.apache.gravitino.maintenance.scheduler.policy.expand; import java.util.ArrayList; import java.util.List; diff --git a/plugins/tms/src/main/java/org/apache/gravitino/maintenance/scheduler/expand/NearestPolicyTableEnumerator.java b/plugins/tms/src/main/java/org/apache/gravitino/maintenance/scheduler/policy/expand/NearestPolicyTableEnumerator.java similarity index 98% rename from plugins/tms/src/main/java/org/apache/gravitino/maintenance/scheduler/expand/NearestPolicyTableEnumerator.java rename to plugins/tms/src/main/java/org/apache/gravitino/maintenance/scheduler/policy/expand/NearestPolicyTableEnumerator.java index d182f3c56c..ae503ec6a0 100644 --- a/plugins/tms/src/main/java/org/apache/gravitino/maintenance/scheduler/expand/NearestPolicyTableEnumerator.java +++ b/plugins/tms/src/main/java/org/apache/gravitino/maintenance/scheduler/policy/expand/NearestPolicyTableEnumerator.java @@ -16,7 +16,7 @@ * specific language governing permissions and limitations * under the License. */ -package org.apache.gravitino.maintenance.scheduler.expand; +package org.apache.gravitino.maintenance.scheduler.policy.expand; import java.io.IOException; import java.io.UncheckedIOException; @@ -33,7 +33,7 @@ import org.apache.gravitino.MetadataObject; import org.apache.gravitino.MetadataObjects; import org.apache.gravitino.NameIdentifier; import org.apache.gravitino.RelationalEntity; -import org.apache.gravitino.maintenance.scheduler.ExpandOccupancyGate; +import org.apache.gravitino.maintenance.scheduler.policy.expand.ExpandOccupancyGate; import org.apache.gravitino.meta.GenericEntity; import org.apache.gravitino.meta.PolicyEntity; import org.apache.gravitino.meta.SchemaEntity; diff --git a/plugins/tms/src/main/java/org/apache/gravitino/maintenance/scheduler/policy/expand/PolicyExpandCrons.java b/plugins/tms/src/main/java/org/apache/gravitino/maintenance/scheduler/policy/expand/PolicyExpandCrons.java new file mode 100644 index 0000000000..2efbde275e --- /dev/null +++ b/plugins/tms/src/main/java/org/apache/gravitino/maintenance/scheduler/policy/expand/PolicyExpandCrons.java @@ -0,0 +1,69 @@ +/* + * 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.policy.expand; + +import com.github.kagkarlsson.scheduler.task.ExecutionComplete; +import com.github.kagkarlsson.scheduler.task.schedule.CronSchedule; +import com.github.kagkarlsson.scheduler.task.schedule.CronStyle; +import com.github.kagkarlsson.scheduler.task.schedule.Schedule; +import com.google.common.annotations.VisibleForTesting; +import java.time.Instant; +import java.time.ZoneId; +import java.time.ZoneOffset; + +/** + * Quartz cron helpers for {@code policy-expand} next-due calculation. + * + * <p>Always evaluates cron expressions in UTC so every scheduler node computes the same next-due + * instant for a given {@code timeTrigger}, regardless of the JVM default time zone. + */ +final class PolicyExpandCrons { + + /** Zone used for all policy-expand cron evaluations. */ + static final ZoneId SCHEDULER_ZONE = ZoneOffset.UTC; + + private PolicyExpandCrons() {} + + /** + * Next fire time for a Quartz cron, strictly after now, in {@link #SCHEDULER_ZONE}. + * + * @param timeTrigger six-field Quartz cron expression + * @return next execution instant + */ + static Instant nextExecution(String timeTrigger) { + return nextExecution(timeTrigger, Instant.now()); + } + + /** + * Next fire time for a Quartz cron, strictly after {@code after}, in {@link #SCHEDULER_ZONE}. + * + * @param timeTrigger six-field Quartz cron expression + * @param after reference instant + * @return next execution instant + */ + @VisibleForTesting + static Instant nextExecution(String timeTrigger, Instant after) { + CronSchedule schedule = new CronSchedule(timeTrigger, SCHEDULER_ZONE, CronStyle.QUARTZ); + Instant next = schedule.getNextExecutionTime(ExecutionComplete.simulatedSuccess(after)); + if (next == null || Schedule.NEVER.equals(next)) { + throw new IllegalArgumentException("timeTrigger has no next execution: " + timeTrigger); + } + return next; + } +} diff --git a/plugins/tms/src/main/java/org/apache/gravitino/maintenance/scheduler/policy/expand/PolicyExpandPipeline.java b/plugins/tms/src/main/java/org/apache/gravitino/maintenance/scheduler/policy/expand/PolicyExpandPipeline.java new file mode 100644 index 0000000000..58dbcedc06 --- /dev/null +++ b/plugins/tms/src/main/java/org/apache/gravitino/maintenance/scheduler/policy/expand/PolicyExpandPipeline.java @@ -0,0 +1,197 @@ +/* + * 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.policy.expand; + +import com.google.common.base.Preconditions; +import java.time.Instant; +import java.util.List; +import java.util.Optional; +import java.util.function.Function; +import javax.annotation.Nullable; +import org.apache.gravitino.Config; +import org.apache.gravitino.maintenance.policy.MaintenanceMinIntervalConfigs; +import org.apache.gravitino.maintenance.policy.MaintenancePolicyDefinitions; +import org.apache.gravitino.maintenance.scheduler.policy.expand.ExpandOccupancyGate; +import org.apache.gravitino.maintenance.scheduler.job.cron.CronJobScheduler; +import org.apache.gravitino.meta.PolicyEntity; +import org.apache.gravitino.policy.ExecutablePolicyContent; +import org.apache.gravitino.policy.Policy; +import org.apache.gravitino.policy.PolicyContent; +import org.apache.gravitino.storage.relational.service.PolicyMetaService; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +/** + * Expands a due {@code policy-expand} lease into one-shot {@code cron-job} rows (§5.2.1). + * + * <p>Does not submit Spark. Returns the next expand due time, or empty when the expand row should + * be removed (missing / disabled / blank trigger / non-maintenance). + * + * <p>Enumeration fills up to {@code enqueueBatchSize} tables that pass nearest-policy and {@code + * minIntervalMs} / in-flight gating (policy content → server conf → code default). + */ +public final class PolicyExpandPipeline { + + private static final Logger LOG = LoggerFactory.getLogger(PolicyExpandPipeline.class); + private static final PolicyMetaService POLICY_SERVICE = PolicyMetaService.getInstance(); + + private final Function<Long, Optional<PolicyEntity>> policyLoader; + private final PolicyTableEnumerator tableEnumerator; + private final CronJobScheduler cronJobScheduler; + private final ExpandOccupancyGate occupancyGate; + private final Config config; + private final int enqueueBatchSize; + + /** + * Loads policies through {@link PolicyMetaService}. + * + * @param tableEnumerator lists attached table ids + * @param cronJobScheduler writes cron-job rows + * @param occupancyGate in-flight / minInterval gate + * @param config gravitino server config for per-type minInterval fallbacks + * @param enqueueBatchSize max cron-jobs per expand pick + */ + public PolicyExpandPipeline( + PolicyTableEnumerator tableEnumerator, + CronJobScheduler cronJobScheduler, + ExpandOccupancyGate occupancyGate, + Config config, + int enqueueBatchSize) { + this( + PolicyExpandPipeline::loadPolicy, + tableEnumerator, + cronJobScheduler, + occupancyGate, + config, + enqueueBatchSize); + } + + /** + * @param policyLoader loads the policy for the expand instance id + * @param tableEnumerator lists attached table ids + * @param cronJobScheduler writes cron-job rows + * @param occupancyGate in-flight / minInterval gate + * @param config gravitino server config for per-type minInterval fallbacks + * @param enqueueBatchSize max cron-jobs per expand pick + */ + PolicyExpandPipeline( + Function<Long, Optional<PolicyEntity>> policyLoader, + PolicyTableEnumerator tableEnumerator, + CronJobScheduler cronJobScheduler, + ExpandOccupancyGate occupancyGate, + Config config, + int enqueueBatchSize) { + this.policyLoader = Preconditions.checkNotNull(policyLoader); + this.tableEnumerator = tableEnumerator; + this.cronJobScheduler = cronJobScheduler; + this.occupancyGate = occupancyGate; + this.config = config; + this.enqueueBatchSize = enqueueBatchSize; + } + + /** + * Runs one expand pick for {@code policyId}. + * + * @param policyId expand task instance id + * @return next due time when the expand row should stay, or empty to delete the row + */ + public Optional<Instant> expand(long policyId) { + Optional<PolicyEntity> loaded = policyLoader.apply(policyId); + if (loaded.isEmpty()) { + LOG.info("Removing expand row for missing policy {}", policyId); + return Optional.empty(); + } + PolicyEntity policy = loaded.get(); + if (!policy.enabled() || !isMaintenance(policy.content())) { + LOG.info( + "Removing expand row for policy {} (enabled={}, maintenance={})", + policyId, + policy.enabled(), + isMaintenance(policy.content())); + return Optional.empty(); + } + String timeTrigger = timeTrigger(policy.content()); + if (timeTrigger == null || timeTrigger.isBlank()) { + LOG.info("Removing expand row for policy {} with blank timeTrigger", policyId); + return Optional.empty(); + } + + // Resolve next due before any cron-job writes so a bad crontab cannot leave orphans. + final Instant nextDue; + try { + nextDue = PolicyExpandCrons.nextExecution(timeTrigger); + } catch (RuntimeException e) { + LOG.warn( + "Removing expand row for policy {} with invalid timeTrigger {}: {}", + policyId, + timeTrigger, + e.toString()); + return Optional.empty(); + } + + long minIntervalMs = + MaintenanceMinIntervalConfigs.resolve( + config, policy.policyType(), policyMinIntervalMs(policy.content())); + List<ExpandTableCandidate> toEnqueue = + tableEnumerator.listTableIds(policy, enqueueBatchSize, occupancyGate, minIntervalMs); + int scheduled = cronJobScheduler.enqueueDueNow(policy, toEnqueue, enqueueBatchSize); + LOG.info( + "Expanded policy {}: scheduled {} cron-job rows for {} tables (type={}, minIntervalMs={}); next expand due {}", + policyId, + scheduled, + toEnqueue.size(), + policy.policyType(), + minIntervalMs, + nextDue); + return Optional.of(nextDue); + } + + private static Optional<PolicyEntity> loadPolicy(long policyId) { + return Optional.ofNullable(POLICY_SERVICE.getPolicyById(policyId)); + } + + private static boolean isMaintenance(@Nullable PolicyContent content) { + if (content == null) { + return false; + } + for (Policy.BuiltInType type : Policy.BuiltInType.values()) { + if (MaintenancePolicyDefinitions.isMaintenance(type) + && type.contentClass().isInstance(content)) { + return true; + } + } + return false; + } + + @Nullable + private static String timeTrigger(@Nullable PolicyContent content) { + if (!(content instanceof ExecutablePolicyContent)) { + return null; + } + return ((ExecutablePolicyContent) content).executionSpec().timeTrigger(); + } + + @Nullable + private static Long policyMinIntervalMs(@Nullable PolicyContent content) { + if (!(content instanceof ExecutablePolicyContent)) { + return null; + } + return ((ExecutablePolicyContent) content).executionSpec().minIntervalMs(); + } +} diff --git a/plugins/tms/src/main/java/org/apache/gravitino/maintenance/scheduler/expand/PolicyTableEnumerator.java b/plugins/tms/src/main/java/org/apache/gravitino/maintenance/scheduler/policy/expand/PolicyTableEnumerator.java similarity index 93% rename from plugins/tms/src/main/java/org/apache/gravitino/maintenance/scheduler/expand/PolicyTableEnumerator.java rename to plugins/tms/src/main/java/org/apache/gravitino/maintenance/scheduler/policy/expand/PolicyTableEnumerator.java index 8d65ee667a..372068686b 100644 --- a/plugins/tms/src/main/java/org/apache/gravitino/maintenance/scheduler/expand/PolicyTableEnumerator.java +++ b/plugins/tms/src/main/java/org/apache/gravitino/maintenance/scheduler/policy/expand/PolicyTableEnumerator.java @@ -16,10 +16,10 @@ * specific language governing permissions and limitations * under the License. */ -package org.apache.gravitino.maintenance.scheduler.expand; +package org.apache.gravitino.maintenance.scheduler.policy.expand; import java.util.List; -import org.apache.gravitino.maintenance.scheduler.ExpandOccupancyGate; +import org.apache.gravitino.maintenance.scheduler.policy.expand.ExpandOccupancyGate; import org.apache.gravitino.meta.PolicyEntity; /** diff --git a/plugins/tms/src/test/java/org/apache/gravitino/maintenance/scheduler/expand/TestNearestMaintenancePolicyResolver.java b/plugins/tms/src/test/java/org/apache/gravitino/maintenance/scheduler/policy/expand/TestNearestMaintenancePolicyResolver.java similarity index 99% rename from plugins/tms/src/test/java/org/apache/gravitino/maintenance/scheduler/expand/TestNearestMaintenancePolicyResolver.java rename to plugins/tms/src/test/java/org/apache/gravitino/maintenance/scheduler/policy/expand/TestNearestMaintenancePolicyResolver.java index 09229b47c6..557093eb99 100644 --- a/plugins/tms/src/test/java/org/apache/gravitino/maintenance/scheduler/expand/TestNearestMaintenancePolicyResolver.java +++ b/plugins/tms/src/test/java/org/apache/gravitino/maintenance/scheduler/policy/expand/TestNearestMaintenancePolicyResolver.java @@ -16,7 +16,7 @@ * specific language governing permissions and limitations * under the License. */ -package org.apache.gravitino.maintenance.scheduler.expand; +package org.apache.gravitino.maintenance.scheduler.policy.expand; import static org.mockito.Mockito.mock; import static org.mockito.Mockito.when; diff --git a/plugins/tms/src/test/java/org/apache/gravitino/maintenance/scheduler/expand/TestNearestPolicyTableEnumerator.java b/plugins/tms/src/test/java/org/apache/gravitino/maintenance/scheduler/policy/expand/TestNearestPolicyTableEnumerator.java similarity index 98% rename from plugins/tms/src/test/java/org/apache/gravitino/maintenance/scheduler/expand/TestNearestPolicyTableEnumerator.java rename to plugins/tms/src/test/java/org/apache/gravitino/maintenance/scheduler/policy/expand/TestNearestPolicyTableEnumerator.java index 4a02e6faae..19a7ba394e 100644 --- a/plugins/tms/src/test/java/org/apache/gravitino/maintenance/scheduler/expand/TestNearestPolicyTableEnumerator.java +++ b/plugins/tms/src/test/java/org/apache/gravitino/maintenance/scheduler/policy/expand/TestNearestPolicyTableEnumerator.java @@ -16,7 +16,7 @@ * specific language governing permissions and limitations * under the License. */ -package org.apache.gravitino.maintenance.scheduler.expand; +package org.apache.gravitino.maintenance.scheduler.policy.expand; import static org.mockito.ArgumentMatchers.any; import static org.mockito.ArgumentMatchers.eq; @@ -30,7 +30,7 @@ import org.apache.gravitino.MetadataObject; import org.apache.gravitino.MetadataObjects; import org.apache.gravitino.RelationalEntity; import org.apache.gravitino.SupportsRelationOperations; -import org.apache.gravitino.maintenance.scheduler.ExpandOccupancyGate; +import org.apache.gravitino.maintenance.scheduler.policy.expand.ExpandOccupancyGate; import org.apache.gravitino.meta.AuditInfo; import org.apache.gravitino.meta.GenericEntity; import org.apache.gravitino.meta.PolicyEntity; diff --git a/plugins/tms/src/test/java/org/apache/gravitino/maintenance/scheduler/policy/expand/TestPolicyExpandCrons.java b/plugins/tms/src/test/java/org/apache/gravitino/maintenance/scheduler/policy/expand/TestPolicyExpandCrons.java new file mode 100644 index 0000000000..f3f17b6348 --- /dev/null +++ b/plugins/tms/src/test/java/org/apache/gravitino/maintenance/scheduler/policy/expand/TestPolicyExpandCrons.java @@ -0,0 +1,58 @@ +/* + * 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.policy.expand; + +import java.time.Instant; +import java.time.ZoneId; +import java.time.ZoneOffset; +import java.util.TimeZone; +import org.junit.jupiter.api.Assertions; +import org.junit.jupiter.api.Test; + +/** Unit tests for {@link PolicyExpandCrons}. */ +class TestPolicyExpandCrons { + + @Test + void testNextExecutionUsesUtcIndependentOfJvmDefaultZone() { + // Daily at 01:00 — wall clock depends on the CronSchedule zone, not the JVM default. + String dailyAtOne = "0 0 1 * * ?"; + Instant after = Instant.parse("2026-06-15T00:30:00Z"); + + TimeZone previous = TimeZone.getDefault(); + try { + TimeZone.setDefault(TimeZone.getTimeZone(ZoneId.of("America/Los_Angeles"))); + Instant fromUs = PolicyExpandCrons.nextExecution(dailyAtOne, after); + + TimeZone.setDefault(TimeZone.getTimeZone(ZoneId.of("Asia/Shanghai"))); + Instant fromCn = PolicyExpandCrons.nextExecution(dailyAtOne, after); + + Assertions.assertEquals(fromUs, fromCn); + Assertions.assertEquals(Instant.parse("2026-06-15T01:00:00Z"), fromUs); + Assertions.assertEquals(ZoneOffset.UTC, PolicyExpandCrons.SCHEDULER_ZONE); + } finally { + TimeZone.setDefault(previous); + } + } + + @Test + void testInvalidCronRejected() { + Assertions.assertThrows( + IllegalArgumentException.class, () -> PolicyExpandCrons.nextExecution("not-a-cron")); + } +} diff --git a/plugins/tms/src/test/java/org/apache/gravitino/maintenance/scheduler/policy/expand/TestPolicyExpandPipeline.java b/plugins/tms/src/test/java/org/apache/gravitino/maintenance/scheduler/policy/expand/TestPolicyExpandPipeline.java new file mode 100644 index 0000000000..b5027c34f2 --- /dev/null +++ b/plugins/tms/src/test/java/org/apache/gravitino/maintenance/scheduler/policy/expand/TestPolicyExpandPipeline.java @@ -0,0 +1,221 @@ +/* + * 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.policy.expand; + +import static org.mockito.ArgumentMatchers.any; +import static org.mockito.ArgumentMatchers.anyInt; +import static org.mockito.ArgumentMatchers.anyLong; +import static org.mockito.ArgumentMatchers.eq; +import static org.mockito.Mockito.mock; +import static org.mockito.Mockito.never; +import static org.mockito.Mockito.verify; +import static org.mockito.Mockito.when; + +import java.time.Instant; +import java.util.List; +import java.util.Optional; +import java.util.function.Function; +import org.apache.gravitino.Config; +import org.apache.gravitino.MetadataObject; +import org.apache.gravitino.Namespace; +import org.apache.gravitino.maintenance.policy.MaintenanceMinIntervalConfigs; +import org.apache.gravitino.maintenance.scheduler.policy.expand.ExpandOccupancyGate; +import org.apache.gravitino.maintenance.scheduler.job.cron.CronJobScheduler; +import org.apache.gravitino.meta.AuditInfo; +import org.apache.gravitino.meta.PolicyEntity; +import org.apache.gravitino.policy.IcebergDataCompactionContent; +import org.apache.gravitino.policy.Policy; +import org.apache.gravitino.policy.PolicyContents; +import org.apache.gravitino.utils.NamespaceUtil; +import org.junit.jupiter.api.Assertions; +import org.junit.jupiter.api.Test; + +class TestPolicyExpandPipeline { + + private static final String CRON = "0 0 2 * * ?"; + + @Test + void testMissingPolicyRemovesExpandRow() { + CronJobScheduler cronJobs = mock(CronJobScheduler.class); + PolicyExpandPipeline pipeline = + pipeline(id -> Optional.empty(), mock(PolicyTableEnumerator.class), cronJobs, allowAll()); + + Assertions.assertTrue(pipeline.expand(1L).isEmpty()); + verify(cronJobs, never()).enqueueDueNow(any(), any(), anyInt()); + } + + @Test + void testDisabledPolicyRemovesExpandRow() { + CronJobScheduler cronJobs = mock(CronJobScheduler.class); + PolicyExpandPipeline pipeline = + pipeline( + lookup(2L, policy(false, CRON, null)), + mock(PolicyTableEnumerator.class), + cronJobs, + allowAll()); + + Assertions.assertTrue(pipeline.expand(2L).isEmpty()); + verify(cronJobs, never()).enqueueDueNow(any(), any(), anyInt()); + } + + @Test + void testBlankTimeTriggerRemovesExpandRow() { + CronJobScheduler cronJobs = mock(CronJobScheduler.class); + PolicyExpandPipeline pipeline = + pipeline( + lookup(3L, policy(true, " ", null)), + mock(PolicyTableEnumerator.class), + cronJobs, + allowAll()); + + Assertions.assertTrue(pipeline.expand(3L).isEmpty()); + verify(cronJobs, never()).enqueueDueNow(any(), any(), anyInt()); + } + + @Test + void testInvalidTimeTriggerRemovesExpandRowWithoutEnqueue() { + PolicyTableEnumerator tables = mock(PolicyTableEnumerator.class); + CronJobScheduler cronJobs = mock(CronJobScheduler.class); + PolicyExpandPipeline pipeline = + pipeline(lookup(4L, policy(true, "not-a-cron", null)), tables, cronJobs, allowAll()); + + Assertions.assertTrue(pipeline.expand(4L).isEmpty()); + verify(tables, never()).listTableIds(any(), anyInt(), any(), anyLong()); + verify(cronJobs, never()).enqueueDueNow(any(), any(), anyInt()); + } + + @Test + void testExpandEnqueuesCronJobsAndReturnsNextDue() { + PolicyEntity policy = policy(true, CRON, null); + ExpandOccupancyGate gate = allowAll(); + PolicyTableEnumerator tables = mock(PolicyTableEnumerator.class); + when(tables.listTableIds( + eq(policy), + eq(2), + eq(gate), + eq(MaintenanceMinIntervalConfigs.DEFAULT_HOURLY_MIN_INTERVAL_MS))) + .thenReturn(List.of(c(10L), c(11L))); + CronJobScheduler cronJobs = mock(CronJobScheduler.class); + when(cronJobs.enqueueDueNow(eq(policy), eq(List.of(c(10L), c(11L))), eq(2))).thenReturn(2); + PolicyExpandPipeline expandPipeline = pipeline(lookup(42L, policy), tables, cronJobs, gate, 2); + + Optional<Instant> next = expandPipeline.expand(42L); + Assertions.assertTrue(next.isPresent()); + Assertions.assertTrue(next.get().isAfter(Instant.now().minusSeconds(1))); + verify(cronJobs).enqueueDueNow(policy, List.of(c(10L), c(11L)), 2); + } + + @Test + void testPassesResolvedMinIntervalToEnumerator() { + PolicyEntity policy = policy(true, CRON, 60_000L); + ExpandOccupancyGate gate = allowAll(); + PolicyTableEnumerator tables = mock(PolicyTableEnumerator.class); + when(tables.listTableIds(eq(policy), eq(10), eq(gate), eq(60_000L))) + .thenReturn(List.of(c(11L))); + CronJobScheduler cronJobs = mock(CronJobScheduler.class); + when(cronJobs.enqueueDueNow(eq(policy), eq(List.of(c(11L))), eq(10))).thenReturn(1); + + PolicyExpandPipeline expandPipeline = pipeline(lookup(42L, policy), tables, cronJobs, gate, 10); + Assertions.assertTrue(expandPipeline.expand(42L).isPresent()); + verify(tables).listTableIds(eq(policy), eq(10), eq(gate), eq(60_000L)); + verify(cronJobs).enqueueDueNow(policy, List.of(c(11L)), 10); + } + + @Test + void testUsesServerDefaultMinIntervalWhenPolicyOmits() { + PolicyEntity policy = policy(true, CRON, null); + ExpandOccupancyGate gate = allowAll(); + PolicyTableEnumerator tables = mock(PolicyTableEnumerator.class); + when(tables.listTableIds( + eq(policy), + eq(10), + eq(gate), + eq(MaintenanceMinIntervalConfigs.DEFAULT_HOURLY_MIN_INTERVAL_MS))) + .thenReturn(List.of(c(10L))); + CronJobScheduler cronJobs = mock(CronJobScheduler.class); + when(cronJobs.enqueueDueNow(eq(policy), eq(List.of(c(10L))), eq(10))).thenReturn(1); + + PolicyExpandPipeline expandPipeline = pipeline(lookup(42L, policy), tables, cronJobs, gate, 10); + Assertions.assertTrue(expandPipeline.expand(42L).isPresent()); + verify(tables) + .listTableIds( + eq(policy), + eq(10), + eq(gate), + eq(MaintenanceMinIntervalConfigs.DEFAULT_HOURLY_MIN_INTERVAL_MS)); + } + + private static PolicyExpandPipeline pipeline( + Function<Long, Optional<PolicyEntity>> policyLoader, + PolicyTableEnumerator tables, + CronJobScheduler cronJobs, + ExpandOccupancyGate gate) { + return pipeline(policyLoader, tables, cronJobs, gate, 100); + } + + private static PolicyExpandPipeline pipeline( + Function<Long, Optional<PolicyEntity>> policyLoader, + PolicyTableEnumerator tables, + CronJobScheduler cronJobs, + ExpandOccupancyGate gate, + int batchSize) { + return new PolicyExpandPipeline( + policyLoader, tables, cronJobs, gate, new Config(false) {}, batchSize); + } + + private static ExpandTableCandidate c(long tableId) { + return new ExpandTableCandidate(tableId, MetadataObject.Type.TABLE); + } + + private static ExpandOccupancyGate allowAll() { + return (tableId, policyId, minIntervalMs, nowMs) -> false; + } + + private static Function<Long, Optional<PolicyEntity>> lookup(long policyId, PolicyEntity policy) { + return id -> id == policyId ? Optional.of(policy) : Optional.empty(); + } + + private static PolicyEntity policy(boolean enabled, String timeTrigger, Long minIntervalMs) { + Namespace ns = NamespaceUtil.ofPolicy("metalake"); + return PolicyEntity.builder() + .withId(42L) + .withName("compaction") + .withNamespace(ns) + .withPolicyType(Policy.BuiltInType.ICEBERG_COMPACTION) + .withEnabled(enabled) + .withContent( + PolicyContents.icebergDataCompaction( + IcebergDataCompactionContent.DEFAULT_MIN_DATA_FILE_MSE, + IcebergDataCompactionContent.DEFAULT_MIN_DELETE_FILE_NUMBER, + IcebergDataCompactionContent.DEFAULT_DATA_FILE_MSE_WEIGHT, + IcebergDataCompactionContent.DEFAULT_DELETE_FILE_NUMBER_WEIGHT, + IcebergDataCompactionContent.DEFAULT_MAX_PARTITION_NUM, + IcebergDataCompactionContent.DEFAULT_REWRITE_STRATEGY, + IcebergDataCompactionContent.DEFAULT_SORT_ORDER, + null, + timeTrigger, + null, + minIntervalMs, + null, + null)) + .withAuditInfo( + AuditInfo.builder().withCreator("test").withCreateTime(Instant.now()).build()) + .build(); + } +}
