Skip to content

Commit ddf3ccc

Browse files
davidzolloDanielLeens
davidzollo
authored and
DanielLeens
committed
[Fix][Engine] Continue task transition when state entry is missing
When a node is removed during scaling down, the IMap state entries may be lost. This causes NullPointerException in updateTaskState, updatePipelineState, and updateJobState, which prevents state progression and can hang pipeline completion. Changes: - PhysicalVertex.updateTaskState: fall back to local cached state when IMap entry is missing, skip distributed write, continue local transition - PhysicalVertex.updateStateTimestamps: null-safe guard for timestamps - SubPlan.updatePipelineState: same null-safe pattern as PhysicalVertex - SubPlan.updateStateTimestamps: null-safe guard for timestamps - PhysicalPlan.updateJobState: same null-safe pattern - PhysicalPlan.getJobStatus/cancelJob: null-safe fallback - PhysicalPlan.updateStateTimestamps: null-safe guard for timestamps - Added regression test testUpdateTaskStateWhenStateEntryMissing
1 parent 6e94897 commit ddf3ccc

4 files changed

Lines changed: 262 additions & 27 deletions

File tree

‎seatunnel-engine/seatunnel-engine-server/src/main/java/org/apache/seatunnel/engine/server/dag/physical/PhysicalPlan.java‎

