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

[davidzollo]

I met a problem during node scaling down.

What this PR fixes

During node scale-down, member removal, or master recovery windows, the distributed state entry for a job, pipeline, or task can disappear from runningJobStateIMap while the in-memory runtime still needs to finish a terminal or recoverable transition. The old PhysicalVertex.updateTaskState path could return early in that situation, so taskFuture might never complete and SubPlan pipeline completion counting could hang indefinitely.

Changes

• PhysicalVertex now routes task transitions through the shared distributed-state transition path and keeps terminal progress when the state entry is temporarily missing.
• PhysicalPlan and SubPlan use the same timestamp-first distributed-state transition helper, so job, pipeline, and task state recovery follow one contract.
• CoordinatorService now fences pending cleanup with pendingJobCleanupIMap.lock(jobId), rechecks owner generation under the fence, and removes the exact state/timestamp key set under per-key locks.
• Master-switch restore now revalidates job ownership and repairs a missing job-state entry through repairMissingJobStateForRestore / initializeMissingJobState instead of blindly deleting the running job info.
• CheckpointCoordinator persists ready-to-close source-task state through the same missing-key/fence pattern, which avoids losing close-readiness during recovery.
• Added focused tests for missing task state completion, distributed transition ordering, cleanup fencing, master restore, job history mapping, and checkpoint ready-to-close recovery.

Compatibility

This PR does not add or change any user-facing option, default value, serialized split/checkpoint format, REST/API contract, or connector SPI.

The flagged PhysicalPlan and PhysicalVertex constructor changes only add Java generic type parameters to existing Hazelcast IMap parameters. Java erasure keeps the runtime constructor descriptor as IMap, IMap, so callers compiled against the raw signature are not broken by a different JVM method descriptor. Removed helper methods in those classes were private implementation details.

Local verification

For this Apache SeaTunnel PR, local execution is intentionally limited to formatting / format checks. Compile, tests, packages, E2E, Docker, services, and real jobs are verified by GitHub CI on the PR head.

[dybyte]

It seems the same issue may also exist in PhysicalPlan.updateJobState() and SubPlan.updatePipelineState().

Could you please share how you discovered this issue?

[DanielCarter-stack]

Issue 1: NPE risk when currExecutionState may be null

Location: PhysicalVertex.java:363

if (stateEntryMissing) {
    log.warn(...);
    current = currExecutionState;  // currExecutionState may be null
}

Related Context:

• Constructor: PhysicalVertex.java:162 - this.currExecutionState = (ExecutionState) runningJobStateIMap.get(taskGroupLocation);
• Caller 1: PhysicalVertex.java:576 - stateProcess() → updateTaskState(ExecutionState.RUNNING)
• Caller 2: PhysicalVertex.java:580 - stateProcess() → updateTaskState(ExecutionState.FAILED)
• Caller 3: PhysicalVertex.java:623 - makeTaskGroupFailing() → updateTaskState(ExecutionState.FAILING)

Issue Description:
Although the PR handles the case where runningJobStateIMap.get() returns null, the fallback currExecutionState itself may also be null. Looking at line 162 of the constructor, currExecutionState initialization depends on runningJobStateIMap. In the following scenarios:

1. When constructing PhysicalVertex, there is no corresponding entry in runningJobStateIMap
2. Although lines 153-160 of the constructor will try to create an entry, under race conditions (such as parallel initialization of multiple PhysicalVertex), currExecutionState may still be null

When currExecutionState is null and stateEntryMissing=true, although the subsequent current.equals(targetState) (line 370) will not execute (due to the current != null check), if the target state transition flow depends on the value of current, it may lead to unexpected behavior.

Potential Risks:

• Risk 1: If currExecutionState is null, current will also be null, causing subsequent state transition logic to be based on incorrect assumptions
• Risk 2: The log will record the null state value, which may cause log parsing errors
• Risk 3: The switch statement in the stateProcess() method depends on the return value of getExecutionState(). If it returns null, a NullPointerException will be thrown

Impact Scope:

• Direct Impact: PhysicalVertex.updateTaskState() method
• Indirect Impact: All paths that call updateTaskState, including stateProcess(), makeTaskGroupFailing(), updateStateByExecutionService()
• Impact Area: Core framework (Engine Server)

Severity: MAJOR

Improvement Suggestions:

// Add additional null check in updateTaskState method
if (stateEntryMissing) {
    log.warn(
            "{} current state is null (possibly due to node removal during scaling down), "
                    + "continue local transition to {}. Task execution location: {}",
            taskFullName,
            targetState,
            taskGroupLocation);
    current = currExecutionState;
    
    // Added: If local state is also null, use default initial state
    if (current == null) {
        log.error(
                "{} Both distributed state and local state are null, assuming CREATED as fallback. "
                        + "Task execution location: {}",
                taskFullName,
                taskGroupLocation);
        current = ExecutionState.CREATED;
    }
}

Rationale:

1. Defensive programming: can continue processing even if currExecutionState is null
2. Use CREATED as a reasonable default value, consistent with the initial state of the state machine
3. Add error logging to facilitate detection of such exceptional situations
4. No need to modify the PhysicalVertex constructor, maintaining backward compatibility

---

Issue 2: Not updating distributed Map when state entry is missing may cause state inconsistency

Location: PhysicalVertex.java:386-398

// now do the actual state transition
if (!stateEntryMissing) {
    RetryUtils.retryWithException(
            () -> {
                updateStateTimestamps(targetState);
                runningJobStateIMap.set(taskGroupLocation, targetState);
                return null;
            },
            new RetryUtils.RetryMaterial(
                    Constant.OPERATION_RETRY_TIME,
                    true,
                    ExceptionUtil::isOperationNeedRetryException,
                    Constant.OPERATION_RETRY_SLEEP));
}
this.currExecutionState = targetState;

Related Context:

