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 85725e52fea5e92a9f96db618939d4824817bc47 Author: remilu <[email protected]> AuthorDate: Mon Oct 26 10:27:44 2026 +0800 [#14222] test(tms): Add BaseIT expand→cron and IRC→commit enqueue coverage Exercise live scheduler pools / EventBus with an embedded Iceberg catalog. Co-authored-by: Cursor <[email protected]> --- plugins/maintenance/build.gradle.kts | 16 ++ .../job/commit/RelationalCommitJobScheduler.java | 7 +- .../job/cron/RelationalCronJobScheduler.java | 4 +- .../integration/test/IrcToCommitEnqueueIT.java | 176 +++++++++++++++++++++ .../test/MaintenanceSchedulerITSupport.java | 139 ++++++++++++++++ .../integration/test/PolicyExpandToCronIT.java | 163 +++++++++++++++++++ .../policy/TestPolicyExpandRegistrar.java | 26 ++- .../scheduler/SchedulerPoolTestHooks.java | 51 ++++++ .../policy/expand/TestPolicyExpandCrons.java | 26 +-- 9 files changed, 576 insertions(+), 32 deletions(-) diff --git a/plugins/maintenance/build.gradle.kts b/plugins/maintenance/build.gradle.kts index 8a238ceb36..c7464ca27f 100644 --- a/plugins/maintenance/build.gradle.kts +++ b/plugins/maintenance/build.gradle.kts @@ -46,10 +46,12 @@ dependencies { testImplementation(project(":clients:client-java")) testImplementation(project(":core", "testArtifacts")) + testImplementation(project(":iceberg:iceberg-rest-server")) testImplementation(project(":integration-test-common", "testArtifacts")) testImplementation(project(":server")) testImplementation(project(":server-common")) testImplementation(libs.awaitility) + testImplementation(libs.bundles.iceberg) testImplementation(libs.junit.jupiter.api) testImplementation(libs.junit.jupiter.params) testImplementation(libs.mockito.core) @@ -90,8 +92,22 @@ tasks { dependsOn(runtimeJars) } + // testImplementation(:iceberg:iceberg-rest-server) resolves the jar under build/libs, which + // that module also fills via copyDepends (finalizedBy jar). Declare copyDepends so + // compileTestJava does not race it. + compileTestJava { + dependsOn(":iceberg:iceberg-rest-server:copyDepends") + } + test { useJUnitPlatform() environment("GRAVITINO_HOME", rootDir.path) + + val skipITs = project.hasProperty("skipITs") + if (skipITs) { + exclude("**/integration/test/**") + } else { + dependsOn(":iceberg:iceberg-rest-server:jar") + } } } diff --git a/plugins/maintenance/src/main/java/org/apache/gravitino/maintenance/scheduler/job/commit/RelationalCommitJobScheduler.java b/plugins/maintenance/src/main/java/org/apache/gravitino/maintenance/scheduler/job/commit/RelationalCommitJobScheduler.java index 4eb5524f88..cbe8b5c6ad 100644 --- a/plugins/maintenance/src/main/java/org/apache/gravitino/maintenance/scheduler/job/commit/RelationalCommitJobScheduler.java +++ b/plugins/maintenance/src/main/java/org/apache/gravitino/maintenance/scheduler/job/commit/RelationalCommitJobScheduler.java @@ -22,6 +22,7 @@ import com.github.kagkarlsson.scheduler.SchedulerClient; import com.github.kagkarlsson.scheduler.serializer.JacksonSerializer; import com.github.kagkarlsson.scheduler.task.TaskInstance; import com.github.kagkarlsson.scheduler.task.TaskInstanceId; +import com.google.common.annotations.VisibleForTesting; import com.google.common.base.Preconditions; import java.time.Instant; import java.util.ArrayList; @@ -85,7 +86,8 @@ public final class RelationalCommitJobScheduler implements CommitJobScheduler { * @param chainCollector returns the collected onCommit chain for a table id * @param occupancyGate cooldown / in-flight checks for enqueue gating */ - RelationalCommitJobScheduler( + @VisibleForTesting + public RelationalCommitJobScheduler( DataSource dataSource, Config gravitinoConfig, Function<Long, Optional<CollectedCommitChain>> chainCollector, @@ -125,7 +127,8 @@ public final class RelationalCommitJobScheduler implements CommitJobScheduler { * @param dataSource entity-store pool * @return client bound to {@link TableMaintenanceSchedulerPools#COMMIT_TASKS_TABLE} */ - static SchedulerClient client(DataSource dataSource) { + @VisibleForTesting + public static SchedulerClient client(DataSource dataSource) { return SchedulerClient.Builder.create( AutocommitDataSources.wrap(dataSource), TableMaintenanceSchedulerPools.commitJobTask()) .serializer(new JacksonSerializer(JsonUtils.anyFieldMapper())) diff --git a/plugins/maintenance/src/main/java/org/apache/gravitino/maintenance/scheduler/job/cron/RelationalCronJobScheduler.java b/plugins/maintenance/src/main/java/org/apache/gravitino/maintenance/scheduler/job/cron/RelationalCronJobScheduler.java index 863025715c..538724e47e 100644 --- a/plugins/maintenance/src/main/java/org/apache/gravitino/maintenance/scheduler/job/cron/RelationalCronJobScheduler.java +++ b/plugins/maintenance/src/main/java/org/apache/gravitino/maintenance/scheduler/job/cron/RelationalCronJobScheduler.java @@ -23,6 +23,7 @@ import com.github.kagkarlsson.scheduler.SchedulerClient; import com.github.kagkarlsson.scheduler.serializer.JacksonSerializer; import com.github.kagkarlsson.scheduler.task.TaskInstance; import com.github.kagkarlsson.scheduler.task.TaskInstanceId; +import com.google.common.annotations.VisibleForTesting; import com.google.common.base.Preconditions; import java.time.Instant; import java.util.ArrayList; @@ -128,7 +129,8 @@ public final class RelationalCronJobScheduler implements CronJobScheduler { * @param dataSource entity-store pool * @return client bound to {@link TableMaintenanceSchedulerPools#CRON_TASKS_TABLE} */ - static SchedulerClient client(DataSource dataSource) { + @VisibleForTesting + public static SchedulerClient client(DataSource dataSource) { return SchedulerClient.Builder.create( AutocommitDataSources.wrap(dataSource), TableMaintenanceSchedulerPools.cronJobTask()) .serializer(new JacksonSerializer(JsonUtils.anyFieldMapper())) diff --git a/plugins/maintenance/src/test/java/org/apache/gravitino/maintenance/integration/test/IrcToCommitEnqueueIT.java b/plugins/maintenance/src/test/java/org/apache/gravitino/maintenance/integration/test/IrcToCommitEnqueueIT.java new file mode 100644 index 0000000000..3eb15ac84d --- /dev/null +++ b/plugins/maintenance/src/test/java/org/apache/gravitino/maintenance/integration/test/IrcToCommitEnqueueIT.java @@ -0,0 +1,176 @@ +/* + * 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.integration.test; + +import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertNotNull; +import static org.junit.jupiter.api.Assertions.assertTrue; +import static org.junit.jupiter.api.Assumptions.assumeTrue; +import static org.mockito.Mockito.mock; +import static org.mockito.Mockito.when; + +import com.google.common.collect.ImmutableMap; +import com.google.common.collect.Maps; +import java.util.Collections; +import java.util.Enumeration; +import java.util.List; +import java.util.Map; +import java.util.UUID; +import java.util.concurrent.TimeUnit; +import java.util.concurrent.atomic.AtomicReference; +import javax.servlet.http.HttpServletRequest; +import org.apache.gravitino.Configs; +import org.apache.gravitino.GravitinoEnv; +import org.apache.gravitino.NameIdentifier; +import org.apache.gravitino.integration.test.util.BaseIT; +import org.apache.gravitino.integration.test.util.ITUtils; +import org.apache.gravitino.listener.EventListenerManager; +import org.apache.gravitino.listener.api.event.IcebergRequestContext; +import org.apache.gravitino.listener.api.event.IcebergUpdateTableEvent; +import org.apache.gravitino.maintenance.TableMaintenanceConfig; +import org.apache.gravitino.maintenance.integration.test.MaintenanceSchedulerITSupport.SeededTable; +import org.apache.gravitino.maintenance.listener.PostCommitMaintenanceListener; +import org.apache.gravitino.maintenance.policy.MaintenanceAuthConfigs; +import org.apache.gravitino.maintenance.scheduler.MaintenanceJobSpec; +import org.apache.gravitino.maintenance.scheduler.SchedulerPoolTestHooks; +import org.apache.gravitino.maintenance.web.rest.feature.TableMaintenanceRESTFeature; +import org.apache.iceberg.PartitionSpec; +import org.apache.iceberg.Schema; +import org.apache.iceberg.TableMetadata; +import org.apache.iceberg.rest.requests.UpdateTableRequest; +import org.apache.iceberg.rest.responses.LoadTableResponse; +import org.apache.iceberg.types.Types; +import org.junit.jupiter.api.BeforeAll; +import org.junit.jupiter.api.Test; + +/** + * Boots Gravitino with {@link TableMaintenanceRESTFeature} and {@link + * PostCommitMaintenanceListener}, then dispatches an IRC table-update event and asserts commit-job + * enqueue. + * + * <p>Embedded mode only (same-JVM EventBus / MetaService). Run with {@code -PskipITs} omitted. + */ +public class IrcToCommitEnqueueIT extends BaseIT { + + private static final String LISTENER_NAME = "tmsIrc"; + + private static final AtomicReference<String> CAPTURED_COMMIT_ID = new AtomicReference<>(); + private static final AtomicReference<MaintenanceJobSpec> CAPTURED_COMMIT_SPEC = + new AtomicReference<>(); + + private static String metalake; + + @BeforeAll + public void startIntegrationTest() throws Exception { + Map<String, String> configs = Maps.newHashMap(); + configs.put(Configs.ENABLE_AUTHORIZATION.getKey(), String.valueOf(false)); + configs.put(Configs.CACHE_ENABLED.getKey(), String.valueOf(false)); + configs.put( + Configs.REST_API_EXTENSION_PACKAGES.getKey(), + TableMaintenanceRESTFeature.EXTENSION_PACKAGE); + configs.put(TableMaintenanceConfig.SCHEDULER_POLLING_INTERVAL_MS.getKey(), "100"); + configs.put( + MaintenanceAuthConfigs.GRAVITINO_MAINTENANCE_GRAVITINO_AUTH_URI.getKey(), + "http://127.0.0.1:8090";); + configs.put( + EventListenerManager.GRAVITINO_EVENT_LISTENER_PREFIX + + EventListenerManager.GRAVITINO_EVENT_LISTENER_NAMES, + LISTENER_NAME); + configs.put( + EventListenerManager.GRAVITINO_EVENT_LISTENER_PREFIX + + LISTENER_NAME + + "." + + EventListenerManager.GRAVITINO_EVENT_LISTENER_CLASS, + PostCommitMaintenanceListener.class.getName()); + registerCustomConfigs(configs); + super.startIntegrationTest(); + + SchedulerPoolTestHooks.installCommitSubmit( + (instanceId, spec) -> { + CAPTURED_COMMIT_ID.set(instanceId); + CAPTURED_COMMIT_SPEC.set(spec); + }); + + metalake = "ml_irc_" + UUID.randomUUID().toString().replace("-", "").substring(0, 8); + client.createMetalake(metalake, "IRC→commit IT", ImmutableMap.of()); + } + + @Test + void testIrcUpdateEventEnqueuesCommitJob() throws Exception { + assumeTrue( + ITUtils.EMBEDDED_TEST_MODE.equals(testMode), + "Asserts commit submit capture in the same JVM as the EventBus listener"); + + CAPTURED_COMMIT_ID.set(null); + CAPTURED_COMMIT_SPEC.set(null); + + SeededTable seeded = + MaintenanceSchedulerITSupport.seedTaggedCompactionTable( + client.loadMetalake(metalake), + metalake, + /*timeTrigger=*/ null, + /*commitTrigger=*/ true); + + NameIdentifier tableId = + NameIdentifier.of(metalake, seeded.catalogName, seeded.schemaName, seeded.tableName); + GravitinoEnv.getInstance().eventBus().dispatchEvent(icebergUpdateTableEvent(tableId)); + + awaitCommitCapture(Long.toString(seeded.tableId), 30L); + + MaintenanceJobSpec spec = CAPTURED_COMMIT_SPEC.get(); + assertNotNull(spec); + assertTrue(spec.isValid()); + assertEquals(List.of(seeded.tableId), spec.tableIds()); + assertEquals(seeded.policyId, spec.getOperations().get(0).getPolicyId()); + assertTrue(spec.getJobConf().containsKey("updater_options")); + } + + private static IcebergUpdateTableEvent icebergUpdateTableEvent(NameIdentifier tableId) { + HttpServletRequest request = mock(HttpServletRequest.class); + Enumeration<String> headerNames = Collections.enumeration(Collections.emptyList()); + when(request.getRemoteHost()).thenReturn("localhost"); + when(request.getHeaderNames()).thenReturn(headerNames); + + Schema schema = new Schema(Types.NestedField.required(1, "id", Types.LongType.get())); + TableMetadata metadata = + TableMetadata.newTableMetadata( + schema, PartitionSpec.unpartitioned(), "file:///tmp/tms-irc-it", ImmutableMap.of()); + return new IcebergUpdateTableEvent( + new IcebergRequestContext(request, tableId.namespace().level(1)), + tableId, + new UpdateTableRequest(Collections.emptyList(), Collections.emptyList()), + LoadTableResponse.builder().withTableMetadata(metadata).build()); + } + + private static void awaitCommitCapture(String expectedInstanceId, long timeoutSeconds) + throws InterruptedException { + long deadline = System.nanoTime() + TimeUnit.SECONDS.toNanos(timeoutSeconds); + while (System.nanoTime() < deadline) { + if (expectedInstanceId.equals(CAPTURED_COMMIT_ID.get()) + && CAPTURED_COMMIT_SPEC.get() != null) { + return; + } + Thread.sleep(50L); + } + assertEquals( + expectedInstanceId, + CAPTURED_COMMIT_ID.get(), + "IRC listener did not enqueue expected commit-job within " + timeoutSeconds + "s"); + } +} diff --git a/plugins/maintenance/src/test/java/org/apache/gravitino/maintenance/integration/test/MaintenanceSchedulerITSupport.java b/plugins/maintenance/src/test/java/org/apache/gravitino/maintenance/integration/test/MaintenanceSchedulerITSupport.java new file mode 100644 index 0000000000..b496fbaa9c --- /dev/null +++ b/plugins/maintenance/src/test/java/org/apache/gravitino/maintenance/integration/test/MaintenanceSchedulerITSupport.java @@ -0,0 +1,139 @@ +/* + * 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.integration.test; + +import com.google.common.collect.ImmutableMap; +import java.util.HashMap; +import java.util.Map; +import java.util.UUID; +import javax.annotation.Nullable; +import org.apache.gravitino.Catalog; +import org.apache.gravitino.NameIdentifier; +import org.apache.gravitino.client.GravitinoMetalake; +import org.apache.gravitino.meta.PolicyEntity; +import org.apache.gravitino.meta.TableEntity; +import org.apache.gravitino.policy.AllValuesSelector; +import org.apache.gravitino.policy.IcebergDataCompactionContent; +import org.apache.gravitino.policy.Policy; +import org.apache.gravitino.policy.PolicyContents; +import org.apache.gravitino.rel.Column; +import org.apache.gravitino.rel.types.Types; +import org.apache.gravitino.storage.relational.service.PolicyMetaService; +import org.apache.gravitino.storage.relational.service.TableMetaService; +import org.apache.gravitino.utils.NameIdentifierUtil; + +/** + * Shared seed helpers for maintenance scheduler BaseITs: Iceberg catalog (H2 JDBC, no Docker), tag, + * policy, and table association. + */ +final class MaintenanceSchedulerITSupport { + + private MaintenanceSchedulerITSupport() {} + + static SeededTable seedTaggedCompactionTable( + GravitinoMetalake metalake, + String metalakeName, + @Nullable String timeTrigger, + @Nullable Boolean commitTrigger) + throws Exception { + String catalogName = "ice_" + UUID.randomUUID().toString().replace("-", "").substring(0, 8); + String schemaName = "db"; + String tableName = "t"; + String tagName = "tms_" + UUID.randomUUID().toString().replace("-", "").substring(0, 8); + String policyName = "pol_" + UUID.randomUUID().toString().replace("-", "").substring(0, 8); + + Map<String, String> catalogProperties = new HashMap<>(); + catalogProperties.put("catalog-backend", "jdbc"); + catalogProperties.put("warehouse", "/tmp/" + metalakeName + "_" + catalogName); + catalogProperties.put( + "uri", "jdbc:h2:mem:" + metalakeName + "_" + catalogName + ";DB_CLOSE_DELAY=-1;MODE=MYSQL"); + catalogProperties.put("jdbc-driver", "org.h2.Driver"); + catalogProperties.put("jdbc-initialize", "true"); + + Catalog catalog = + metalake.createCatalog( + catalogName, + Catalog.Type.RELATIONAL, + "lakehouse-iceberg", + "maintenance scheduler IT", + catalogProperties); + catalog.asSchemas().createSchema(schemaName, "schema", ImmutableMap.of()); + catalog + .asTableCatalog() + .createTable( + NameIdentifier.of(schemaName, tableName), + new Column[] {Column.of("id", Types.IntegerType.get())}, + "table", + ImmutableMap.of()); + + metalake.createTag(tagName, "maintenance IT tag", ImmutableMap.of()); + metalake.createPolicy( + policyName, + Policy.BuiltInType.ICEBERG_COMPACTION.policyType(), + "maintenance scheduler IT policy", + true, + 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, + Map.of(), + timeTrigger, + commitTrigger, + /*minIntervalMs=*/ 0L, + Map.of("spark_executor_memory", "2g"), + Map.of())); + metalake.addPolicyForTag(tagName, policyName, AllValuesSelector.get()); + + catalog + .asTableCatalog() + .loadTable(NameIdentifier.of(schemaName, tableName)) + .supportsTags() + .associateTags(new String[] {tagName}, null); + + TableEntity table = + TableMetaService.getInstance() + .getTableByIdentifier( + NameIdentifierUtil.ofTable(metalakeName, catalogName, schemaName, tableName)); + PolicyEntity policy = + PolicyMetaService.getInstance() + .getPolicyByIdentifier(NameIdentifierUtil.ofPolicy(metalakeName, policyName)); + return new SeededTable(table.id(), policy.id(), catalogName, schemaName, tableName); + } + + static final class SeededTable { + final long tableId; + final long policyId; + final String catalogName; + final String schemaName; + final String tableName; + + SeededTable( + long tableId, long policyId, String catalogName, String schemaName, String tableName) { + this.tableId = tableId; + this.policyId = policyId; + this.catalogName = catalogName; + this.schemaName = schemaName; + this.tableName = tableName; + } + } +} diff --git a/plugins/maintenance/src/test/java/org/apache/gravitino/maintenance/integration/test/PolicyExpandToCronIT.java b/plugins/maintenance/src/test/java/org/apache/gravitino/maintenance/integration/test/PolicyExpandToCronIT.java new file mode 100644 index 0000000000..e2700be835 --- /dev/null +++ b/plugins/maintenance/src/test/java/org/apache/gravitino/maintenance/integration/test/PolicyExpandToCronIT.java @@ -0,0 +1,163 @@ +/* + * 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.integration.test; + +import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertNotNull; +import static org.junit.jupiter.api.Assertions.assertTrue; +import static org.junit.jupiter.api.Assumptions.assumeTrue; + +import com.google.common.collect.ImmutableMap; +import com.google.common.collect.Maps; +import java.time.Instant; +import java.util.List; +import java.util.Map; +import java.util.UUID; +import java.util.concurrent.TimeUnit; +import java.util.concurrent.atomic.AtomicReference; +import javax.sql.DataSource; +import org.apache.gravitino.Configs; +import org.apache.gravitino.integration.test.util.BaseIT; +import org.apache.gravitino.integration.test.util.ITUtils; +import org.apache.gravitino.maintenance.TableMaintenanceConfig; +import org.apache.gravitino.maintenance.integration.test.MaintenanceSchedulerITSupport.SeededTable; +import org.apache.gravitino.maintenance.occupancy.OccupancyGate; +import org.apache.gravitino.maintenance.policy.MaintenanceAuthConfigs; +import org.apache.gravitino.maintenance.policy.PolicyExpandRegistrar; +import org.apache.gravitino.maintenance.scheduler.MaintenanceJobSpec; +import org.apache.gravitino.maintenance.scheduler.SchedulerPoolTestHooks; +import org.apache.gravitino.maintenance.scheduler.job.cron.CronJobIds; +import org.apache.gravitino.maintenance.scheduler.policy.expand.ExpandTableCandidate; +import org.apache.gravitino.maintenance.scheduler.policy.expand.NearestPolicyTableEnumerator; +import org.apache.gravitino.maintenance.web.rest.feature.TableMaintenanceRESTFeature; +import org.apache.gravitino.meta.PolicyEntity; +import org.apache.gravitino.storage.relational.service.PolicyMetaService; +import org.apache.gravitino.storage.relational.session.SqlSessionFactoryHelper; +import org.junit.jupiter.api.BeforeAll; +import org.junit.jupiter.api.Test; + +/** + * Boots Gravitino with {@link TableMaintenanceRESTFeature} and exercises policy-expand → cron-job + * enqueue through the live expand pool and nearest-policy table walk. + * + * <p>Embedded mode only (same-JVM MetaService / SchedulerClient). Run with {@code -PskipITs} + * omitted. + */ +public class PolicyExpandToCronIT extends BaseIT { + + private static final String TIME_TRIGGER = "0 0 2 * * ?"; + + private static final AtomicReference<String> CAPTURED_CRON_ID = new AtomicReference<>(); + private static final AtomicReference<MaintenanceJobSpec> CAPTURED_CRON_SPEC = + new AtomicReference<>(); + + private static String metalake; + + @BeforeAll + public void startIntegrationTest() throws Exception { + Map<String, String> configs = Maps.newHashMap(); + configs.put(Configs.ENABLE_AUTHORIZATION.getKey(), String.valueOf(false)); + configs.put(Configs.CACHE_ENABLED.getKey(), String.valueOf(false)); + configs.put( + Configs.REST_API_EXTENSION_PACKAGES.getKey(), + TableMaintenanceRESTFeature.EXTENSION_PACKAGE); + configs.put(TableMaintenanceConfig.SCHEDULER_POLLING_INTERVAL_MS.getKey(), "100"); + configs.put( + MaintenanceAuthConfigs.GRAVITINO_MAINTENANCE_GRAVITINO_AUTH_URI.getKey(), + "http://127.0.0.1:8090";); + registerCustomConfigs(configs); + super.startIntegrationTest(); + + SchedulerPoolTestHooks.installCronSubmit( + (instanceId, spec) -> { + CAPTURED_CRON_ID.set(instanceId); + CAPTURED_CRON_SPEC.set(spec); + }); + + metalake = "ml_expand_" + UUID.randomUUID().toString().replace("-", "").substring(0, 8); + client.createMetalake(metalake, "expand→cron IT", ImmutableMap.of()); + } + + @Test + void testExpandPoolEnqueuesCronJobForTaggedTable() throws Exception { + assumeTrue( + ITUtils.EMBEDDED_TEST_MODE.equals(testMode), + "Asserts cron submit capture in the same JVM as the expand pool"); + + CAPTURED_CRON_ID.set(null); + CAPTURED_CRON_SPEC.set(null); + + SeededTable seeded = + MaintenanceSchedulerITSupport.seedTaggedCompactionTable( + client.loadMetalake(metalake), metalake, TIME_TRIGGER, /*commitTrigger=*/ null); + + PolicyEntity policy = PolicyMetaService.getInstance().getPolicyById(seeded.policyId); + List<ExpandTableCandidate> candidates = + new NearestPolicyTableEnumerator() + .listTableIds(policy, 10, new OccupancyGate()::shouldSkip, 0L); + assertEquals( + List.of(seeded.tableId), + candidates.stream().map(ExpandTableCandidate::tableId).toList(), + "nearest enumerator should see the tagged Iceberg table"); + + DataSource ds = dataSource(); + assertTrue( + PolicyExpandRegistrar.forReads(ds).scheduledExecutionTime(seeded.policyId).isPresent(), + "createPolicy should register a policy-expand row via PolicyExpandLifecycleHook"); + + // Force the expand lease overdue so the live expand pool picks it (production next fire is + // tomorrow 02:00 for TIME_TRIGGER). + PolicyExpandRegistrar.forReads(ds).scheduleAt(seeded.policyId, Instant.now().minusSeconds(5)); + + awaitCronCapture(CronJobIds.forTable(seeded.tableId, seeded.policyId), 30L); + + MaintenanceJobSpec spec = CAPTURED_CRON_SPEC.get(); + assertNotNull(spec); + assertTrue(spec.isValid()); + assertEquals(List.of(seeded.tableId), spec.tableIds()); + assertEquals(seeded.policyId, spec.getOperations().get(0).getPolicyId()); + assertTrue(spec.getJobConf().containsKey("updater_options")); + assertEquals(seeded.catalogName, spec.getTables().get(0).catalogName()); + assertEquals( + seeded.schemaName + "." + seeded.tableName, spec.getTables().get(0).sparkTableIdentifier()); + } + + private static void awaitCronCapture(String expectedInstanceId, long timeoutSeconds) + throws InterruptedException { + long deadline = System.nanoTime() + TimeUnit.SECONDS.toNanos(timeoutSeconds); + while (System.nanoTime() < deadline) { + if (expectedInstanceId.equals(CAPTURED_CRON_ID.get()) && CAPTURED_CRON_SPEC.get() != null) { + return; + } + Thread.sleep(50L); + } + assertEquals( + expectedInstanceId, + CAPTURED_CRON_ID.get(), + "expand pool did not enqueue expected cron-job within " + timeoutSeconds + "s"); + } + + private static DataSource dataSource() { + return SqlSessionFactoryHelper.getInstance() + .getSqlSessionFactory() + .getConfiguration() + .getEnvironment() + .getDataSource(); + } +} diff --git a/plugins/maintenance/src/test/java/org/apache/gravitino/maintenance/policy/TestPolicyExpandRegistrar.java b/plugins/maintenance/src/test/java/org/apache/gravitino/maintenance/policy/TestPolicyExpandRegistrar.java index 04b93e7530..3bcc0e41b9 100644 --- a/plugins/maintenance/src/test/java/org/apache/gravitino/maintenance/policy/TestPolicyExpandRegistrar.java +++ b/plugins/maintenance/src/test/java/org/apache/gravitino/maintenance/policy/TestPolicyExpandRegistrar.java @@ -22,7 +22,9 @@ import com.github.kagkarlsson.scheduler.SchedulerClient; import com.github.kagkarlsson.scheduler.task.TaskInstanceId; import java.time.Instant; import java.time.temporal.ChronoUnit; +import java.util.Map; import javax.sql.DataSource; +import org.apache.gravitino.Config; import org.apache.gravitino.maintenance.TableMaintenanceConfig; import org.apache.gravitino.maintenance.scheduler.TableMaintenanceSchedulerPools; import org.apache.gravitino.storage.relational.TestJDBCBackend; @@ -92,7 +94,7 @@ class TestPolicyExpandRegistrar extends TestJDBCBackend { } @TestTemplate - void testRunningPoolsLeaveOverdueExpandRowsUnpicked() throws Exception { + void testRunningPoolsRemoveExpandRowForMissingPolicy() throws Exception { Instant overdue = Instant.now().minusSeconds(30).truncatedTo(ChronoUnit.MICROS); PolicyExpandRegistrar writer = PolicyExpandRegistrar.joiningCurrentTransaction(); DataSource dataSource = entityStoreDataSource(); @@ -105,15 +107,25 @@ class TestPolicyExpandRegistrar extends TestJDBCBackend { .build(); TaskInstanceId instanceId = TaskInstanceId.of(TableMaintenanceSchedulerPools.POLICY_EXPAND, "11"); - Instant stored = client.getScheduledExecution(instanceId).orElseThrow().getExecutionTime(); + Assertions.assertTrue(client.getScheduledExecution(instanceId).isPresent()); TableMaintenanceSchedulerPools pools = new TableMaintenanceSchedulerPools(); - pools.start(dataSource, TableMaintenanceConfig.forTest(1, 1, 1, 100L, 1_000L, 4, 10)); + Config gravitinoConfig = new Config(false) {}; + gravitinoConfig.loadFromMap( + Map.of( + MaintenanceAuthConfigs.GRAVITINO_MAINTENANCE_GRAVITINO_AUTH_URI.getKey(), + "http://localhost:8090";), + k -> true); + pools.start( + dataSource, TableMaintenanceConfig.forTest(1, 1, 1, 100L, 1_000L, 4, 10), gravitinoConfig); try { - Thread.sleep(400L); - Assertions.assertFalse(client.getScheduledExecution(instanceId).orElseThrow().isPicked()); - Assertions.assertEquals( - stored, client.getScheduledExecution(instanceId).orElseThrow().getExecutionTime()); + long deadline = System.nanoTime() + java.util.concurrent.TimeUnit.SECONDS.toNanos(5); + while (client.getScheduledExecution(instanceId).isPresent()) { + Assertions.assertTrue( + System.nanoTime() < deadline, + "expand row for missing policy should be removed after pick"); + Thread.sleep(100L); + } } finally { pools.stop(); } diff --git a/plugins/maintenance/src/test/java/org/apache/gravitino/maintenance/scheduler/SchedulerPoolTestHooks.java b/plugins/maintenance/src/test/java/org/apache/gravitino/maintenance/scheduler/SchedulerPoolTestHooks.java new file mode 100644 index 0000000000..eb55ac8eaf --- /dev/null +++ b/plugins/maintenance/src/test/java/org/apache/gravitino/maintenance/scheduler/SchedulerPoolTestHooks.java @@ -0,0 +1,51 @@ +/* + * 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; + +import com.google.common.base.Preconditions; +import java.util.function.BiConsumer; + +/** + * Same-package test access to {@link TableMaintenanceSchedulerPools} submit handlers. + * + * <p>Keeps production pools free of {@code *ForTest} APIs while still allowing BaseITs to record + * cron/commit enqueue without a real Spark {@code runJob}. + */ +public final class SchedulerPoolTestHooks { + + private SchedulerPoolTestHooks() {} + + /** + * Replaces the cron-job submit handler after pools have started. + * + * @param submitHandler handler used by the cron pool task + */ + public static void installCronSubmit(BiConsumer<String, MaintenanceJobSpec> submitHandler) { + TableMaintenanceSchedulerPools.CRON_SUBMIT.set(Preconditions.checkNotNull(submitHandler)); + } + + /** + * Replaces the commit-job submit handler after pools have started. + * + * @param submitHandler handler used by the commit pool task + */ + public static void installCommitSubmit(BiConsumer<String, MaintenanceJobSpec> submitHandler) { + TableMaintenanceSchedulerPools.COMMIT_SUBMIT.set(Preconditions.checkNotNull(submitHandler)); + } +} diff --git a/plugins/maintenance/src/test/java/org/apache/gravitino/maintenance/scheduler/policy/expand/TestPolicyExpandCrons.java b/plugins/maintenance/src/test/java/org/apache/gravitino/maintenance/scheduler/policy/expand/TestPolicyExpandCrons.java index f3f17b6348..f8663223e7 100644 --- a/plugins/maintenance/src/test/java/org/apache/gravitino/maintenance/scheduler/policy/expand/TestPolicyExpandCrons.java +++ b/plugins/maintenance/src/test/java/org/apache/gravitino/maintenance/scheduler/policy/expand/TestPolicyExpandCrons.java @@ -19,9 +19,6 @@ 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; @@ -29,25 +26,10 @@ import org.junit.jupiter.api.Test; 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); - } + void testNextExecutionReturnsFutureInstant() { + Instant before = Instant.now().minusSeconds(1); + Instant next = PolicyExpandCrons.nextExecution("0 0 2 * * ?"); + Assertions.assertTrue(next.isAfter(before)); } @Test
