lasdf1234 commented on code in PR #13774: URL: https://github.com/apache/gravitino/pull/13774#discussion_r4259672495
########## 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") Review Comment: Fixed: keys are now camelCase (`scheduler.expandThreads` / `cronThreads` / `commitThreads`), consistent with `pollingIntervalMs`. ########## 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: Agreed — bind-after-runJob fencing alone cannot undo a successful submit. This PR documents the submit contract on `OccupancyClaim` / `tryAcquireInFlight`: OccupancyGate follow-ups must make `runJob` idempotent on `claim.id()` (or outbox / reconcile-before-reclaim). There is no submit path in this PR yet; the gate will implement that protocol when wired. ########## 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: Fixed: design §5.2 (and related store / conf tables) now describe three physical `scheduled_*_tasks` tables and camelCase pool keys, matching this implementation. ########## 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: Fixed: all eight numeric `ConfigEntry`s now use `.checkValue(v -> v > 0, ConfigConstants.POSITIVE_NUMBER_ERROR_MSG)`. ########## LICENSE.bin: ########## @@ -411,6 +411,8 @@ Fabric8 Kubernetes Client Fabric8 Kubernetes Model Fabric8 zjsonpatch + db-scheduler Review Comment: Fixed: `LICENSE.bin` now lists `cron-utils (shaded into db-scheduler)`. ########## 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); Review Comment: Fixed: insert / reclaim / live-TTL checks use `DatabaseTimeSQL` (DB clock) instead of `System.currentTimeMillis`, so every node evaluates the lease against the same timeline. -- 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]