• initStateFuture(): PhysicalVertex.java:182-204 - restore state from runningJobStateIMap
• JobMaster failover: JobMaster.java - new Master restores job status from IMap
• Monitoring system: may read task status from runningJobStateIMap for display

Issue Description:
When stateEntryMissing=true, the PR chooses to only update the local state currExecutionState, not the distributed Map runningJobStateIMap. This leads to the following problems:

1. State inconsistency: Local state has been updated (e.g., FAILED), but the distributed Map still has no entry for this task
2. Failover failure: If JobMaster fails over, the new Master cannot restore this task's state from runningJobStateIMap
3. Monitoring blind spot: Monitoring systems that depend on runningJobStateIMap cannot see the latest state of this task

Potential Risks:

• Risk 1: After JobMaster failover, the new Master may think the task does not exist, leading to duplicate execution or state confusion
• Risk 2: When operations personnel view task status from Hazelcast IMap, they will get incomplete information
• Risk 3: If currExecutionState has been updated but there is no record in IMap, it may cause errors in certain logic that depends on IMap (such as counting completed tasks)

Impact Scope:

• Direct Impact: State persistence of PhysicalVertex
• Indirect Impact: JobMaster failover recovery, monitoring system, task statistics logic
• Impact Area: Core framework (state management and failover recovery)

Severity: MAJOR

Improvement Suggestions:

There are two possible improvement approaches:

Approach 1: Still try to write to distributed Map (recommended)

// now do the actual state transition
if (!stateEntryMissing) {
    RetryUtils.retryWithException(
            () -> {
                updateStateTimestamps(targetState);
                runningJobStateIMap.set(taskGroupLocation, targetState);
                return null;
            },
            new RetryUtils.RetryMaterial(
                    Constant.OPERATION_RETRY_TIME,
                    true,
                    ExceptionUtil::isOperationNeedRetryException,
                    Constant.OPERATION_RETRY_SLEEP));
} else {
    // Added: Try to recreate even if the entry was originally missing
    try {
        updateStateTimestamps(targetState);
        runningJobStateIMap.put(taskGroupLocation, targetState);
        log.info(
                "{} Recreated state entry in distributed map for state transition to {}",
                taskFullName,
                targetState);
    } catch (Exception e) {
        // If write fails, log error but don't block local state transition
        log.warn(
                "{} Failed to recreate state entry in distributed map: {}",
                taskFullName,
                ExceptionUtils.getMessage(e));
    }
}
this.currExecutionState = targetState;

Approach 2: Accept state inconsistency, but add documentation

/**
 * Update task state. When the state entry is missing from the distributed map
 * (e.g., due to node removal during scaling down), only the local state is updated.
 * This is a known trade-off to allow task state progression even when the
 * distributed state is unavailable.
 * 
 * Note: This may cause temporary inconsistency between local and distributed state.
 * The local state is the source of truth for task lifecycle management.
 */
public synchronized void updateTaskState(@NonNull ExecutionState targetState) {
    // ... existing implementation ...
}

Rationale:

Advantages of Approach 1:

1. Maintain state consistency: even if the entry was originally missing, try to recreate it
2. Support failover recovery: new Master can restore complete state from distributed Map
3. Monitoring visibility: monitoring systems can see the latest state

Advantages of Approach 2:

1. If recreating the entry may cause other problems (such as entry ID conflicts), then approach 2 is safer
2. Clearly inform users that this is a known trade-off
3. Leave room for future optimization

Recommend Approach 1, because:

• runningJobStateIMap.put() should be an idempotent operation
• If the entry truly cannot be created (e.g., Hazelcast cluster failure), the catch block will catch and log the exception
• Local state transition will still proceed, maintaining fault tolerance

---

Issue 3: Test cases do not cover the scenario where currExecutionState is null

Location: TaskTest.java:312-397

@Test
@SetEnvironmentVariable(key = SKIP_CHECK_JAR, value = "true")
public void testUpdateTaskStateWhenStateEntryMissing() throws MalformedURLException {
    // ... test code ...
    
    runningJobState.remove(physicalVertex.getTaskGroupLocation());
    physicalVertex.makeTaskGroupFailing(new RuntimeException("test missing state entry"));
    
    Assertions.assertTrue(stateFuture.isDone());
    Assertions.assertEquals(ExecutionState.FAILED, physicalVertex.getExecutionState());
    Assertions.assertEquals(ExecutionState.FAILED, stateFuture.join().getExecutionState());
}

Related Context:

• PhysicalVertex constructor: PhysicalVertex.java:127-180 - currExecutionState initialization logic
• Other test cases: other test methods in TaskTest.java

Issue Description:
Current test cases only verify the following scenarios:

1. Create PhysicalVertex and initialize (at this time currExecutionState will be correctly set)
2. Delete entry in runningJobState
3. Call makeTaskGroupFailing to trigger state transition

But this test does not cover the following scenarios:

1. Case where currExecutionState itself is null: if before runningJobState.remove(), currExecutionState is already null
2. Concurrent scenarios: multiple threads call updateTaskState simultaneously and the state entry is missing
3. Continuous state transitions: whether state can correctly advance when the state entry is missing multiple times

Potential Risks:

• Risk 1: The scenario where currExecutionState is null mentioned in Issue 1 is not covered by tests
• Risk 2: Boundary conditions may lead to unexpected behavior in production environment
• Risk 3: Insufficient test coverage may hide potential state machine bugs

Impact Scope:

• Direct Impact: Test coverage and code quality
• Indirect Impact: Production environment stability
• Impact Area: Single test case

Severity: MINOR

Improvement Suggestions:

