nielifeng opened a new issue, #11563:
URL: https://github.com/apache/seatunnel/issues/11563
### Search before asking
- [x] I searched existing issues and found no duplicate. This is different
from
#10655, which failed in `completePendingCheckpoint()`, and #9708, which
failed
in `reportCheckpointErrorFromTask()`.
### What happened
On current `dev`, restarting a Zeta Worker while bounded BATCH jobs were
running
caused one job to fail during pipeline restore with a `NullPointerException`
in
`CheckpointCoordinator.isNoErrorCompleted()`.
Generic reproduction:
1. Run bounded BATCH jobs on a separated Zeta Master/Worker deployment.
2. Stop one Worker process abruptly and allow the service manager to restart
it.
3. Wait for the Worker to rejoin and the affected pipeline to restore.
4. Observe that a job can fail in `isNoErrorCompleted()` instead of
continuing
through the existing restore path.
Expected behavior:
- Worker recovery should not fail a job because a checkpoint coordinator
state
entry has already been cleaned.
- Missing checkpoint state should be treated as not completed, allowing the
existing restore logic to continue.
Root-cause analysis:
`isNoErrorCompleted()` reads the coordinator state from
`runningJobStateIMap` and
then invokes `status.equals(...)`:
```java
CheckpointCoordinatorStatus status =
(CheckpointCoordinatorStatus)
runningJobStateIMap.get(checkpointStateImapKey);
return latestCompletedCheckpoint.getCheckpointType().isFinalCheckpoint()
&& (status.equals(CheckpointCoordinatorStatus.FINISHED)
|| status.equals(CheckpointCoordinatorStatus.SUSPEND))
&& !latestCompletedCheckpoint.isRestored();
```
After #10687, checkpoint state transitions intentionally skip persistence
when
the state entry has already been cleaned. The Worker recovery path can
therefore
have a non-null `latestCompletedCheckpoint` while
`runningJobStateIMap.get(...)`
returns null. `isNoErrorCompleted()` still dereferences that nullable state.
A minimal local fix uses constant-side equality so missing state returns
false
and follows the existing restore path:
```java
CheckpointCoordinatorStatus.FINISHED.equals(status)
|| CheckpointCoordinatorStatus.SUSPEND.equals(status)
```
A regression covering a missing checkpoint state was added locally.
`CheckpointCoordinatorTest` passed 14/14, and the related targeted test
passed
1/1. The fix has not yet been submitted as a PR or cluster-validated.
This was discovered while validating #10997, but the failure is in the
checkpoint recovery path and is independent of `ReportMetricsOperation`
offloading.
### SeaTunnel Version
`3.0.0-SNAPSHOT`, commit `dbead9b`
### SeaTunnel Config
```conf
seatunnel:
engine:
job-schedule-strategy: WAIT
slot-service:
dynamic-slot: false
slot-allocate-strategy: SLOT_RATIO
checkpoint:
interval: 30000
timeout: 120000
min-pause: 5000
storage:
type: hdfs
```
Job configuration:
```conf
env {
job.mode = "BATCH"
parallelism = 1
}
source {
FakeSource {
plugin_output = "benchmark"
}
}
sink {
Console {
plugin_input = "benchmark"
log.print.data = false
}
}
```
### Running Command
```shell
# Submit bounded batch jobs through the REST submit-job API.
# On one Worker host:
pid=$(pgrep -fo org.apache.seatunnel.core.starter.seatunnel.SeaTunnelServer)
kill -9 "$pid"
# Allow the service manager to restart only this Worker.
```
### Error Exception
```log
Restore pipeline ... error with exception:
java.lang.NullPointerException
at
org.apache.seatunnel.engine.server.checkpoint.CheckpointCoordinator.isNoErrorCompleted(CheckpointCoordinator.java:1399)
at
org.apache.seatunnel.engine.server.checkpoint.CheckpointManager.isCompletedPipeline(CheckpointManager.java:260)
at
org.apache.seatunnel.engine.server.dag.physical.SubPlan.restorePipeline(SubPlan.java:520)
at
org.apache.seatunnel.engine.server.dag.physical.SubPlan.stateProcess(SubPlan.java:742)
at
org.apache.seatunnel.engine.server.dag.physical.SubPlan.updatePipelineState(SubPlan.java:397)
at
org.apache.seatunnel.engine.server.dag.physical.SubPlan.lambda$addPhysicalVertexCallBack$4(SubPlan.java:229)
at
java.util.concurrent.CompletableFuture$UniAccept.tryFire(CompletableFuture.java:714)
at
java.util.concurrent.CompletableFuture$Completion.run(CompletableFuture.java:478)
at org.apache.seatunnel.api.tracing.MDCRunnable.run(MDCRunnable.java:43)
at
java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1128)
at
java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:628)
at java.lang.Thread.run(Thread.java:829)
```
The same recovery path can log that the checkpoint state has already been
cleaned before the job reaches its terminal state.
### Zeta or Flink or Spark Version
Zeta `3.0.0-SNAPSHOT`, Hazelcast `5.1`
### Java or Scala Version
OpenJDK `11`
### Screenshots
Not applicable.
### Are you willing to submit PR?
- [x] Yes I am willing to submit a PR!
### Code of Conduct
- [x] I agree to follow this project's [Code of
Conduct](https://www.apache.org/foundation/policies/conduct)
--
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]
