roryqi commented on code in PR #14054:
URL: https://github.com/apache/gravitino/pull/14054#discussion_r4334867651
##########
plugins/tms/src/main/java/org/apache/gravitino/maintenance/occupancy/OccupancyGate.java:
##########
@@ -78,22 +82,62 @@ public OccupancyGate() {
/**
* @param tableId table entity id
* @param policyId policy entity id
- * @return true when any open row is a live unbound claim or an active
Gravitino job
+ * @return true when any open row is a live unbound claim or a still-bound
soft-slot row
*/
public boolean isInFlight(long tableId, long policyId) {
+ reclaimTerminalBound(tableId, policyId);
if (actionMetaService.hasLiveUnboundClaim(tableId, policyId,
claimTimeoutMs)) {
return true;
}
+ // After reclaim, any remaining bound open row occupies the soft slot
(ACTIVE job, or a
+ // terminal/missing job whose markFinished raced/failed). Fail closed so
UNIQUE(table, policy,
+ // finished_at=0) cannot be ignored while still blocking
tryAcquireInFlight.
+ for (TableMaintenanceActionPO open :
actionMetaService.listInFlight(tableId, policyId)) {
+ if (open.jobId() != null) {
+ return true;
+ }
+ }
+ return false;
+ }
+
+ /**
+ * Job-terminal reconciler failsafe (§5.2.2): when Spark finish callback was
missed, mark the
+ * action finished using the best-known snapshot id. Does not invent stub
metrics rows (those
+ * would pollute {@code table_snapshot_metrics}).
+ *
+ * @param tableId table entity id
+ * @param policyId policy entity id
+ */
+ public void reclaimTerminalBound(long tableId, long policyId) {
for (TableMaintenanceActionPO open :
actionMetaService.listInFlight(tableId, policyId)) {
if (open.jobId() == null) {
continue;
}
- JobEntity job = loadJob(open.jobId(), tableId);
+ JobEntity job = loadJob(open.jobId(), policyId);
+ // Missing job meta (purged / unresolvable) must free the soft slot;
otherwise UNIQUE(table,
+ // policy, finished_at=0) blocks every future acquire.
if (job != null && ACTIVE.contains(job.status())) {
- return true;
+ continue;
+ }
+ long finishedAt =
+ job != null && job.finishedAt() != null && job.finishedAt() > 0
+ ? job.finishedAt()
+ : System.currentTimeMillis();
+ Long afterSnapshotId =
Review Comment:
Could we avoid using `beforeSnapshotId` as `afterSnapshotId` here? If the
Job becomes terminal before the finish listener samples S1, this closes the
action with S0. The listener's later `markFinished` cannot correct it because
its update requires `finished_at = 0`. The TMS spec (§5.2.3) calls for sampling
S1 before setting `after_snapshot_id`; reconciliation needs to preserve that
distinction.
--
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]