@Test
@SetEnvironmentVariable(key = SKIP_CHECK_JAR, value = "true")
public void testUpdateTaskStateWhenStateEntryMissingAndLocalStateNull() throws MalformedURLException {
    IdGenerator idGenerator = new IdGenerator();
    
    // ... setup LogicalDag and PhysicalPlan (same as existing tests) ...
    
    PhysicalVertex physicalVertex =
            physicalPlan.getPipelineList().get(0).getPhysicalVertexList().get(0);
    PassiveCompletableFuture<TaskExecutionState> stateFuture = physicalVertex.initStateFuture();
    physicalVertex.startPhysicalVertex();
    
    // Delete state entry
    runningJobState.remove(physicalVertex.getTaskGroupLocation());
    
    // Use reflection to set currExecutionState to null, simulating extreme scenario
    java.lang.reflect.Field field = PhysicalVertex.class.getDeclaredField("currExecutionState");
    field.setAccessible(true);
    field.set(physicalVertex, null);
    
    // Trigger state transition
    physicalVertex.makeTaskGroupFailing(new RuntimeException("test with null local state"));
    
    // Verify: Even if local state is null, task should still reach terminal state
    Assertions.assertTrue(stateFuture.isDone());
    Assertions.assertEquals(ExecutionState.FAILED, physicalVertex.getExecutionState());
    Assertions.assertEquals(ExecutionState.FAILED, stateFuture.join().getExecutionState());
}

@Test
@SetEnvironmentVariable(key = SKIP_CHECK_JAR, value = "true")
public void testConcurrentStateUpdateWhenEntryMissing() throws MalformedURLException, Exception {
    IdGenerator idGenerator = new IdGenerator();
    
    // ... setup LogicalDag and PhysicalPlan ...
    
    PhysicalVertex physicalVertex =
            physicalPlan.getPipelineList().get(0).getPhysicalVertexList().get(0);
    physicalVertex.startPhysicalVertex();
    
    // Delete state entry
    runningJobState.remove(physicalVertex.getTaskGroupLocation());
    
    // Use CountDownLatch to simulate concurrent calls
    java.util.concurrent.CountDownLatch startLatch = new java.util.concurrent.CountDownLatch(1);
    java.util.concurrent.CountDownLatch doneLatch = new java.util.concurrent.CountDownLatch(2);
    
    ExecutorService executor = Executors.newFixedThreadPool(2);
    
    // Thread 1: Try to update to FAILING
    executor.submit(() -> {
        try {
            startLatch.await();
            physicalVertex.makeTaskGroupFailing(new RuntimeException("thread 1"));
        } catch (Exception e) {
            e.printStackTrace();
        } finally {
            doneLatch.countDown();
        }
    });
    
    // Thread 2: Try to update to CANCELING
    executor.submit(() -> {
        try {
            startLatch.await();
            physicalVertex.cancel();
        } catch (Exception e) {
            e.printStackTrace();
        } finally {
            doneLatch.countDown();
        }
    });
    
    startLatch.countDown(); // Start both threads simultaneously
    doneLatch.await(5, TimeUnit.SECONDS);
    
    // Verify: Should reach some terminal state, should not throw exception or deadlock
    ExecutionState finalState = physicalVertex.getExecutionState();
    Assertions.assertTrue(
            finalState.isEndState(),
            "Final state should be terminal, but was: " + finalState);
}

Rationale:

1. First test: verifies the extreme scenario when currExecutionState is null, ensuring code robustness
2. Second test: verifies thread safety under concurrent scenarios, ensuring the synchronized keyword takes effect
3. Improves test coverage and reduces the risk of unexpected situations in production environment
4. Test code is clear, easy to understand and maintain

---

Issue 4: Missing JavaDoc documentation updates

Location: PhysicalVertex.java:351

Related Context:

• JavaDoc of PhysicalVertex class: PhysicalVertex.java:67-72
• JavaDoc of other methods: such as resetExecutionState(), updateStateByExecutionService()

Issue Description:
The updateTaskState method adds important fault tolerance logic (handling stateEntryMissing), but does not update or add JavaDoc to explain this behavior. This will lead to:

1. Other developers not understanding why runningJobStateIMap is not updated in certain cases
2. Code maintainers may mistakenly think this is a bug and try to "fix" it
3. Developers using this API are unclear about its fault tolerance features

Potential Risks:

• Risk 1: Reduced code maintainability, subsequent developers may misunderstand the design intent
• Risk 2: May lead to inappropriate "fixes" that break the original fault tolerance logic
• Risk 3: Violates Apache project's high standards for code documentation

Impact Scope:

• Direct Impact: Code maintainability
• Indirect Impact: Future maintainers' understanding cost
• Impact Area: Single method

Severity: MINOR

Improvement Suggestions:

/**
 * Update the task state in both the distributed state map and the local state.
 * 
 * <p>This method handles the scenario where the task state entry is missing from
 * the distributed map (e.g., due to node removal during scaling down). In such cases:
 * <ul>
 *   <li>The local state ({@code currExecutionState}) is used as a fallback for the
 *       current state</li>
 *   <li>The distributed map is not updated if the entry is missing (to avoid
 *       creating orphaned entries)</li>
 *   <li>The local state is always updated to ensure the task can progress to a
 *       terminal state</li>
 * </ul>
 *
 * <p>This design ensures that task state transitions can complete even when the
 * distributed state is temporarily unavailable, preventing the task from hanging
 * in a non-terminal state.
 *
 * @param targetState the target state to transition to, must not be null
 * @see ExecutionState
 * @see #currExecutionState
 * @see #stateProcess()
 */
public synchronized void updateTaskState(@NonNull ExecutionState targetState) {
    // ... method implementation ...
}

Rationale:

1. Clearly explains the method's behavior and fault tolerance design
2. Explains why the distributed Map is not updated when stateEntryMissing
3. Provides background information (node removal scenario), helping to understand the design intent
4. Uses standard JavaDoc format, including @param, @see and other tags
5. Complies with Apache project documentation standards

---

[DanielLeens]

Thanks for working on this. I reviewed the latest head locally against the full Zeta state-transition path, including the downstream readers of runningJobStateIMap.

What This PR Fixes

