[Fix][Engine] Continue task transition when state entry is missing

seatunnel-engine/seatunnel-engine-server/src/main/java/org/apache/seatunnel/engine/server/checkpoint/CheckpointCoordinator.java

@@ -551,7 +551,9 @@ protected void scheduleTriggerPendingCheckpoint(
      */
     protected void readyToClose(TaskLocation taskLocation) {
         readyToCloseStartingTask.add(taskLocation);
-        updateReadyToCloseStartingTask();
+        if (!updateReadyToCloseStartingTask()) {
+            return;
+        }
         if (readyToCloseStartingTask.size() == plan.getStartingSubtasks().size()) {
             tryTriggerPendingCheckpoint(CheckpointType.COMPLETED_POINT_TYPE);
         }
@@ -583,22 +585,10 @@ private Set<TaskLocation> loadReadyToCloseStartingTask() {
         }
     }
 
-    private void updateReadyToCloseStartingTask() {
+    private boolean updateReadyToCloseStartingTask() {
         try {
-            RetryUtils.retryWithException(
-                    () -> {
-                        runningJobStateIMap.compute(
-                                readyToCloseImapKey,
-                                (k, exist) -> {
-                                    Set<TaskLocation> merged =
-                                            exist instanceof Set
-                                                    ? new HashSet<>((Set<TaskLocation>) exist)
-                                                    : new HashSet<>();
-                                    merged.addAll(readyToCloseStartingTask);
-                                    return merged;
-                                });
-                        return null;
-                    },
+            return RetryUtils.retryWithException(
+                    this::persistReadyToCloseStartingTask,
                     new RetryUtils.RetryMaterial(
                             Constant.OPERATION_RETRY_TIME,
                             true,
@@ -617,6 +607,56 @@ private void updateReadyToCloseStartingTask() {
         }
     }
 
+    /**
+     * Persists ready-to-close state with a cleanup fence only when the key must be created.
+     *
+     * <p>The state lock is released before taking the job fence and then reacquired, preserving the
+     * cleanup-to-state lock order. Existing keys never take the job-wide fence.
+     */
+    private boolean persistReadyToCloseStartingTask() {
+        boolean stateLocked = false;
+        boolean missingStateFenceLocked = false;
+        try {
+            runningJobStateIMap.lock(readyToCloseImapKey);
+            stateLocked = true;
+            if (runningJobStateIMap.get(readyToCloseImapKey) == null) {
+                runningJobStateIMap.unlock(readyToCloseImapKey);
+                stateLocked = false;
+                checkpointManager.lockStatePersistenceFence();
+                missingStateFenceLocked = true;
+                runningJobStateIMap.lock(readyToCloseImapKey);
+                stateLocked = true;
+            }
+            if (!isStatePersistenceAllowed()) {
+                LOG.info(
+                        "Skip persisting {} because its job generation no longer owns distributed state",
+                        readyToCloseImapKey);
+                return false;
+            }
+            runningJobStateIMap.compute(
+                    readyToCloseImapKey,
+                    (k, exist) -> {
+                        Set<TaskLocation> merged =
+                                exist instanceof Set
+                                        ? new HashSet<>((Set<TaskLocation>) exist)
+                                        : new HashSet<>();
+                        merged.addAll(readyToCloseStartingTask);
+                        return merged;
+                    });
+            return true;
+        } finally {
+            try {
+                if (stateLocked) {
+                    runningJobStateIMap.unlock(readyToCloseImapKey);
+                }
+            } finally {
+                if (missingStateFenceLocked) {
+                    checkpointManager.unlockStatePersistenceFence();
+                }
+            }
+        }
+    }
+
     protected void readyToCloseIdleTask(TaskLocation taskLocation) {
         if (plan.getStartingSubtasks().contains(taskLocation)) {
             throw new UnsupportedOperationException("Unsupported close starting task");
@@ -1163,7 +1203,7 @@ protected void cleanPendingCheckpoint(CheckpointCloseReason closedReason) {
             // (completed/failed/cancelled). During a reset (master failover), the IMap entry
             // must be preserved so restoreCoordinator() can recover from it.
             if (closedReason != CheckpointCloseReason.CHECKPOINT_COORDINATOR_RESET) {
-                runningJobStateIMap.remove(readyToCloseImapKey);
+                removeReadyToCloseStateIfOwned();
             }
             scheduler.shutdownNow();
             scheduler =
@@ -1430,20 +1470,31 @@ private synchronized void updateStatus(@NonNull CheckpointCoordinatorStatus targ
         try {
             RetryUtils.retryWithException(
                     () -> {
-                        Object currentStatus = runningJobStateIMap.get(checkpointStateImapKey);
-                        if (currentStatus == null) {
-                            LOG.warn(
-                                    String.format(
-                                            "%s has already been cleaned, skip persisting transition to %s",
-                                            checkpointStateImapKey, targetStatus));
-                            return null;
+                        runningJobStateIMap.lock(checkpointStateImapKey);
+                        try {
+                            if (!isStatePersistenceAllowed()) {
+                                LOG.info(
+                                        "Skip persisting {} because its job generation no longer owns distributed state",
+                                        checkpointStateImapKey);
+                                return null;
+                            }
+                            Object currentStatus = runningJobStateIMap.get(checkpointStateImapKey);
+                            if (currentStatus == null) {
+                                LOG.warn(
+                                        String.format(
+                                                "%s has already been cleaned, skip persisting transition to %s",
+                                                checkpointStateImapKey, targetStatus));
+                                return null;
+                            }
+                            LOG.info(
+                                    "Turn {} state from {} to {}",
+                                    checkpointStateImapKey,
+                                    currentStatus,
+                                    targetStatus);
+                            runningJobStateIMap.set(checkpointStateImapKey, targetStatus);
+                        } finally {
+                            runningJobStateIMap.unlock(checkpointStateImapKey);
                         }
-                        LOG.info(
-                                "Turn {} state from {} to {}",
-                                checkpointStateImapKey,
-                                currentStatus,
-                                targetStatus);
-                        runningJobStateIMap.set(checkpointStateImapKey, targetStatus);
                         return null;
                     },
                     new RetryUtils.RetryMaterial(
@@ -1459,6 +1510,33 @@ private synchronized void updateStatus(@NonNull CheckpointCoordinatorStatus targ
         }
     }
 
+    /**
+     * Removes ready-to-close recovery state only while this job generation still owns the key.
+     *
+     * <p>The state-key lock serializes this delete with ready-to-close compute and terminal job
+     * cleanup; the generation check prevents an old coordinator from deleting a newer value.
+     */
+    private void removeReadyToCloseStateIfOwned() {
+        runningJobStateIMap.lock(readyToCloseImapKey);
+        try {
+            if (isStatePersistenceAllowed()) {
+                runningJobStateIMap.remove(readyToCloseImapKey);
+            }
+        } finally {
+            runningJobStateIMap.unlock(readyToCloseImapKey);
+        }
+    }
+
+    /**
+     * Applies generation fencing when a production JobMaster is present.
+     *
+     * <p>Standalone coordinator tests and tooling intentionally construct a manager without a job
+     * master; those instances retain their historical in-memory behavior.
+     */
+    private boolean isStatePersistenceAllowed() {
+        return !checkpointManager.hasJobMaster() || checkpointManager.isStatePersistenceAllowed();
+    }
+
     /**
      * Schedules a schema-change-before checkpoint if no schema change is currently in progress.
      *

seatunnel-engine/seatunnel-engine-server/src/main/java/org/apache/seatunnel/engine/server/checkpoint/CheckpointManager.java

@@ -199,6 +199,47 @@ public CheckpointCoordinator getCheckpointCoordinator(int pipelineId) {
         return coordinator;
     }
 
+    /**
+     * Returns whether this checkpoint manager's exact job generation may mutate distributed state.
+     *
+     * <p>The JobMaster checks both the durable owner token and pending cleanup fence.
+     */
+    boolean isStatePersistenceAllowed() {
+        return jobMaster.isStatePersistenceAllowed();
+    }
+
+    /**
+     * Returns whether this manager is attached to a production job generation.
+     *
+     * <p>Standalone coordinator tests intentionally omit a JobMaster and retain legacy in-memory
+     * behavior.
+     */
+    boolean hasJobMaster() {
+        return jobMaster != null;
+    }
+
+    /**
+     * Locks the owning job's cleanup fence before checkpoint state creates a missing key.
+     *
+     * <p>Existing checkpoint keys remain independently synchronized by their state-key locks.
+     */
+    void lockStatePersistenceFence() {
+        if (jobMaster != null) {
+            jobMaster.lockStatePersistenceFence();
+        }
+    }
+
+    /**
+     * Releases the missing-key checkpoint persistence fence.
+     *
+     * <p>The caller releases its checkpoint state-key lock first to preserve lock ordering.
+     */
+    void unlockStatePersistenceFence() {
+        if (jobMaster != null) {
+            jobMaster.unlockStatePersistenceFence();
+        }
+    }
+
     /**
      * Called by the {@link Task}. <br>
      * used by Task to report the {@link SeaTunnelTaskState} of the state machine.