roryqi commented on code in PR #13774:
URL: https://github.com/apache/gravitino/pull/13774#discussion_r4259303877
##########
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:
[P2] Do not base a multi-node lease on each JVM's System.currentTimeMillis.
If node A is behind by more than claimTimeoutMs, the created_at it inserts is
already expired according to node B, so B can immediately reclaim while A is
still submitting; a clock-ahead node can conversely block reclaim much longer
than configured. Use database time in the insert/reclaim predicate, or a
database-generated lease timestamp, so every contender evaluates the TTL
against the same clock.
--
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]