• User pain: during scale-down, member removal, or similar failure windows, a job/pipeline/task state entry can disappear from runningJobStateIMap, and the old code can then fail the transition on a null state instead of continuing toward the terminal state.
• Fix approach: this PR adds local fallbacks (currJobStatus, currPipelineStatus, currExecutionState) so the in-memory object can still progress when the distributed entry is missing.
• One-line summary: the fallback direction is useful, but the current head still does not repair the shared distributed state source, so the cluster-level contract is still incomplete.

Runtime Chain Rechecked

task / pipeline / job transition
  -> PhysicalVertex.updateTaskState()
  -> SubPlan.updatePipelineState()
  -> PhysicalPlan.updateJobState()
      -> fallback to local curr*Status when runningJobStateIMap.get(...) is null

downstream shared-state readers
  -> JobHistoryService.toJobStateMapper()
      -> still reads runningJobStateIMap directly
  -> CoordinatorService.restoreJobFromMasterActiveSwitch()
      -> removes running job info if jobState is still null

Findings

Issue 1: the fallback advances only the local object, but does not repair the missing entry in runningJobStateIMap

• Location:
  • PhysicalPlan.java:255-270,299-313
  • SubPlan.java:390-404
  • PhysicalVertex.java:397-410
  • JobHistoryService.java:267-299
  • CoordinatorService.java:719-721
• Why this is a blocker:
  • the state machine can keep moving locally, but other cluster paths still read the distributed map directly
  • that means history/detail views and active-master recovery can still observe missing state
• Severity: High

Issue 2: missing timestamp entries are now only warned and skipped, while the state can still keep moving

• Location:
  • PhysicalPlan.java:275-289
  • SubPlan.java:449-460
  • PhysicalVertex.java:474-486
• Why this matters:
  • state and state timestamps can diverge again even after the fallback
• Severity: Medium

Issue 3: the new regression tests prove local completion, but not distributed-state repair or restore safety

• Location:
  • TaskTest.java:314-497
• Why this matters:
  • the tests do not assert that runningJobStateIMap was repaired
  • they also do not cover pipeline-state loss or active-master restore after the fallback
• Severity: Medium

Merge Conclusion

Conclusion: can merge after fixes

1. Blocking items

• Issue 1: please make the fallback repair the distributed state entry, not only the local object.

2. Non-blocking follow-up

• Issue 2: add a concrete repair strategy for missing timestamp arrays.
• Issue 3: extend the regression coverage to distributed-state repair and restore paths.

Overall, this is a real bug and the PR is moving in the right direction. But for Zeta, the distributed map is part of the runtime contract, not just an implementation detail, and the current head still leaves that contract partially broken.

[DanielLeens]

Thanks for working on this. I re-reviewed the latest head from scratch again against the full Zeta state-transition path and the downstream readers of runningJobStateIMap.

What this PR solves

• User pain: during scale-down, member removal, or similar failure windows, a job / pipeline / task state entry can disappear from runningJobStateIMap, and the old code can then fail the transition on a null state instead of continuing toward a terminal state.
• Fix approach: the PR caches local fallback states (currJobStatus, currPipelineStatus, currExecutionState) and recreates the missing map entry when the transition runs.
• One-line summary: the fallback direction is useful, but the current head still does not repair the shared distributed-state contract end to end.

Full runtime path I checked

state transition on the active master / worker
  -> PhysicalPlan.updateJobState(...) [PhysicalPlan.java:308-330]
  -> SubPlan.updatePipelineState(...) [SubPlan.java:345-406]
  -> PhysicalVertex.updateTaskState(...) [PhysicalVertex.java:351-417]
      -> when the distributed state entry is missing
          -> fall back to local curr* state
          -> recreate the map entry with the target state

downstream readers that still consume the distributed map directly
  -> JobHistoryService.toJobStateMapper(...) [JobHistoryService.java:257-306]
      -> reads runningJobStateIMap for job / pipeline / task states
  -> CoordinatorService.restoreJobFromMasterActiveSwitch(...) [CoordinatorService.java:506-521]
      -> still branches on the distributed job-state entry directly

Findings

Issue 1: the fallback only repairs the transition caller, but it still leaves the broader distributed-state contract incomplete

• Location:
  • seatunnel-engine/seatunnel-engine-server/src/main/java/org/apache/seatunnel/engine/server/dag/physical/PhysicalPlan.java:259-330
  • seatunnel-engine/seatunnel-engine-server/src/main/java/org/apache/seatunnel/engine/server/dag/physical/SubPlan.java:345-406
  • seatunnel-engine/seatunnel-engine-server/src/main/java/org/apache/seatunnel/engine/server/dag/physical/PhysicalVertex.java:351-417
  • seatunnel-engine/seatunnel-engine-server/src/main/java/org/apache/seatunnel/engine/server/master/JobHistoryService.java:257-306
  • seatunnel-engine/seatunnel-engine-server/src/main/java/org/apache/seatunnel/engine/server/CoordinatorService.java:506-521
• Why this is still a blocker:
  the current fix recreates the missing entry only inside the immediate transition path, but other cluster readers still fetch the distributed map directly. So the state machine can move locally while history rendering or active-master recovery still observes null and reasons from an incomplete cluster view.
• Risk:
  incorrect job / pipeline / task state can still leak into history and recovery flows, especially in the same scale-down / master-switch window this PR is targeting.
• Better fix:
  either restore the missing entry at the shared state-management boundary before other readers run, or teach the downstream readers to consume the same fallback contract consistently instead of mixing cached local state and raw map reads.
• Severity: High

Tests

• The new regression tests cover the local transition methods when the state entry disappears.
• They still do not exercise JobHistoryService or CoordinatorService on the same failure window, so the end-to-end distributed-state contract is still unproven.

CI

• Build is currently queued on the latest head.
• Even if CI turns green, I would still treat Issue 1 as a real code blocker on the current revision.

