Skip to content

Commit 75084be

Browse files
davidzollodavidzollo
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 a0cefad commit 75084be

4 files changed

Lines changed: 235 additions & 35 deletions

File tree

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

Lines changed: 61 additions & 13 deletions
Original file line numberDiff line numberDiff line change
@@ -88,6 +88,8 @@ public class PhysicalPlan {
8888

8989
private volatile boolean isRunning = false;
9090

91+
private volatile JobStatus currJobStatus;
92+
9193
public PhysicalPlan(
9294
@NonNull List<SubPlan> pipelineList,
9395
@NonNull ExecutorService executorService,
@@ -128,6 +130,7 @@ public PhysicalPlan(
128130

129131
this.runningJobStateIMap = runningJobStateIMap;
130132
this.runningJobStateTimestampsIMap = runningJobStateTimestampsIMap;
133+
this.currJobStatus = (JobStatus) runningJobStateIMap.get(jobId);
131134
}
132135

133136
public void setJobMaster(JobMaster jobMaster) {
@@ -193,14 +196,18 @@ public void addPipelineEndCallback(SubPlan subPlan) {
193196

194197
public void cancelJob() {
195198
JobStatus jobStatus = getJobStatus();
199+
if (jobStatus == null) {
200+
log.error("{} job state is null, cannot cancel", jobFullName);
201+
return;
202+
}
196203
if (jobStatus.isEndState()) {
197204
log.warn(
198205
String.format(
199206
"%s is in end state %s, can not be cancel", jobFullName, jobStatus));
200207
return;
201208
}
202209

203-
if (((JobStatus) runningJobStateIMap.get(jobId)).ordinal() <= JobStatus.PENDING.ordinal()) {
210+
if (jobStatus.ordinal() <= JobStatus.PENDING.ordinal()) {
204211
// Tasks with the status 'INITIALIZING', 'CREATED', 'PENDING' need to be set directly to
205212
// the 'CANCELLED' state because it has not yet started running
206213
updateJobState(JobStatus.CANCELED);
@@ -229,6 +236,14 @@ private void updateStateTimestamps(@NonNull JobStatus targetState) {
229236
// we must update runningJobStateTimestampsIMap first and then can update
230237
// runningJobStateIMap
231238
Long[] stateTimestamps = runningJobStateTimestampsIMap.get(jobId);
239+
if (stateTimestamps == null) {
240+
log.warn(
241+
"{} state timestamps entry missing from distributed map, "
242+
+ "skip timestamp update for target state {}",
243+
jobFullName,
244+
targetState);
245+
return;
246+
}
232247
stateTimestamps[targetState.ordinal()] = System.currentTimeMillis();
233248
runningJobStateTimestampsIMap.set(jobId, stateTimestamps);
234249
}
@@ -244,6 +259,27 @@ public synchronized Long getStateTimestamp(@NonNull JobStatus jobStatus) {
244259
public synchronized void updateJobState(@NonNull JobStatus targetState) {
245260
try {
246261
JobStatus current = (JobStatus) runningJobStateIMap.get(jobId);
262+
boolean stateEntryMissing = false;
263+
if (current == null) {
264+
stateEntryMissing = true;
265+
current = currJobStatus;
266+
log.warn(
267+
"{} job state entry missing from distributed map (possibly due to node "
268+
+ "removal during scaling down), using local state {} as fallback, "
269+
+ "target state: {}",
270+
jobFullName,
271+
current,
272+
targetState);
273+
}
274+
if (current == null) {
275+
log.error(
276+
"{} both distributed and local job state are null, "
277+
+ "cannot transition to {}",
278+
jobFullName,
279+
targetState);
280+
return;
281+
}
282+
247283
log.debug(
248284
"Try to update the {} state from {} to {}", jobFullName, current, targetState);
249285

@@ -261,17 +297,20 @@ public synchronized void updateJobState(@NonNull JobStatus targetState) {
261297

262298
// Now do the actual state transition, we must update runningJobStateTimestampsIMap
263299
// first and then can update runningJobStateIMap
264-
RetryUtils.retryWithException(
265-
() -> {
266-
updateStateTimestamps(targetState);
267-
runningJobStateIMap.set(jobId, targetState);
268-
return null;
269-
},
270-
new RetryUtils.RetryMaterial(
271-
Constant.OPERATION_RETRY_TIME,
272-
true,
273-
ExceptionUtil::isOperationNeedRetryException,
274-
Constant.OPERATION_RETRY_SLEEP));
300+
if (!stateEntryMissing) {
301+
RetryUtils.retryWithException(
302+
() -> {
303+
updateStateTimestamps(targetState);
304+
runningJobStateIMap.set(jobId, targetState);
305+
return null;
306+
},
307+
new RetryUtils.RetryMaterial(
308+
Constant.OPERATION_RETRY_TIME,
309+
true,
310+
ExceptionUtil::isOperationNeedRetryException,
311+
Constant.OPERATION_RETRY_SLEEP));
312+
}
313+
this.currJobStatus = targetState;
275314
log.info(
276315
String.format(
277316
"%s turned from state %s to %s.", jobFullName, current, targetState));
@@ -289,7 +328,16 @@ public JobImmutableInformation getJobImmutableInformation() {
289328
}
290329

291330
public JobStatus getJobStatus() {
292-
return (JobStatus) runningJobStateIMap.get(jobId);
331+
JobStatus status = (JobStatus) runningJobStateIMap.get(jobId);
332+
if (status == null) {
333+
log.warn(
334+
"{} job state entry missing from distributed map, "
335+
+ "using local cached state {} as fallback",
336+
jobFullName,
337+
currJobStatus);
338+
return currJobStatus;
339+
}
340+
return status;
293341
}
294342

295343
public String getJobFullName() {

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

Lines changed: 41 additions & 11 deletions
Original file line numberDiff line numberDiff line change
@@ -351,6 +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;
357+
if (current == null) {
358+
stateEntryMissing = true;
359+
current = currExecutionState;
360+
log.warn(
361+
"{} state entry missing from distributed map (possibly due to node "
362+
+ "removal during scaling down), using local state {} as fallback, "
363+
+ "target state: {}",
364+
taskFullName,
365+
current,
366+
targetState);
367+
}
368+
if (current == null) {
369+
log.error(
370+
"{} both distributed and local state are null, "
371+
+ "cannot transition to {}",
372+
taskFullName,
373+
targetState);
374+
return;
375+
}
354376
log.debug(
355377
String.format(
356378
"Try to update the task %s state from %s to %s",
@@ -372,17 +394,19 @@ public synchronized void updateTaskState(@NonNull ExecutionState targetState) {
372394
}
373395

374396
// now do the actual state transition
375-
RetryUtils.retryWithException(
376-
() -> {
377-
updateStateTimestamps(targetState);
378-
runningJobStateIMap.set(taskGroupLocation, targetState);
379-
return null;
380-
},
381-
new RetryUtils.RetryMaterial(
382-
Constant.OPERATION_RETRY_TIME,
383-
true,
384-
ExceptionUtil::isOperationNeedRetryException,
385-
Constant.OPERATION_RETRY_SLEEP));
397+
if (!stateEntryMissing) {
398+
RetryUtils.retryWithException(
399+
() -> {
400+
updateStateTimestamps(targetState);
401+
runningJobStateIMap.set(taskGroupLocation, targetState);
402+
return null;
403+
},
404+
new RetryUtils.RetryMaterial(
405+
Constant.OPERATION_RETRY_TIME,
406+
true,
407+
ExceptionUtil::isOperationNeedRetryException,
408+
Constant.OPERATION_RETRY_SLEEP));
409+
}
386410
this.currExecutionState = targetState;
387411
log.info(
388412
String.format(
@@ -451,6 +475,12 @@ private void updateStateTimestamps(@NonNull ExecutionState targetState) {
451475
// we must update runningJobStateTimestampsIMap first and then can update
452476
// runningJobStateIMap
453477
Long[] stateTimestamps = runningJobStateTimestampsIMap.get(taskGroupLocation);
478+
if (stateTimestamps == null) {
479+
log.warn(
480+
"{} state timestamps entry missing from distributed map, skip timestamp update",
481+
taskFullName);
482+
return;
483+
}
454484
stateTimestamps[targetState.ordinal()] = System.currentTimeMillis();
455485
runningJobStateTimestampsIMap.set(taskGroupLocation, stateTimestamps);
456486
}

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

Lines changed: 41 additions & 11 deletions
Original file line numberDiff line numberDiff line change
@@ -340,6 +340,28 @@ public boolean canRestorePipeline() {
340340
public synchronized void updatePipelineState(@NonNull PipelineStatus targetState) {
341341
try {
342342
PipelineStatus current = (PipelineStatus) runningJobStateIMap.get(pipelineLocation);
343+
// When a node is removed during scaling down, the IMap entry may be lost.
344+
// Fall back to the local cached state to allow state progression.
345+
boolean stateEntryMissing = false;
346+
if (current == null) {
347+
stateEntryMissing = true;
348+
current = currPipelineStatus;
349+
log.warn(
350+
"{} state entry missing from distributed map (possibly due to node "
351+
+ "removal during scaling down), using local state {} as fallback, "
352+
+ "target state: {}",
353+
pipelineFullName,
354+
current,
355+
targetState);
356+
}
357+
if (current == null) {
358+
log.error(
359+
"{} both distributed and local state are null, "
360+
+ "cannot transition to {}",
361+
pipelineFullName,
362+
targetState);
363+
return;
364+
}
343365
log.debug(
344366
String.format(
345367
"Try to update the %s state from %s to %s",
@@ -364,17 +386,19 @@ public synchronized void updatePipelineState(@NonNull PipelineStatus targetState
364386
// we must update runningJobStateTimestampsIMap first and then can update
365387
// runningJobStateIMap
366388
PipelineStatus finalTargetState = targetState;
367-
RetryUtils.retryWithException(
368-
() -> {
369-
updateStateTimestamps(finalTargetState);
370-
runningJobStateIMap.set(pipelineLocation, finalTargetState);
371-
return null;
372-
},
373-
new RetryUtils.RetryMaterial(
374-
Constant.OPERATION_RETRY_TIME,
375-
true,
376-
exception -> ExceptionUtil.isOperationNeedRetryException(exception),
377-
Constant.OPERATION_RETRY_SLEEP));
389+
if (!stateEntryMissing) {
390+
RetryUtils.retryWithException(
391+
() -> {
392+
updateStateTimestamps(finalTargetState);
393+
runningJobStateIMap.set(pipelineLocation, finalTargetState);
394+
return null;
395+
},
396+
new RetryUtils.RetryMaterial(
397+
Constant.OPERATION_RETRY_TIME,
398+
true,
399+
exception -> ExceptionUtil.isOperationNeedRetryException(exception),
400+
Constant.OPERATION_RETRY_SLEEP));
401+
}
378402
this.currPipelineStatus = targetState;
379403
log.info(
380404
String.format(
@@ -418,6 +442,12 @@ private void updateStateTimestamps(@NonNull PipelineStatus targetState) {
418442
// we must update runningJobStateTimestampsIMap first and then can update
419443
// runningJobStateIMap
420444
Long[] stateTimestamps = runningJobStateTimestampsIMap.get(pipelineLocation);
445+
if (stateTimestamps == null) {
446+
log.warn(
447+
"{} state timestamps entry missing from distributed map, skip timestamp update",
448+
pipelineFullName);
449+
return;
450+
}
421451
stateTimestamps[targetState.ordinal()] = System.currentTimeMillis();
422452
runningJobStateTimestampsIMap.set(pipelineLocation, stateTimestamps);
423453
}

‎seatunnel-engine/seatunnel-engine-server/src/test/java/org/apache/seatunnel/engine/server/dag/TaskTest.java‎

Lines changed: 92 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -53,7 +53,9 @@
5353
import org.apache.seatunnel.engine.server.dag.physical.PhysicalVertex;
5454
import org.apache.seatunnel.engine.server.dag.physical.PlanUtils;
5555
import org.apache.seatunnel.engine.server.dag.physical.SubPlan;
56+
import org.apache.seatunnel.engine.server.execution.ExecutionState;
5657
import org.apache.seatunnel.engine.server.execution.Task;
58+
import org.apache.seatunnel.engine.server.execution.TaskExecutionState;
5759
import org.apache.seatunnel.engine.server.execution.TaskGroupLocation;
5860

5961
import org.junit.jupiter.api.Assertions;
@@ -307,6 +309,96 @@ public void testTaskGroupAndTaskLocationInfos() {
307309
}
308310
}
309311

312+
// Regression test: when the state entry is removed from the distributed map
313+
// (e.g. during node scaling down), updateTaskState should still complete
314+
// the local state transition instead of throwing NPE.
315+
@Test
316+
@SetEnvironmentVariable(key = SKIP_CHECK_JAR, value = "true")
317+
public void testUpdateTaskStateWhenStateEntryMissing() throws MalformedURLException {
318+
IdGenerator idGenerator = new IdGenerator();
319+
320+
Action fake =
321+
new SourceAction<>(
322+
idGenerator.getNextId(),
323+
"fake",
324+
createFakeSource(),
325+
Sets.newHashSet(new URL("file:///fake.jar")),
326+
Collections.emptySet());
327+
LogicalVertex fakeVertex = new LogicalVertex(fake.getId(), fake, 2);
328+
329+
List<Column> columns = new ArrayList<>();
330+
columns.add(PhysicalColumn.of("id", BasicType.INT_TYPE, 11L, 0, true, 111, ""));
331+
332+
CatalogTable catalogTable =
333+
CatalogTable.of(
334+
TableIdentifier.of("default", TablePath.DEFAULT),
335+
TableSchema.builder().columns(columns).build(),
336+
new HashMap<>(),
337+
Collections.emptyList(),
338+
"fake");
339+
340+
Action console =
341+
new SinkAction<>(
342+
idGenerator.getNextId(),
343+
"console",
344+
new ConsoleSink(catalogTable, ReadonlyConfig.fromMap(new HashMap<>())),
345+
Sets.newHashSet(new URL("file:///console.jar")),
346+
Collections.emptySet());
347+
LogicalVertex consoleVertex = new LogicalVertex(console.getId(), console, 2);
348+
349+
LogicalEdge edge = new LogicalEdge(fakeVertex, consoleVertex);
350+
351+
JobConfig config = new JobConfig();
352+
config.setName("test");
353+
LogicalDag logicalDag = new LogicalDag(config, idGenerator);
354+
logicalDag.addLogicalVertex(fakeVertex);
355+
logicalDag.addLogicalVertex(consoleVertex);
356+
logicalDag.addEdge(edge);
357+
358+
JobImmutableInformation jobImmutableInformation =
359+
new JobImmutableInformation(
360+
2,
361+
"Test",
362+
nodeEngine.getSerializationService(),
363+
logicalDag,
364+
Collections.emptyList(),
365+
Collections.emptyList());
366+
367+
IMap<Object, Object> runningJobState =
368+
nodeEngine.getHazelcastInstance().getMap("testRunningJobStateNullStateEntry");
369+
IMap<Object, Long[]> runningJobStateTimestamp =
370+
nodeEngine
371+
.getHazelcastInstance()
372+
.getMap("testRunningJobStateTimestampNullStateEntry");
373+
374+
PhysicalPlan physicalPlan =
375+
PlanUtils.fromLogicalDAG(
376+
logicalDag,
377+
nodeEngine,
378+
jobImmutableInformation,
379+
System.currentTimeMillis(),
380+
Executors.newCachedThreadPool(),
381+
server.getClassLoaderService(),
382+
instance.getFlakeIdGenerator(Constant.SEATUNNEL_ID_GENERATOR_NAME),
383+
runningJobState,
384+
runningJobStateTimestamp,
385+
QueueType.BLOCKINGQUEUE,
386+
new EngineConfig())
387+
.f0();
388+
389+
PhysicalVertex physicalVertex =
390+
physicalPlan.getPipelineList().get(0).getPhysicalVertexList().get(0);
391+
PassiveCompletableFuture<TaskExecutionState> stateFuture = physicalVertex.initStateFuture();
392+
physicalVertex.startPhysicalVertex();
393+
394+
runningJobState.remove(physicalVertex.getTaskGroupLocation());
395+
physicalVertex.makeTaskGroupFailing(new RuntimeException("test missing state entry"));
396+
397+
Assertions.assertTrue(stateFuture.isDone());
398+
Assertions.assertEquals(ExecutionState.FAILED, physicalVertex.getExecutionState());
399+
Assertions.assertEquals(ExecutionState.FAILED, stateFuture.join().getExecutionState());
400+
}
401+
310402
private static FakeSource createFakeSource() {
311403
Config fakeSourceConfig =
312404
ConfigFactory.parseMap(

0 commit comments

Comments
 (0)