lasdf1234 commented on code in PR #14054:
URL: https://github.com/apache/gravitino/pull/14054#discussion_r4335109536
##########
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:
Fixed in 91361e192.
Failsafe reclaim now keeps only a real `afterSnapshotId` (or null). It no
longer copies `beforeSnapshotId` into `after_snapshot_id`, so reconciliation
can still tell “S1 never sampled” from a true S1. Covered by
`testReclaimDoesNotCopyBeforeSnapshotIdIntoAfter`.
##########
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);
Review Comment:
Fixed in 91361e192.
Job lookup now goes through `TableMetaService.getMetalakeNameByTableId`
(table → metalake), not the current policy. Reclaim uses `BoundJobLookup`: only
`TERMINAL` / confirmed `MISSING` close the action; `ACTIVE` and `UNRESOLVED`
leave the soft slot open. Covered by
`testActiveJobNotReclaimedWhenPolicyIdIsUnknown` and
`testUnresolvedTableMetalakeDoesNotReclaimBoundRow`.
--
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]