Conclusion: merge after fixes

1. Blocking items

• Issue 1: the PR repairs only the immediate transition caller, but the shared distributed-state readers still observe missing state directly.

2. Suggested follow-up

• After fixing the shared-state contract, I would add one regression that covers either history rendering or active-master recovery on the same missing-entry window.

The latest head moves in the right direction, but I still do not think the current revision closes the cluster-level state contract safely enough to merge.

[DanielLeens]

Thanks for the update. I rechecked the current head against the latest dev, and this branch is still behind upstream.

• base branch: dev
• head SHA: 82d9abb6a42291a1a72a1fec2b9dc848ad433382
• compare status: diverged
• ahead_by: 2
• behind_by: 15
• mergeable_state: unknown
• CI / merge gate snapshot: failing (failing: Build [failure])

Because the PR is still behind the latest base, any CI result and merge gate signal here is mixed with upstream drift. The lowest-cost next step is to sync with the latest dev and rerun CI first. If anything still fails after the sync, I'm happy to take another look on top of the updated head.

[DanielLeens]

Thanks for the update. I rechecked the current head against the latest dev, and the source-level blockers from my previous Daniel review still stand on this branch.

• base branch: dev
• head SHA: 82d9abb6a42291a1a72a1fec2b9dc848ad433382
• compare status: diverged
• ahead_by: 2
• behind_by: 23
• CI / merge gate snapshot: Build FAILURE

The current failing CI signal does not look actionable on top of this stale head yet, so the lowest-cost next step is to sync with the latest dev and rerun CI first. If anything still fails after the sync, I am happy to take another look on top of the updated head.

[DanielLeens]

Thanks for the focused update. I re-reviewed the latest head and found no remaining source-level merge blocker. The code changes look good to me.

[DanielLeens]

Rechecked the current head 62b84bb4f2c8 against the latest dev after the CI / merge-gate fact changed.

Runtime path rechecked

state transition on the active master / worker
  -> PhysicalPlan.updateJobState(...) [PhysicalPlan.java:308-330]
  -> SubPlan.updatePipelineState(...) [SubPlan.java:345-406]
  -> PhysicalVertex.updateTaskState(...) [PhysicalVertex.java:351-417]
      -> when the distributed state entry is missing
          -> fall back to local curr* state
          -> recreate the map entry with the target state

downstream readers that still consume the distributed map directly
  -> JobHistoryService.toJobStateMapper(...) [JobHistoryService.java:257-306]
      -> reads runningJobStateIMap for job / pipeline / task states
  -> CoordinatorService.restoreJobFromMasterActiveSwitch(...) [CoordinatorService.java:506-521]
      -> still branches on the distributed job-state entry directly

Merge conclusion

Conclusion: source blocker cleared from Daniel's side; sync latest dev and rerun CI first

1. Blocking items

• None reopened from Daniel's source review on this unchanged head.
• This branch is behind upstream (compare status: diverged, ahead_by=3, behind_by=149), and the current failing gate is FAILURE gate: Build: https://github.com/apache/seatunnel/runs/86810034333

2. Lowest-cost next step

• Please sync with the latest dev and rerun the failing gate first.
• If it still fails on the updated head, I can help narrow down the new failure from there.

Daniel only has READ permission on this repository, so a write-capable maintainer may still need to handle the final approval / dismissal step once the gate is clear.

[dybyte]

+1 if CI passes

[DanielLeens]

Thanks @dybyte — aligned.

From Daniel's side, the earlier source-level blockers on this PR were already closed on the current head. What remains here is still the merge gate rather than a reopened engine logic issue from my review path.

This branch is still diverged from dev (behind_by=151), and the current failing required lane is Build. So the lowest-cost next step is still to sync the latest dev and rerun the failing gate first. If it still fails after the sync, please share the refreshed run link and I can help narrow down whether the new failure is really tied to this PR or just upstream drift.

[SEZ9]

Thanks for working on this. I re-reviewed the latest head 3e019e75a372 from scratch.

What this PR fixes

• User pain: During node scale-down/member removal the task state entry can vanish from runningJobStateIMap, so PhysicalVertex.updateTaskState bailed out early, taskFuture never completed, and SubPlan pipeline-completion counting hung indefinitely.
• Fix approach: Guard the missing entry with stateEntryMissing, continue the local transition using currExecutionState, and skip only the distributed map write; the PR also substantially reworks CoordinatorService cleanup/restore to run under a per-job pendingJobCleanupIMap lock fence, adds repairMissingJobStateForRestore/initializeMissingJobState for master-switch recovery, and removes both state and timestamp keys under a per-key lock in cleanupPendingJobStateMaps.
• One-line summary: The core fix and regression test look sound, but the diff is far larger than the description implies — the +468/-93 CoordinatorService rewrite (lock fencing, owner-generation checks, missing-state repair) is a significant concurrency change that needs its own justification and test coverage, and the flagged PhysicalPlan.java public API change plus missing doc updates should be resolved before merge.

Runtime chain I rechecked

1. Task-state fix path (described change):
   TaskExecutionService -> PhysicalVertex.updateTaskState()  PhysicalVertex.java
     stateEntryMissing guard -> transition from currExecutionState instead of early return
     -> end-state check (null-safe) -> taskFuture.complete()
       -> SubPlan pipeline completion counting unblocks  SubPlan.java

2. Cleanup path (changed in this diff):
   CoordinatorService.schedulePendingJobCleanup()  CoordinatorService.java:764
     -> cleanupScheduler.schedule -> processPendingJobCleanup(jobId)  CoordinatorService.java:767 (was processPendingJobCleanup(jobId, record))
       -> pendingJobCleanupIMap.lock(jobId)  CoordinatorService.java:791
         -> re-read record inside fence -> shouldCleanup()/isCleanupDelayElapsed()
         -> runningJobInfoIMap.remove(jobId, currentJobInfo) CAS  CoordinatorService.java:~815
         -> cleanupPendingJobStateMaps(record)  CoordinatorService.java:~890
           -> union of stateKeys + timestampKeys; per key: runningJobStateIMap.lock(key) -> remove from both maps -> unlock
         -> removePendingJobCleanupRecord(jobId, record) (conditional remove preserves newer generation)
       -> pendingJobCleanupIMap.unlock(jobId)