Lines changed: 49 additions & 13 deletions
Original file line numberDiff line numberDiff line change
@@ -91,6 +91,8 @@ public class PhysicalPlan {
9191

9292
private volatile boolean isRunning = false;
9393

94+
private volatile JobStatus currJobStatus;
95+
9496
public PhysicalPlan(
9597
@NonNull List<SubPlan> pipelineList,
9698
@NonNull ExecutorService executorService,
@@ -131,6 +133,7 @@ public PhysicalPlan(
131133

132134
this.runningJobStateIMap = runningJobStateIMap;
133135
this.runningJobStateTimestampsIMap = runningJobStateTimestampsIMap;
136+
this.currJobStatus = (JobStatus) runningJobStateIMap.get(jobId);
134137
}
135138

136139
public void setJobMaster(JobMaster jobMaster) {
@@ -197,14 +200,18 @@ public void addPipelineEndCallback(SubPlan subPlan) {
197200

198201
public void cancelJob() {
199202
JobStatus jobStatus = getJobStatus();
203+
if (jobStatus == null) {
204+
log.error("{} job state is null, cannot cancel", jobFullName);
205+
return;
206+
}
200207
if (jobStatus.isEndState()) {
201208
log.warn(
202209
String.format(
203210
"%s is in end state %s, can not be cancel", jobFullName, jobStatus));
204211
return;
205212
}
206213

207-
if (((JobStatus) runningJobStateIMap.get(jobId)).ordinal() <= JobStatus.PENDING.ordinal()) {
214+
if (jobStatus.ordinal() <= JobStatus.PENDING.ordinal()) {
208215
// Tasks with the status 'INITIALIZING', 'CREATED', 'PENDING' need to be set directly to
209216
// the 'CANCELLED' state because it has not yet started running
210217
updateJobState(JobStatus.CANCELED);
@@ -249,22 +256,29 @@ public List<SubPlan> getPipelineList() {
249256
return pipelineList;
250257
}
251258

252-
private void updateStateInfo(JobStatus current, JobStatus targetState) throws Exception {
259+
private void updateStateInfo(
260+
JobStatus current, JobStatus targetState, boolean stateEntryMissing) throws Exception {
261+
if (stateEntryMissing) {
262+
log.info(
263+
"{} job state entry missing from distributed map, recreate it with target state {}",
264+
jobFullName,
265+
targetState);
266+
}
253267
RetryUtils.retryWithException(
254268
() -> {
255269
updateStateTimestamps(targetState);
256-
if (runningJobStateIMap.get(jobId) != null) {
257-
runningJobStateIMap.set(jobId, targetState);
258-
}
270+
runningJobStateIMap.set(jobId, targetState);
259271
return null;
260272
},
261273
new RetryUtils.RetryMaterial(
262274
Constant.OPERATION_RETRY_TIME,
263275
true,
264276
ExceptionUtil::isOperationNeedRetryException,
265277
Constant.OPERATION_RETRY_SLEEP));
278+
this.currJobStatus = targetState;
266279
log.info(
267280
String.format("%s turned from state %s to %s.", jobFullName, current, targetState));
281+
reportJobStateEvent(targetState);
268282
}
269283

270284
private void updateStateTimestamps(@NonNull JobStatus targetState) {
@@ -273,10 +287,11 @@ private void updateStateTimestamps(@NonNull JobStatus targetState) {
273287
Long[] stateTimestamps = runningJobStateTimestampsIMap.get(jobId);
274288
if (stateTimestamps == null) {
275289
log.warn(
276-
"{} state timestamps have already been cleaned, skip persisting transition to {}",
290+
"{} state timestamps entry missing from distributed map, "
291+
+ "recreate it for target state {}",
277292
jobFullName,
278293
targetState);
279-
return;
294+
stateTimestamps = new Long[JobStatus.values().length];
280295
}
281296
stateTimestamps[targetState.ordinal()] = System.currentTimeMillis();
282297
runningJobStateTimestampsIMap.set(jobId, stateTimestamps);
@@ -293,12 +308,26 @@ public synchronized Long getStateTimestamp(@NonNull JobStatus jobStatus) {
293308
public synchronized void updateJobState(@NonNull JobStatus targetState) {
294309
try {
295310
JobStatus current = (JobStatus) runningJobStateIMap.get(jobId);
311+
boolean stateEntryMissing = false;
296312
if (current == null) {
313+
stateEntryMissing = true;
314+
current = currJobStatus;
297315
log.warn(
298-
"{} current state is null, skip transition to {}",
316+
"{} job state entry missing from distributed map (possibly due to node "
317+
+ "removal during scaling down), using local state {} as fallback, "
318+
+ "target state: {}",
299319
jobFullName,
320+
current,
321+
targetState);
322+
}
323+
if (current == null) {
324+
current = JobStatus.CREATED;
325+
log.error(
326+
"{} both distributed and local job state are null, "
327+
+ "use {} as fallback for target state {}",
328+
jobFullName,
329+
current,
300330
targetState);
301-
return;
302331
}
303332
log.debug(
304333
"Try to update the {} state from {} to {}", jobFullName, current, targetState);
@@ -317,9 +346,7 @@ public synchronized void updateJobState(@NonNull JobStatus targetState) {
317346

318347
// Now do the actual state transition, we must update runningJobStateTimestampsIMap
319348
// first and then can update runningJobStateIMap
320-
updateStateInfo(current, targetState);
321-
reportJobStateEvent(targetState);
322-
349+
updateStateInfo(current, targetState, stateEntryMissing);
323350
stateProcess();
324351
} catch (Exception e) {
325352
log.error(ExceptionUtils.getMessage(e));
@@ -334,7 +361,16 @@ public JobImmutableInformation getJobImmutableInformation() {
334361
}
335362

336363
public JobStatus getJobStatus() {
337-
return (JobStatus) runningJobStateIMap.get(jobId);
364+
JobStatus status = (JobStatus) runningJobStateIMap.get(jobId);
365+
if (status == null) {
366+
log.warn(
367+
"{} job state entry missing from distributed map, "
368+
+ "using local cached state {} as fallback",
369+
jobFullName,
370+
currJobStatus);
371+
return currJobStatus;
372+
}
373+
return status;
338374
}
339375

340376
public String getJobFullName() {

‎seatunnel-engine/seatunnel-engine-server/src/main/java/org/apache/seatunnel/engine/server/dag/physical/PhysicalVertex.java‎

Lines changed: 29 additions & 8 deletions
Original file line numberDiff line numberDiff line change
@@ -351,13 +351,28 @@ public TaskGroup getTaskGroup() {
351351
public synchronized void updateTaskState(@NonNull ExecutionState targetState) {
352352
try {
353353
ExecutionState current = (ExecutionState) runningJobStateIMap.get(taskGroupLocation);
354+
// When a node is removed during scaling down, the IMap entry may be lost.
355+
// Fall back to the local cached state to allow state progression.
356+
boolean stateEntryMissing = false;
354357
if (current == null) {
358+
stateEntryMissing = true;
359+
current = currExecutionState;
355360
log.warn(
356-
"{} current state is null, skip transition to {}. Task execution location: {}",
361+
"{} state entry missing from distributed map (possibly due to node "
362+
+ "removal during scaling down), using local state {} as fallback, "
363+
+ "target state: {}",
357364
taskFullName,
358-
targetState,
359-
taskGroupLocation);
360-
return;
365+
current,
366+
targetState);
367+
}
368+
if (current == null) {
369+
current = ExecutionState.CREATED;
370+
log.error(
371+
"{} both distributed and local state are null, "
372+
+ "use {} as fallback for target state {}",
373+
taskFullName,
374+
current,
375+
targetState);
361376
}
362377
log.debug(
363378
String.format(
@@ -380,12 +395,18 @@ public synchronized void updateTaskState(@NonNull ExecutionState targetState) {
380395
}
381396

382397
// now do the actual state transition
398+
boolean missingStateEntry = stateEntryMissing;
383399
RetryUtils.retryWithException(
384400
() -> {
385401
updateStateTimestamps(targetState);
386-
if (runningJobStateIMap.get(taskGroupLocation) != null) {
387-
runningJobStateIMap.set(taskGroupLocation, targetState);
402+
if (missingStateEntry) {
403+
log.info(
404+
"{} task state entry missing from distributed map, recreate it with target state {}. Task execution location: {}",
405+
taskFullName,
406+
targetState,
407+
taskGroupLocation);
388408
}
409+
runningJobStateIMap.set(taskGroupLocation, targetState);
389410
return null;
390411
},
391412
new RetryUtils.RetryMaterial(
@@ -470,10 +491,10 @@ private void updateStateTimestamps(@NonNull ExecutionState targetState) {
470491
Long[] stateTimestamps = runningJobStateTimestampsIMap.get(taskGroupLocation);
471492
if (stateTimestamps == null) {
472493
log.warn(
473-
"{} state timestamps have already been cleaned, skip persisting transition to {}",
494+
"{} state timestamps entry missing from distributed map, recreate it for target state {}",
474495
taskFullName,
475496
targetState);
476-
return;
497+
stateTimestamps = new Long[ExecutionState.values().length];
477498
}
478499
stateTimestamps[targetState.ordinal()] = System.currentTimeMillis();
479500
runningJobStateTimestampsIMap.set(taskGroupLocation, stateTimestamps);

‎seatunnel-engine/seatunnel-engine-server/src/main/java/org/apache/seatunnel/engine/server/dag/physical/SubPlan.java‎

Lines changed: 27 additions & 6 deletions
Original file line numberDiff line numberDiff line change
@@ -345,12 +345,28 @@ public boolean canRestorePipeline() {
345345
public synchronized void updatePipelineState(@NonNull PipelineStatus targetState) {
346346
try {
347347
PipelineStatus current = (PipelineStatus) runningJobStateIMap.get(pipelineLocation);
348+
// When a node is removed during scaling down, the IMap entry may be lost.
349+
// Fall back to the local cached state to allow state progression.
350+
boolean stateEntryMissing = false;
348351
if (current == null) {
352+
stateEntryMissing = true;
353+
current = currPipelineStatus;
349354
log.warn(
350-
"{} current state is null, skip transition to {}",
355+
"{} state entry missing from distributed map (possibly due to node "
356+
+ "removal during scaling down), using local state {} as fallback, "
357+
+ "target state: {}",
351358
pipelineFullName,
359+
current,
360+
targetState);
361+
}
362+
if (current == null) {
363+
current = PipelineStatus.CREATED;
364+
log.error(
365+
"{} both distributed and local state are null, "
366+
+ "use {} as fallback for target state {}",
367+
pipelineFullName,
368+
current,
352369
targetState);
353-
return;
354370
}
355371
log.debug(
356372
String.format(
@@ -376,12 +392,17 @@ public synchronized void updatePipelineState(@NonNull PipelineStatus targetState
376392
// we must update runningJobStateTimestampsIMap first and then can update
377393
// runningJobStateIMap
378394
PipelineStatus finalTargetState = targetState;
395+
boolean missingStateEntry = stateEntryMissing;
379396
RetryUtils.retryWithException(
380397
() -> {
381398
updateStateTimestamps(finalTargetState);
382-
if (runningJobStateIMap.get(pipelineLocation) != null) {
383-
runningJobStateIMap.set(pipelineLocation, finalTargetState);
399+
if (missingStateEntry) {
400+
log.info(
401+
"{} pipeline state entry missing from distributed map, recreate it with target state {}",
402+
pipelineFullName,
403+
finalTargetState);
384404
}
405+
runningJobStateIMap.set(pipelineLocation, finalTargetState);
385406
return null;
386407
},
387408
new RetryUtils.RetryMaterial(
@@ -440,10 +461,10 @@ private void updateStateTimestamps(@NonNull PipelineStatus targetState) {
440461
Long[] stateTimestamps = runningJobStateTimestampsIMap.get(pipelineLocation);
441462
if (stateTimestamps == null) {
442463
log.warn(
443-
"{} state timestamps have already been cleaned, skip persisting transition to {}",
464+
"{} state timestamps entry missing from distributed map, recreate it for target state {}",
444465
pipelineFullName,
445466
targetState);
446-
return;
467+
stateTimestamps = new Long[PipelineStatus.values().length];
447468
}
448469
stateTimestamps[targetState.ordinal()] = System.currentTimeMillis();
449470
runningJobStateTimestampsIMap.set(pipelineLocation, stateTimestamps);

0 commit comments

Comments
 (0)