roryqi commented on code in PR #13774:
URL: https://github.com/apache/gravitino/pull/13774#discussion_r4259277857
##########
plugins/tms/src/main/java/org/apache/gravitino/maintenance/storage/service/TableMaintenanceActionMetaService.java:
##########
@@ -43,42 +49,125 @@ public static TableMaintenanceActionMetaService
getInstance() {
}
/**
- * Inserts an in-flight occupancy row before {@code runJob}. {@code job_id}
is bound later via
- * {@link #bindJobId(long, long)}.
+ * Tries to acquire an unbound occupancy claim before {@code runJob}.
+ *
+ * <ol>
+ * <li>CAS-{@code UPDATE created_at} on an expired unbound row ({@code
finished_at = 0})
+ * <li>Otherwise {@code INSERT} with {@code finished_at = 0}; unique
conflict means lost race
+ * </ol>
+ *
+ * <p>{@code finished_at = 0} is the soft exclusive slot (like {@code
deleted_at}); it stays until
+ * {@link #markFinished}. {@code created_at} is the unbound claim lease
start.
*
* @param tableId table entity id
* @param policyId policy entity id
- * @return generated action id
+ * @param nowMs current epoch millis
+ * @param claimTimeoutMs unbound claim TTL
+ * @return claim ({@code id} + acquire-time {@code created_at}) when
acquired, empty when another
+ * live claim or open run holds the slot
*/
@Monitored(
metricsSource = GRAVITINO_RELATIONAL_STORE_METRIC_NAME,
- baseMetricName = "insertTableMaintenanceAction")
- public long insertInFlight(long tableId, long policyId) {
+ baseMetricName = "tryAcquireTableMaintenanceAction")
+ public Optional<OccupancyClaim> tryAcquireInFlight(
+ long tableId, long policyId, long nowMs, long claimTimeoutMs) {
+ Preconditions.checkArgument(claimTimeoutMs > 0, "claimTimeoutMs must be
positive");
+ long expireBefore = nowMs - claimTimeoutMs;
+ int reclaimed =
+ SessionUtils.doWithCommitAndFetchResult(
+ TableMaintenanceActionMapper.class,
+ mapper -> mapper.reclaimExpiredUnbound(tableId, policyId, nowMs,
expireBefore));
+ if (reclaimed == 1) {
+ TableMaintenanceActionPO unbound =
+ SessionUtils.getWithoutCommit(
+ TableMaintenanceActionMapper.class,
+ mapper -> mapper.selectUnboundByTableAndPolicy(tableId,
policyId));
+ Preconditions.checkState(unbound != null && unbound.id() != null,
"reclaim lost unbound row");
+ // Fence with the created_at this acquire wrote (nowMs), never re-read
created_at.
+ return Optional.of(new OccupancyClaim(unbound.id(), nowMs));
+ }
+
+ TableMaintenanceActionPO existing =
+ SessionUtils.getWithoutCommit(
+ TableMaintenanceActionMapper.class,
+ mapper -> mapper.selectUnboundByTableAndPolicy(tableId, policyId));
+ if (existing != null
+ && existing.createdAt() != null
+ && nowMs - existing.createdAt() < claimTimeoutMs) {
+ return Optional.empty();
+ }
+
TableMaintenanceActionPO actionMeta =
TableMaintenanceActionPO.builder()
.withTableId(tableId)
.withPolicyId(policyId)
- .withCreatedAt(System.currentTimeMillis())
+ .withCreatedAt(nowMs)
+ .withFinishedAt(0L)
.build();
- SessionUtils.doWithCommit(
- TableMaintenanceActionMapper.class, mapper ->
mapper.insertAction(actionMeta));
+ try {
+ SessionUtils.doWithCommit(
+ TableMaintenanceActionMapper.class, mapper ->
mapper.insertAction(actionMeta));
+ } catch (RuntimeException e) {
+ if (isUniqueViolation(e)) {
+ return Optional.empty();
+ }
+ throw e;
+ }
Preconditions.checkState(actionMeta.id() != null, "insert did not return
generated id");
- return actionMeta.id();
+ return Optional.of(new OccupancyClaim(actionMeta.id(), nowMs));
}
/**
- * Binds {@code job_run_meta.job_run_id} after a successful {@code runJob}.
+ * Acquires an unbound claim using {@link
TableMaintenanceConfig#claimTimeoutMs()} or fails.
*
- * @param id action row id
+ * @param tableId table entity id
+ * @param policyId policy entity id
+ * @param config maintenance config supplying the unbound claim TTL
+ * @return acquired claim
+ */
+ public OccupancyClaim insertInFlight(long tableId, long policyId,
TableMaintenanceConfig config) {
+ Preconditions.checkNotNull(config, "config must not be null");
+ return insertInFlight(tableId, policyId, config.claimTimeoutMs());
+ }
+
+ /**
+ * Acquires an unbound claim with the given claim timeout or fails.
+ *
+ * @param tableId table entity id
+ * @param policyId policy entity id
+ * @param claimTimeoutMs unbound claim TTL
+ * @return acquired claim
+ */
+ @Monitored(
+ metricsSource = GRAVITINO_RELATIONAL_STORE_METRIC_NAME,
+ baseMetricName = "insertTableMaintenanceAction")
+ public OccupancyClaim insertInFlight(long tableId, long policyId, long
claimTimeoutMs) {
+ Optional<OccupancyClaim> acquired =
+ tryAcquireInFlight(tableId, policyId, System.currentTimeMillis(),
claimTimeoutMs);
+ Preconditions.checkState(acquired.isPresent(), "failed to acquire unbound
occupancy claim");
+ return acquired.get();
+ }
+
+ /**
+ * Binds {@code job_run_meta.job_run_id} after a successful {@code runJob}.
Does not release the
+ * soft slot ({@code finished_at} stays {@code 0} until {@link
#markFinished}).
+ *
+ * <p>Pass the {@link OccupancyClaim} from acquire / reclaim as-is. Do not
re-read {@code
+ * created_at} from the row: after another node reclaims, that would observe
the new fence and
+ * defeat reclaim fencing. The CAS matches zero rows for a stale claim.
+ *
+ * @param claim claim returned by {@link #tryAcquireInFlight} / {@link
#insertInFlight}
* @param jobId Gravitino job run id
*/
@Monitored(
metricsSource = GRAVITINO_RELATIONAL_STORE_METRIC_NAME,
baseMetricName = "bindTableMaintenanceActionJobId")
- public void bindJobId(long id, long jobId) {
+ public void bindJobId(OccupancyClaim claim, long jobId) {
Review Comment:
[P1] The fence is checked only after the non-idempotent runJob side effect,
so it still cannot prevent duplicate Spark submissions. For example, A acquires
the claim and runJob succeeds, but A pauses or dies before this bind. After
claimTimeoutMs, B reclaims the still-unbound row and calls runJob again. A
failing this CAS, or never reaching it, does not undo the first submitted job.
Please make submission idempotent using the claim id/token, persist submission
through an outbox/atomic protocol, or reconcile an unbound row with the
already-created job before allowing reclaim; fencing only the later database
bind is too late.
##########
plugins/tms/src/main/java/org/apache/gravitino/maintenance/TableMaintenanceConfig.java:
##########
@@ -0,0 +1,255 @@
+/*
+ * 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;
+
+import org.apache.gravitino.Config;
+import org.apache.gravitino.config.ConfigBuilder;
+import org.apache.gravitino.config.ConfigConstants;
+import org.apache.gravitino.config.ConfigEntry;
+
+/** Configuration for table maintenance scheduler pools (§7.2). */
+public final class TableMaintenanceConfig {
+
+ public static final ConfigEntry<Integer> SCHEDULER_EXPAND_THREADS =
+ new ConfigBuilder("gravitino.maintenance.scheduler.expand.threads")
+ .doc("Thread count for the policy-expand db-scheduler pool")
+ .version(ConfigConstants.VERSION_2_0_0)
+ .intConf()
+ .createWithDefault(4);
Review Comment:
[P2] Please validate these scheduler settings when the ConfigEntry is
parsed. None of the thread counts, intervals, missed-heartbeat limit, batch
size, or claim timeout has a positive-value check. A zero/negative thread or
heartbeat value can make Scheduler.start fail after executors have already
started, while an invalid batch size fails later in an unrelated pipeline. The
existing Gravitino config pattern is checkValue(value -> value > 0,
ConfigConstants.POSITIVE_NUMBER_ERROR_MSG); it should be applied to all eight
numeric entries here.
##########
plugins/tms/src/main/java/org/apache/gravitino/maintenance/scheduler/TableMaintenanceSchedulerPools.java:
##########
@@ -0,0 +1,151 @@
+/*
+ * 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.github.kagkarlsson.scheduler.Scheduler;
+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.util.ArrayList;
+import java.util.List;
+import javax.sql.DataSource;
+import org.apache.gravitino.maintenance.TableMaintenanceConfig;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+/**
+ * Starts and stops three isolated db-scheduler pools (§5.2, §7.2), each with
its own task table.
Review Comment:
[P2] Please update the referenced design spec to match this storage layout.
Section 5.2 currently specifies three task kinds in one scheduled_tasks table
and explicitly says that table uses upstream db-scheduler DDL, whereas this
implementation creates three physical tables and states that sharing is
unsupported. Since this Javadoc claims conformance with section 5.2, leaving
the merged spec unchanged gives follow-up implementations two conflicting
sources of truth.
##########
scripts/mysql/schema-2.0.0-mysql.sql:
##########
@@ -668,23 +668,73 @@ CREATE TABLE IF NOT EXISTS `iceberg_cleanup_job` (
KEY `idx_object` (`catalog_id`, `namespace`(255), `table_name`(128), `state`)
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_bin COMMENT 'async
Iceberg table cleanup jobs';
+CREATE TABLE IF NOT EXISTS `scheduled_expand_tasks` (
+ `task_name` VARCHAR(100) NOT NULL COMMENT 'task name',
+ `task_instance` VARCHAR(100) NOT NULL COMMENT 'task instance',
+ `task_data` BLOB COMMENT 'serialized task data',
+ `execution_time` TIMESTAMP(6) NOT NULL COMMENT 'next execution time',
+ `picked` BOOLEAN NOT NULL COMMENT 'whether the task is currently picked',
+ `picked_by` VARCHAR(50) COMMENT 'scheduler instance that picked the task',
+ `last_heartbeat` TIMESTAMP(6) NULL COMMENT 'last heartbeat from the
picker',
+ `version` BIGINT NOT NULL COMMENT 'optimistic lock version',
+ `consecutive_failures` INT COMMENT 'consecutive failure count',
+ `last_success` TIMESTAMP(6) NULL COMMENT 'last successful execution time',
+ `last_failure` TIMESTAMP(6) NULL COMMENT 'last failed execution time',
+ PRIMARY KEY (`task_name`, `task_instance`),
+ KEY `execution_time_idx` (`execution_time`),
+ KEY `last_heartbeat_idx` (`last_heartbeat`)
+) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_bin COMMENT 'table
maintenance expand schedule metadata';
+
+CREATE TABLE IF NOT EXISTS `scheduled_cron_tasks` (
+ `task_name` VARCHAR(100) NOT NULL COMMENT 'task name',
+ `task_instance` VARCHAR(100) NOT NULL COMMENT 'task instance',
+ `task_data` BLOB COMMENT 'serialized task data',
+ `execution_time` TIMESTAMP(6) NOT NULL COMMENT 'next execution time',
+ `picked` BOOLEAN NOT NULL COMMENT 'whether the task is currently picked',
+ `picked_by` VARCHAR(50) COMMENT 'scheduler instance that picked the task',
+ `last_heartbeat` TIMESTAMP(6) NULL COMMENT 'last heartbeat from the
picker',
+ `version` BIGINT NOT NULL COMMENT 'optimistic lock version',
+ `consecutive_failures` INT COMMENT 'consecutive failure count',
+ `last_success` TIMESTAMP(6) NULL COMMENT 'last successful execution time',
+ `last_failure` TIMESTAMP(6) NULL COMMENT 'last failed execution time',
+ PRIMARY KEY (`task_name`, `task_instance`),
+ KEY `execution_time_idx` (`execution_time`),
+ KEY `last_heartbeat_idx` (`last_heartbeat`)
+) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_bin COMMENT 'table
maintenance cron schedule metadata';
+
+CREATE TABLE IF NOT EXISTS `scheduled_commit_tasks` (
+ `task_name` VARCHAR(100) NOT NULL COMMENT 'task name',
+ `task_instance` VARCHAR(100) NOT NULL COMMENT 'task instance',
+ `task_data` BLOB COMMENT 'serialized task data',
+ `execution_time` TIMESTAMP(6) NOT NULL COMMENT 'next execution time',
+ `picked` BOOLEAN NOT NULL COMMENT 'whether the task is currently picked',
+ `picked_by` VARCHAR(50) COMMENT 'scheduler instance that picked the task',
+ `last_heartbeat` TIMESTAMP(6) NULL COMMENT 'last heartbeat from the
picker',
+ `version` BIGINT NOT NULL COMMENT 'optimistic lock version',
+ `consecutive_failures` INT COMMENT 'consecutive failure count',
+ `last_success` TIMESTAMP(6) NULL COMMENT 'last successful execution time',
+ `last_failure` TIMESTAMP(6) NULL COMMENT 'last failed execution time',
+ PRIMARY KEY (`task_name`, `task_instance`),
+ KEY `execution_time_idx` (`execution_time`),
+ KEY `last_heartbeat_idx` (`last_heartbeat`)
+) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_bin COMMENT 'table
maintenance commit schedule metadata';
+
CREATE TABLE IF NOT EXISTS `table_maintenance_action` (
`id` BIGINT(20) UNSIGNED NOT NULL AUTO_INCREMENT COMMENT 'auto increment
id',
- `job_id` BIGINT(20) UNSIGNED NULL COMMENT 'optional job run id mapped to
job_run_meta.job_run_id, null until bound after runJob',
+ `job_id` BIGINT(20) UNSIGNED NULL COMMENT 'job run id, maps to
job_run_meta.job_run_id',
`table_id` BIGINT(20) UNSIGNED NOT NULL COMMENT 'table id from table_meta',
`policy_id` BIGINT(20) UNSIGNED NOT NULL COMMENT 'policy id from
policy_meta',
`before_snapshot_id` BIGINT(20) NULL COMMENT 'Iceberg snapshot id before
maintenance, null until sampled',
`after_snapshot_id` BIGINT(20) NULL COMMENT 'Iceberg snapshot id after
maintenance, null while pending',
- `finished_at` BIGINT(20) UNSIGNED NULL COMMENT 'finish timestamp in
milliseconds, null while in flight',
+ `finished_at` BIGINT(20) UNSIGNED NOT NULL DEFAULT 0 COMMENT 'finish
timestamp in milliseconds, 0 while in flight',
`created_at` BIGINT(20) UNSIGNED NOT NULL COMMENT 'creation timestamp in
milliseconds',
PRIMARY KEY (`id`),
UNIQUE KEY `uk_tma_job_id` (`job_id`),
- KEY `idx_tma_table_policy_finished` (`table_id`, `policy_id`,
`finished_at`),
+ UNIQUE KEY `uk_tma_table_policy_finished` (`table_id`, `policy_id`,
`finished_at`),
Review Comment:
[P2] This collision does not require concurrent finishers. The API accepts
any positive wall-clock millisecond, so a sequential execution is enough:
finish job 1 at T, acquire and bind job 2, then finish job 2 at the same T
because both early-exit jobs complete within one millisecond or because the
clock moved backward. The second UPDATE violates this unique key and leaves job
2 at finished_at = 0. Serializing the calls therefore does not make timestamp
values unique; please keep the history index non-unique and enforce only the
active slot separately.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