3. Master-switch restore path (changed in this diff):
   CoordinatorService.restoreAllRunningJobFromMasterNodeSwitch()
     -> restoreJobFromMasterActiveSwitch(jobId, jobInfo)  CoordinatorService.java:1049
       -> getRunningJobInfoWithRetry(jobId) + initializationTimestamp ownership check (new early return)
       -> jobState == null branch:
         -> getOwnedPendingCleanup(jobId, jobInfo) -> schedulePendingJobCleanup() if fenced
         -> repairMissingJobStateForRestore(jobId, jobInfo)  CoordinatorService.java:~1149
           -> pendingJobCleanupIMap.lock(jobId) -> owner re-check + containsKey(jobId) guard
           -> RetryUtils.retryWithException -> initializeMissingJobState(jobId, jobInfo)  CoordinatorService.java:~1185
             -> runningJobStateIMap.lock(jobId) -> timestamp-first write, then CREATED state insert
       -> jobState end-state branch -> schedulePendingJobCleanup() or terminal-zombie handling
       -> active branch: pendingJobCleanupIMap.lock(jobId) -> owner re-check -> jobMaster.init(..., true)  CoordinatorService.java:~1111 -> unlock
       -> pendingJobMasterMap enqueue as PendingSourceState.RESTORE

Findings

Issue 1: Possible public API change in seatunnel-engine/seatunnel-engine-server/src/main/java/org/apache/seatunnel/engine/server/dag/physical/PhysicalPlan.java

• Location: seatunnel-engine/seatunnel-engine-server/src/main/java/org/apache/seatunnel/engine/server/dag/physical/PhysicalPlan.java
• Why it matters: A public/protected signature appears to be removed or changed. Verify backward compatibility for downstream connectors.
• Severity: High

Issue 2: Config/option changes without doc updates

• Why it matters: Consider updating docs/example configs for the new/changed options.
• Severity: Low

Review conclusion

Conclusion: can merge after the blocking items are fixed

1. Blocking items

• Issue 1: Possible public API change in seatunnel-engine/seatunnel-engine-server/src/main/java/org/apache/seatunnel/engine/server/dag/physical/PhysicalPlan.java

2. Suggested (non-blocking) follow-ups

• Issue 2: Config/option changes without doc updates

Nice progress overall — once the blocking points above are addressed this should be in good shape. Happy to discuss.

[DanielLeens]

Thanks for the update. I re-reviewed the current head after the final conflict resolution.

The earlier distributed-state blocker is closed: missing job/pipeline/task entries are repaired through the shared state path with race-safe updates, and downstream history/master-recovery readers no longer rely on a caller-local transition fallback alone. The focused tests cover the repaired shared-state behavior.

I do not see a source-level blocker. The required Build is failing, so the technical verdict is clear_waiting_ci. After CI is green, the remaining gate is one maintainer review.

[dybyte]

+1 if CI passes.
I think it would be better to update the PR description to reflect the broader scope of the changes.

[DanielLeens]

Thanks for the update. I re-reviewed the current head 9465ab191cc6 from scratch, including the distributed job / pipeline / task state transition path and the checkpoint ready-to-close persistence path.

I noticed @dybyte already left an approval with the condition that CI passes. After tracing the latest code, I largely agree with the source-level direction: I do not see a new source-side blocker from Daniel's side now. The previous distributed-state concern is addressed by moving the missing-state repair into the shared DistributedStateTransition contract instead of relying on caller-local fallbacks only.

What this PR fixes

• User pain: during node scale-down, member removal, or master recovery windows, a job / pipeline / task entry can disappear from runningJobStateIMap; the old path could then stop the transition instead of continuing toward a terminal or recoverable state.
• Fix approach: job / pipeline / task transitions now go through a shared distributed-state helper, with timestamp-first persistence, terminal-state priority, owner-token checks, and a cleanup fence when a missing state key needs to be recreated.
• In one sentence: this makes Zeta state recovery much safer when distributed runtime state and cleanup race with each other.

Full execution path I checked

Job startup / recovery
  -> PhysicalPlanGenerator builds PhysicalPlan / SubPlan / PhysicalVertex
  -> PhysicalVertex(...) calls DistributedStateTransition.initialize(...)
      -> initializes INITIALIZING / CREATED timestamps
      -> creates or reads the task state from runningJobStateIMap

Task state transition
  -> PhysicalVertex.updateTaskState(target)
      -> DistributedStateTransition.transition(...)
          -> lock the task state key
          -> if the state entry is missing: release the state lock, acquire the cleanup fence, then lock the state key again
          -> check JobMaster ownership / pending-cleanup fence
          -> persist the timestamp before the state
          -> putIfAbsent / replace runningJobStateIMap
          -> keep terminal states from being overwritten by late non-terminal writers
      -> stateProcess() / taskFuture completion
      -> SubPlan aggregates task completion

Pipeline / job transition
  -> SubPlan.updatePipelineState(...)
  -> PhysicalPlan.updateJobState(...)
      -> reuse the same distributed transition semantics
      -> JobHistoryService / JobMaster recovery reads durable state instead of caller-local fallback only

Checkpoint close path
  -> CheckpointCoordinator.readyToClose(taskLocation)
      -> updateReadyToCloseStartingTask()
      -> persistReadyToCloseStartingTask()
          -> acquire the cleanup fence only when the ready-to-close key must be created
          -> persist readyToCloseStartingTask
      -> when all source tasks are ready, trigger the final COMPLETED_POINT checkpoint

