[Fix][Zeta] Converge notifyCompleted failure handling in CheckpointCoordinator

[hyboll]

Fixes: #10655

Purpose of this pull request

Fix avoid NPE in completePendingCheckpoint when notifyCompleted clears pendingCheckpoints

Does this PR introduce any user-facing change?

No

How was this patch tested?

Test case is added in the CheckpointCoordinatorTest.testCompletePendingCheckpointShouldNotThrowNPEWhenNotifyCompletedClearsPendingMap

Check list

• If any new Jar binary package adding in your PR, please add License Notice according
  New License Guide
• If necessary, please update the documentation to describe the new feature. https://github.com/apache/seatunnel/tree/dev/docs
• If necessary, please update incompatible-changes.md to describe the incompatibility caused by this PR.
• If you are contributing the connector code, please check that the following files are updated:
  1. Update plugin-mapping.properties and add new connector information in it
  2. Update the pom file of seatunnel-dist
  3. Add ci label in label-scope-conf
  4. Add e2e testcase in seatunnel-e2e
  5. Update connector plugin_config

[chl-wxp]

Why is it disabled on the Windows system?

[hyboll]

Under the Windows system, this test method will throw a FileNotFoundException because HADOOP_HOME is not found. This bug fix only added an if condition and has nothing to do with the environment, so I disabled Windows. If necessary, I can add support for the Windows system.

[DanielLeens]

I think this patch fixes the NPE symptom, but it still leaves the failure path in an inconsistent state.

notifyCompleted() catches internal failures and calls handleCoordinatorError() (CheckpointCoordinator.java:372-393), which already:

• updates the coordinator status to FAILED
• clears pendingCheckpoints
• resets pendingCounter to 0
• shuts the coordinator down via cleanPendingCheckpoint() (CheckpointCoordinator.java:860-893)

After control returns to completePendingCheckpoint(), the method still continues with the normal completion flow (CheckpointCoordinator.java:1004-1019). That means:

• pendingCounter.decrementAndGet() now runs after the cleanup path has already reset the counter
• for final checkpoints, isCompleted() can still become true and overwrite the failure path with FINISHED / SUSPEND

So the null-check removes one crash, but the coordinator can still report a completed final checkpoint after notifyCompleted() has already failed.

I think we need an early exit once notifyCompleted() has triggered cleanup, for example by returning immediately when the coordinator is already failed/shutdown, or by making notifyCompleted() propagate a failure signal back to completePendingCheckpoint().

[hyboll]

I think this patch fixes the NPE symptom, but it still leaves the failure path in an inconsistent state.

notifyCompleted() catches internal failures and calls handleCoordinatorError() (CheckpointCoordinator.java:372-393), which already:

• updates the coordinator status to FAILED
• clears pendingCheckpoints
• resets pendingCounter to 0
• shuts the coordinator down via cleanPendingCheckpoint() (CheckpointCoordinator.java:860-893)

After control returns to completePendingCheckpoint(), the method still continues with the normal completion flow (CheckpointCoordinator.java:1004-1019). That means:

• pendingCounter.decrementAndGet() now runs after the cleanup path has already reset the counter
• for final checkpoints, isCompleted() can still become true and overwrite the failure path with FINISHED / SUSPEND

So the null-check removes one crash, but the coordinator can still report a completed final checkpoint after notifyCompleted() has already failed.

I think we need an early exit once notifyCompleted() has triggered cleanup, for example by returning immediately when the coordinator is already failed/shutdown, or by making notifyCompleted() propagate a failure signal back to completePendingCheckpoint().

Thanks for the suggestion. I've looked into the code again and I'm thinking of making notifyCompleted() return a boolean to handle the flow control. One quick question though: since there are three places calling this method, I plan to add the conditional checks to the other two call sites as well. Does that sound feasible to you?

[DanielLeens]

@hyboll Yes, that direction makes sense to me. The important part is to propagate the failure signal consistently to all three call sites of notifyCompleted(), not just completePendingCheckpoint().

Right now the same catch-and-cleanup path can be hit from:

• completePendingCheckpoint() (CheckpointCoordinator.java:1003)
• allTaskReady() (CheckpointCoordinator.java:361)
• restoreCoordinator() (CheckpointCoordinator.java:493)

So if notifyCompleted() fails and handleCoordinatorError() has already switched the coordinator into the failed / shutdown path, each caller needs to stop its normal follow-up logic immediately.

In particular, allTaskReady() and restoreCoordinator() should not continue into scheduling / triggering the next checkpoint after a failed notify, and completePendingCheckpoint() still needs the early exit before pendingCounter.decrementAndGet() and the final FINISHED / SUSPEND transition.

So a boolean return value is fine, or an explicit exception / status check, as long as the contract is: once notifyCompleted() has triggered cleanup, the caller must bail out and not continue the success path.

[DanielLeens]

I re-checked the latest HEAD locally.

The blocker from my previous review looks addressed now:

• notifyCompleted() returns a failure signal and allTaskReady() / restoreCoordinator() / completePendingCheckpoint() all bail out on that signal (CheckpointCoordinator.java:361-367, 497-500, 1009-1016)
• the new regression tests now cover all three caller paths (CheckpointCoordinatorTest.java:427-575)

I do not see the earlier inconsistent-success-path issue in the current revision.
Thanks for following through on the full control-flow fix.

[dybyte]

LGTM. Thanks @hyboll