Review result

I did not find a new source-level blocker in the current code. The compatibility impact is safe: no public API, config option, default value, protocol, serialization format, or user-visible behavior is changed. The extra Hazelcast map locks / retries are on lifecycle state-transition paths rather than the record hot path, and the lock ordering is explicit.

The added tests cover the important missing-state, late-writer, cleanup-fence, recovery-reader, and checkpoint ready-to-close paths. I also checked the new UT stability pattern: the new tests mainly use controlled latches, Awaitility, mocked map behavior, and executor shutdown; I do not see a high-risk flaky-test pattern introduced by this PR.

Current merge gate

There is still one blocker before merge:

• CI-1: Build is still failing, and the failing engine-v2-it lane overlaps this PR's area. The current Build check is https://github.com/apache/seatunnel/runs/90112649649. In the fork job, engine-v2-it (8, ubuntu-latest) fails in CheckpointEnableIT.testZetaStreamingCheckpointInterval and CheckpointEnableIT.testZetaStreamingCheckpointNoInterval with Awaitility timeouts while checking the restored checkpoint sink output. Because this PR changes the Zeta checkpoint / distributed-state recovery path, I cannot classify that lane as unrelated noise yet.

The branch is also currently behind the latest dev (behind_by=6, compare status: diverged), so the lowest-cost next step is to sync with the latest dev and rerun Build first. If the same engine-v2-it checkpoint timeout still appears after the sync, I would treat that as the next thing to debug against this PR's checkpoint/state-transition changes.

5. Merge conclusion

Conclusion: merge after fixes

1. Blocking item

• CI-1: the required Build check must pass, or the engine-v2-it checkpoint failure must be proven unrelated after syncing with the latest dev.

2. Non-blocking suggestion

• +1 to @dybyte's note that the PR description should reflect the broader distributed-state recovery scope. This is helpful for maintainers, but I would not treat it as a code blocker.

Overall, the source changes look solid to me now. Once the Build gate is green, especially the Zeta checkpoint lane, I am comfortable with this from Daniel's side.

[davidzollo]

本轮已继续推进这个 PR 的 merge readiness：

Pushed commit: 7278866

处理内容：

• 将 PR head 同步到最新 upstream/dev；
• 本次 merge 无代码冲突；
• 远端回读后 PR 仍为 mergeable=MERGEABLE，reviewDecision=APPROVED。

本地验证边界：按 Apache SeaTunnel 本地规则，本轮未执行 compile/test/package/E2E/Docker/service；只执行格式化/格式检查：
./mvnw -nsu -DskipTests -DskipIT=true -Dmaven.gitcommitid.skip=true spotless:apply spotless:check
结果：BUILD SUCCESS。

当前功能验证继续以新 head 的 GitHub CI 为准；PR checks 已重新触发，Notify/labeler 正在运行，Build check 仍在生成/排队。

[SEZ9]

Thanks for working on this. I re-reviewed the latest head 7278866376f0 from scratch.

What this PR fixes

• User pain: During node scale-down/member removal, a task's state entry can vanish from runningJobStateIMap, causing PhysicalVertex.updateTaskState to bail out early so taskFuture never completes and SubPlan pipeline completion counting hangs forever.
• Fix approach: Guard the missing-entry case in PhysicalVertex.updateTaskState (continue local transition via currExecutionState, null-safe end-state checks, skip the distributed write only when the entry is gone), plus a substantial hardening pass in CoordinatorService: processPendingJobCleanup now runs under a per-job pendingJobCleanupIMap.lock(jobId) fence, cleanupPendingJobStateMaps removes the key-set union of state/timestamp keys under per-key locks, and master-switch restore gains ownership re-validation plus repairMissingJobStateForRestore/initializeMissingJobState to recreate a missing job-state entry instead of blindly runningJobInfoIMap.remove(jobId).
• One-line summary: The core PhysicalVertex fix looks sound and is covered by testUpdateTaskStateWhenStateEntryMissing, but the PR title/description undersell a large (+468/-93) CoordinatorService refactor that changes cleanup locking and restore semantics — I want the flagged possible public API changes in PhysicalPlan/PhysicalVertex confirmed as intentional, the truncated initializeMissingJobState timestamp-repair path reviewed for the indeterminate-write case (IndeterminateOperationStateException is newly imported), and the concurrency additions ideally split into or at least documented as a separate concern with their own tests.

Runtime chain I rechecked

Cleanup path (changed):
  CoordinatorService.schedulePendingJobCleanup()  CoordinatorService.java:764
    -> CoordinatorService.processPendingJobCleanup(jobId)  CoordinatorService.java:788 (now re-reads record under pendingJobCleanupIMap.lock(jobId))
      -> shouldCleanup()/isCleanupDelayElapsed()  CoordinatorService.java:851-860
      -> isCleanupOwnedByCurrentJob(currentJobInfo, jobId, record)  CoordinatorService.java:868 (old jobId-only overload removed)
      -> runningJobInfoIMap.remove(jobId, currentJobInfo)  CoordinatorService.java:817 (CAS on owner generation)
      -> cleanupPendingJobStateMaps(record)  CoordinatorService.java:889 (union of stateKeys+timestampKeys, per-key runningJobStateIMap.lock/unlock, replaces removeKeys())
      -> removePendingJobCleanupRecord(jobId, record)  CoordinatorService.java:922

Master-switch restore path (changed):
  CoordinatorService.restoreAllRunningJobFromMasterNodeSwitch()
    -> restoreJobFromMasterActiveSwitch(jobId, jobInfo)  CoordinatorService.java:1049
      -> getRunningJobInfoWithRetry(jobId) + initializationTimestamp ownership check  CoordinatorService.java:1050-1056
      -> on jobState == null: getOwnedPendingCleanup() -> schedulePendingJobCleanup(), else repairMissingJobStateForRestore()  CoordinatorService.java:1072-1083
        -> repairMissingJobStateForRestore()  CoordinatorService.java:1148 (pendingJobCleanupIMap.lock(jobId), RetryUtils.retryWithException)
          -> initializeMissingJobState(jobId, jobInfo)  CoordinatorService.java:1188 (runningJobStateIMap.lock(jobId), timestamp-first write, then CREATED)
      -> jobMaster.init(...) now inside pendingJobCleanupIMap.lock(jobId) with owner + cleanup-record re-check  CoordinatorService.java:1111-1132

Task-state path (per PR description, not shown in diff excerpt):
  TaskExecutionService state report -> PhysicalVertex.updateTaskState()  PhysicalVertex.java
    -> stateEntryMissing guard -> transition via currExecutionState (skips runningJobStateIMap write when entry absent)
    -> end-state -> taskFuture.complete(...) -> SubPlan pipeline completion counting
    verified by TaskTest.testUpdateTaskStateWhenStateEntryMissing (removes IMap entry, FAILING -> FAILED, asserts terminal state + future completion)

Findings

Issue 1: Possible public API change in seatunnel-engine/seatunnel-engine-server/src/main/java/org/apache/seatunnel/engine/server/dag/physical/PhysicalPlan.java

• Location: seatunnel-engine/seatunnel-engine-server/src/main/java/org/apache/seatunnel/engine/server/dag/physical/PhysicalPlan.java
• Why it matters: A public/protected signature appears to be removed or changed. Verify backward compatibility for downstream connectors.
• Evidence: seatunnel-engine/seatunnel-engine-server/src/main/java/org/apache/seatunnel/engine/server/dag/physical/PhysicalVertex.java
• Severity: High

Issue 2: Config/option changes without doc updates

• Why it matters: Consider updating docs/example configs for the new/changed options.
• Severity: Low

Review conclusion

Conclusion: can merge after the blocking items are fixed

1. Blocking items

• Issue 1: Possible public API change in seatunnel-engine/seatunnel-engine-server/src/main/java/org/apache/seatunnel/engine/server/dag/physical/PhysicalPlan.java

2. Suggested (non-blocking) follow-ups

• Issue 2: Config/option changes without doc updates

Nice progress overall — once the blocking points above are addressed this should be in good shape. Happy to discuss.

[davidzollo]

Correction for the compatibility evidence above: the private helper names are PhysicalPlan.updateStateInfo, PhysicalPlan.updateStateTimestamps, and PhysicalVertex.updateStateTimestamps.

The compatibility conclusion is unchanged:

• PhysicalPlan and PhysicalVertex constructor changes only add generic type parameters to existing Hazelcast IMap arguments.
• Java erasure keeps the JVM descriptor as IMap, IMap, so existing callers do not lose a different runtime constructor descriptor.
• The removed helpers are private implementation details, not public/protected API.

The PR description has been updated to document the broader CoordinatorService cleanup/restore and DistributedStateTransition scope.

[davidzollo]

本轮已针对最新 CI 失败做修复：Build 里 engine-v2-it 的 CheckpointEnableIT.testZetaStreamingCheckpointInterval 和 testZetaStreamingCheckpointNoInterval 在 restore 后等待 sink assert 超时。原因是 #10551 新增的 generation fence 只按 jobId 检查 pending cleanup；同一个 jobId 立即从 savepoint restore 时，旧 generation 的延迟 cleanup 记录还可能存在，导致新 generation 的 checkpoint/ready-to-close 状态写入被误阻断。\n\n修复：JobMaster.isStatePersistenceAllowed() 现在只拦截 ownerInitializationTimestamp 等于当前 generation 的 cleanup record；旧 generation cleanup 不再 fence 当前 restore generation，同时保留当前 generation terminal cleanup 对 late writer 的防护。已补充 JobMasterTest.testOldGenerationCleanupDoesNotFenceCurrentGeneration 覆盖这个场景。\n\n本地按任务边界只执行格式验证：./mvnw -pl seatunnel-engine/seatunnel-engine-server spotless:apply 和 spotless:check 均通过；未运行本地 compile/test/E2E。已推送到 PR head，等待 GitHub CI 重跑。

[DanielLeens]

Thanks for the update. I rechecked the latest head 36c296679193 again, including the checkpoint completion path and the cleanup-generation fence in JobMaster.

After tracing the current delta, I still do not see a new source-side blocker from Daniel's side now. The latest changes look like focused hardening on top of the earlier distributed-state repair, not a reopened regression.

Runtime path I rechecked

state transition / recovery
  -> DistributedStateTransition
      -> repair missing runtime state entries
  -> JobMaster state persistence fence
      -> ignore stale cleanup records from older generations

checkpoint close path
  -> CheckpointCoordinator.readyToClose(...)
  -> CheckpointCoordinator.isNoErrorCompleted()
      -> null-safe completed/suspend state matrix

Current finding

Issue 1: the old CI fact is stale now, and the current head still needs a fresh Build signal on top of the latest dev

• compare status: diverged
• behind by: 10
• latest Build: https://github.com/apache/seatunnel/runs/91136485502

On the current head, the first failing signal has shifted away from the older engine-v2-it focus and is now failing first in a seatunnel-ci-tools unit-test fork error, plus multiple broader lanes. So the older CI note on this PR is no longer the right next-step guidance.

The lowest-cost next step is to sync with the latest dev and rerun Build on the synced head. If the synced head still reproduces a Zeta checkpoint / state-transition lane tied to this PR's area, that would be the right next thing to debug.

Merge conclusion

Conclusion: can merge after fixes

1. Blocking items

• Sync the latest dev and rerun Build on the updated head.

2. Non-blocking suggestions

• No new source-side change is required from Daniel in this round.

Rechecked the latest head: I do not see a new code-side blocker from Daniel's side now. If GitHub still shows older historical changes requested, please treat that as review-summary residue rather than a reopened blocker from this round.