This is an automated email from the ASF dual-hosted git repository.
davidzollo pushed a commit to branch dev
in repository https://gitbox.apache.org/repos/asf/seatunnel.git
The following commit(s) were added to refs/heads/dev by this push:
new e360c00cc4 [Fix][Zeta] Fix JobNotFoundException in
updateTaskExecutionState (#10506)
e360c00cc4 is described below
commit e360c00cc4fb8f0bb1a2c9c04c40ccc8d928faf6
Author: Doyeon Kim <[email protected]>
AuthorDate: Wed Jul 8 00:41:22 2026 +0900
[Fix][Zeta] Fix JobNotFoundException in updateTaskExecutionState (#10506)
---
.../exception/JobRestoreInProgressException.java | 33 ++++
.../engine/common/utils/ExceptionUtil.java | 7 +-
.../engine/server/CoordinatorService.java | 14 ++
.../engine/server/TaskExecutionService.java | 14 ++
.../engine/server/CoordinatorServiceTest.java | 198 +++++++++++++++++++++
5 files changed, 265 insertions(+), 1 deletion(-)
diff --git
a/seatunnel-engine/seatunnel-engine-common/src/main/java/org/apache/seatunnel/engine/common/exception/JobRestoreInProgressException.java
b/seatunnel-engine/seatunnel-engine-common/src/main/java/org/apache/seatunnel/engine/common/exception/JobRestoreInProgressException.java
new file mode 100644
index 0000000000..6da05a75ab
--- /dev/null
+++
b/seatunnel-engine/seatunnel-engine-common/src/main/java/org/apache/seatunnel/engine/common/exception/JobRestoreInProgressException.java
@@ -0,0 +1,33 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.seatunnel.engine.common.exception;
+
+public class JobRestoreInProgressException extends SeaTunnelEngineException {
+ public JobRestoreInProgressException(String message) {
+ super(message);
+ }
+
+ public JobRestoreInProgressException(String message, Throwable cause) {
+ super(message, cause);
+ }
+
+ @Override
+ public Throwable createException(String s, Throwable throwable) {
+ return new JobRestoreInProgressException(s, throwable);
+ }
+}
diff --git
a/seatunnel-engine/seatunnel-engine-common/src/main/java/org/apache/seatunnel/engine/common/utils/ExceptionUtil.java
b/seatunnel-engine/seatunnel-engine-common/src/main/java/org/apache/seatunnel/engine/common/utils/ExceptionUtil.java
index c812ba54f4..97d28441bb 100644
---
a/seatunnel-engine/seatunnel-engine-common/src/main/java/org/apache/seatunnel/engine/common/utils/ExceptionUtil.java
+++
b/seatunnel-engine/seatunnel-engine-common/src/main/java/org/apache/seatunnel/engine/common/utils/ExceptionUtil.java
@@ -24,6 +24,7 @@ import
org.apache.seatunnel.common.utils.function.RunnableWithException;
import org.apache.seatunnel.common.utils.function.SupplierWithException;
import org.apache.seatunnel.engine.common.exception.JobDefineCheckException;
import org.apache.seatunnel.engine.common.exception.JobNotFoundException;
+import
org.apache.seatunnel.engine.common.exception.JobRestoreInProgressException;
import org.apache.seatunnel.engine.common.exception.SeaTunnelEngineException;
import com.hazelcast.client.impl.protocol.ClientExceptionFactory;
@@ -60,7 +61,11 @@ public final class ExceptionUtil {
new ImmutableTriple<>(
ClientProtocolErrorCodes.USER_EXCEPTIONS_RANGE_START + 2,
JobDefineCheckException.class,
- JobDefineCheckException::new));
+ JobDefineCheckException::new),
+ new ImmutableTriple<>(
+
ClientProtocolErrorCodes.USER_EXCEPTIONS_RANGE_START + 3,
+ JobRestoreInProgressException.class,
+ JobRestoreInProgressException::new));
private ExceptionUtil() {}
diff --git
a/seatunnel-engine/seatunnel-engine-server/src/main/java/org/apache/seatunnel/engine/server/CoordinatorService.java
b/seatunnel-engine/seatunnel-engine-server/src/main/java/org/apache/seatunnel/engine/server/CoordinatorService.java
index 4cd0c92475..c58a29ad20 100644
---
a/seatunnel-engine/seatunnel-engine-server/src/main/java/org/apache/seatunnel/engine/server/CoordinatorService.java
+++
b/seatunnel-engine/seatunnel-engine-server/src/main/java/org/apache/seatunnel/engine/server/CoordinatorService.java
@@ -37,6 +37,7 @@ import
org.apache.seatunnel.engine.common.config.server.ConnectorJarStorageConfi
import org.apache.seatunnel.engine.common.config.server.ScheduleStrategy;
import org.apache.seatunnel.engine.common.exception.JobException;
import org.apache.seatunnel.engine.common.exception.JobNotFoundException;
+import
org.apache.seatunnel.engine.common.exception.JobRestoreInProgressException;
import org.apache.seatunnel.engine.common.exception.SavePointFailedException;
import org.apache.seatunnel.engine.common.exception.SeaTunnelEngineException;
import org.apache.seatunnel.engine.common.job.JobResult;
@@ -1795,6 +1796,19 @@ public class CoordinatorService {
taskExecutionState.getExecutionState()));
TaskGroupLocation taskGroupLocation =
taskExecutionState.getTaskGroupLocation();
JobMaster runningJobMaster =
runningJobMasterMap.get(taskGroupLocation.getJobId());
+
+ if (runningJobMaster == null &&
!restoreAllJobFromMasterNodeSwitchFuture.isDone()) {
+ // Restore still in progress, return early and let worker retry
+ // This is acceptable because worker already has retry logic
+ logger.info(
+ String.format(
+ "Job %s not found and restore still in progress,
worker will retry",
+ taskGroupLocation.getJobId()));
+ throw new JobRestoreInProgressException(
+ String.format(
+ "Job %s not running (restore in progress)",
+ taskGroupLocation.getJobId()));
+ }
if (runningJobMaster == null) {
throw new JobNotFoundException(
String.format("Job %s not running",
taskGroupLocation.getJobId()));
diff --git
a/seatunnel-engine/seatunnel-engine-server/src/main/java/org/apache/seatunnel/engine/server/TaskExecutionService.java
b/seatunnel-engine/seatunnel-engine-server/src/main/java/org/apache/seatunnel/engine/server/TaskExecutionService.java
index b3bc49dd6d..ca45dc10d9 100644
---
a/seatunnel-engine/seatunnel-engine-server/src/main/java/org/apache/seatunnel/engine/server/TaskExecutionService.java
+++
b/seatunnel-engine/seatunnel-engine-server/src/main/java/org/apache/seatunnel/engine/server/TaskExecutionService.java
@@ -29,6 +29,7 @@ import
org.apache.seatunnel.engine.common.config.ConfigProvider;
import org.apache.seatunnel.engine.common.config.SeaTunnelConfig;
import org.apache.seatunnel.engine.common.config.server.ThreadShareMode;
import org.apache.seatunnel.engine.common.exception.JobNotFoundException;
+import
org.apache.seatunnel.engine.common.exception.JobRestoreInProgressException;
import org.apache.seatunnel.engine.common.utils.PassiveCompletableFuture;
import org.apache.seatunnel.engine.common.utils.concurrent.CompletableFuture;
import org.apache.seatunnel.engine.core.classloader.ClassLoaderService;
@@ -639,6 +640,18 @@ public class TaskExecutionService implements
DynamicMetricsProvider {
} catch (JobNotFoundException e) {
logger.warning("send notify task status failed because can't
find job", e);
notifyStateSuccess = true;
+ } catch (JobRestoreInProgressException e) {
+ logger.info(ExceptionUtils.getMessage(e));
+ logger.info(
+ String.format(
+ "notify the job of the task(%s) status failed,
retry in %s millis",
+ taskGroupLocation, sleepTime));
+ try {
+ Thread.sleep(sleepTime);
+ } catch (InterruptedException ex) {
+ Thread.currentThread().interrupt();
+ logger.severe(e);
+ }
} catch (ExecutionException e) {
if (e.getCause() instanceof JobNotFoundException) {
logger.warning("send notify task status failed because
can't find job", e);
@@ -652,6 +665,7 @@ public class TaskExecutionService implements
DynamicMetricsProvider {
try {
Thread.sleep(sleepTime);
} catch (InterruptedException ex) {
+ Thread.currentThread().interrupt();
logger.severe(e);
}
}
diff --git
a/seatunnel-engine/seatunnel-engine-server/src/test/java/org/apache/seatunnel/engine/server/CoordinatorServiceTest.java
b/seatunnel-engine/seatunnel-engine-server/src/test/java/org/apache/seatunnel/engine/server/CoordinatorServiceTest.java
index 59d86896a7..8c934a9b5b 100644
---
a/seatunnel-engine/seatunnel-engine-server/src/test/java/org/apache/seatunnel/engine/server/CoordinatorServiceTest.java
+++
b/seatunnel-engine/seatunnel-engine-server/src/test/java/org/apache/seatunnel/engine/server/CoordinatorServiceTest.java
@@ -25,6 +25,7 @@ import
org.apache.seatunnel.engine.common.config.ConfigProvider;
import org.apache.seatunnel.engine.common.config.EngineConfig;
import org.apache.seatunnel.engine.common.config.SeaTunnelConfig;
import org.apache.seatunnel.engine.common.config.server.ScheduleStrategy;
+import org.apache.seatunnel.engine.common.exception.JobNotFoundException;
import org.apache.seatunnel.engine.common.exception.SeaTunnelEngineException;
import org.apache.seatunnel.engine.common.job.JobResult;
import org.apache.seatunnel.engine.common.job.JobStatus;
@@ -43,6 +44,7 @@ import
org.apache.seatunnel.engine.server.dag.physical.SubPlan;
import org.apache.seatunnel.engine.server.execution.ExecutionState;
import org.apache.seatunnel.engine.server.execution.PendingJobInfo;
import org.apache.seatunnel.engine.server.execution.PendingSourceState;
+import org.apache.seatunnel.engine.server.execution.TaskExecutionState;
import org.apache.seatunnel.engine.server.execution.TaskGroupContext;
import org.apache.seatunnel.engine.server.execution.TaskGroupLocation;
import org.apache.seatunnel.engine.server.execution.TaskLocation;
@@ -1260,6 +1262,202 @@ public class CoordinatorServiceTest {
.count());
}
+ @Test
+ void testUpdateTaskExecutionStateRetriesUntilRestoreComplete() throws
Exception {
+ CountDownLatch restoreLatch = new CountDownLatch(1);
+
+ JobInformation jobInformation =
+ submitJob(
+
"CoordinatorServiceTest_testUpdateTaskExecutionStateRetriesUntilRestoreComplete",
+ "batch_fake_to_console.conf",
+
"test_update_task_execution_state_retries_until_restore_complete");
+ CoordinatorService coordinatorService =
jobInformation.coordinatorService;
+ await().atMost(60000, TimeUnit.MILLISECONDS)
+ .untilAsserted(
+ () ->
+ Assertions.assertEquals(
+ JobStatus.RUNNING,
+
coordinatorService.getJobStatus(jobInformation.jobId)));
+ JobMaster jobMaster =
coordinatorService.getJobMaster(jobInformation.jobId);
+
+ Field mapField =
CoordinatorService.class.getDeclaredField("runningJobMasterMap");
+ mapField.setAccessible(true);
+ @SuppressWarnings("unchecked")
+ Map<Long, JobMaster> runningJobMasterMap =
+ (Map<Long, JobMaster>) mapField.get(coordinatorService);
+ runningJobMasterMap.remove(jobInformation.jobId);
+
+ CompletableFuture<Void> delayedFuture =
+ CompletableFuture.runAsync(
+ () -> {
+ try {
+ restoreLatch.await(10, TimeUnit.SECONDS);
+ } catch (InterruptedException e) {
+ Thread.currentThread().interrupt();
+ }
+ });
+
+ PassiveCompletableFuture<Void> delayedPassiveFuture =
+ new PassiveCompletableFuture<>(new
CompletableFuture<>(delayedFuture));
+
+ Field futureField =
+ CoordinatorService.class.getDeclaredField(
+ "restoreAllJobFromMasterNodeSwitchFuture");
+ futureField.setAccessible(true);
+ futureField.set(coordinatorService, delayedPassiveFuture);
+
+ ExecutorService restoreExecutor = Executors.newSingleThreadExecutor();
+ try {
+ restoreExecutor.submit(
+ () -> {
+ try {
+ Thread.sleep(1000);
+ runningJobMasterMap.put(jobInformation.jobId,
jobMaster);
+ restoreLatch.countDown();
+ } catch (InterruptedException e) {
+ Thread.currentThread().interrupt();
+ }
+ });
+
+ TaskGroupLocation taskGroupLocation =
+ jobMaster
+ .getPhysicalPlan()
+ .getPipelineList()
+ .get(0)
+ .getPhysicalVertexList()
+ .get(0)
+ .getTaskGroupLocation();
+ TaskExecutionState taskExecutionState =
+ new TaskExecutionState(taskGroupLocation,
ExecutionState.FAILED);
+
+ await().atMost(10000, TimeUnit.MILLISECONDS)
+ .untilAsserted(
+ () ->
+ Assertions.assertDoesNotThrow(
+ () ->
+
coordinatorService.updateTaskExecutionState(
+
taskExecutionState)));
+ } finally {
+ restoreExecutor.shutdownNow();
+ jobInformation.coordinatorServiceTest.shutdown();
+ }
+ }
+
+ @Test
+ void testUpdateTaskExecutionStateThrowsWhenJobNotFound() {
+ JobInformation jobInformation =
+ submitJob(
+
"CoordinatorServiceTest_testUpdateTaskExecutionStateThrowsWhenJobNotFound",
+ "batch_fake_to_console.conf",
+
"test_update_task_execution_state_throws_when_job_not_found");
+ CoordinatorService coordinatorService =
jobInformation.coordinatorService;
+
+ try {
+ TaskGroupLocation fakeLocation = new TaskGroupLocation(99999L, 0,
0);
+ TaskExecutionState taskExecutionState =
+ new TaskExecutionState(fakeLocation,
ExecutionState.FAILED);
+
+ Assertions.assertThrows(
+ JobNotFoundException.class,
+ () ->
coordinatorService.updateTaskExecutionState(taskExecutionState));
+ } finally {
+ jobInformation.coordinatorServiceTest.shutdown();
+ }
+ }
+
+ @Test
+ void testUpdateTaskExecutionStateWhenRestoreFailed() throws Exception {
+ JobInformation jobInformation =
+ submitJob(
+
"CoordinatorServiceTest_testUpdateTaskExecutionStateWhenRestoreFailed",
+ "batch_fake_to_console.conf",
+
"test_update_task_execution_state_when_restore_failed");
+ CoordinatorService coordinatorService =
jobInformation.coordinatorService;
+ await().atMost(60000, TimeUnit.MILLISECONDS)
+ .untilAsserted(
+ () ->
+ Assertions.assertEquals(
+ JobStatus.RUNNING,
+
coordinatorService.getJobStatus(jobInformation.jobId)));
+
+ try {
+ Field mapField =
CoordinatorService.class.getDeclaredField("runningJobMasterMap");
+ mapField.setAccessible(true);
+ @SuppressWarnings("unchecked")
+ Map<Long, JobMaster> runningJobMasterMap =
+ (Map<Long, JobMaster>) mapField.get(coordinatorService);
+ runningJobMasterMap.remove(jobInformation.jobId);
+
+ CompletableFuture<Void> failedFuture = new CompletableFuture<>();
+ failedFuture.completeExceptionally(new
SeaTunnelEngineException("Restore failed"));
+ PassiveCompletableFuture<Void> failedPassiveFuture =
+ new PassiveCompletableFuture<>(failedFuture);
+
+ Field futureField =
+ CoordinatorService.class.getDeclaredField(
+ "restoreAllJobFromMasterNodeSwitchFuture");
+ futureField.setAccessible(true);
+ futureField.set(coordinatorService, failedPassiveFuture);
+
+ TaskGroupLocation fakeLocation = new
TaskGroupLocation(jobInformation.jobId, 0, 0);
+ TaskExecutionState taskExecutionState =
+ new TaskExecutionState(fakeLocation,
ExecutionState.FAILED);
+
+ Assertions.assertThrows(
+ JobNotFoundException.class,
+ () ->
coordinatorService.updateTaskExecutionState(taskExecutionState));
+ } finally {
+ jobInformation.coordinatorServiceTest.shutdown();
+ }
+ }
+
+ @Test
+ void testUpdateTaskExecutionStateWhenJobRemovedDuringRestore() throws
Exception {
+ JobInformation jobInformation =
+ submitJob(
+
"CoordinatorServiceTest_testUpdateTaskExecutionStateWhenJobRemovedDuringRestore",
+ "batch_fake_to_console.conf",
+
"test_update_task_execution_state_job_removed_during_restore");
+ CoordinatorService coordinatorService =
jobInformation.coordinatorService;
+ await().atMost(60000, TimeUnit.MILLISECONDS)
+ .untilAsserted(
+ () ->
+ Assertions.assertEquals(
+ JobStatus.RUNNING,
+
coordinatorService.getJobStatus(jobInformation.jobId)));
+
+ try {
+ Field mapField =
CoordinatorService.class.getDeclaredField("runningJobMasterMap");
+ mapField.setAccessible(true);
+ @SuppressWarnings("unchecked")
+ Map<Long, JobMaster> runningJobMasterMap =
+ (Map<Long, JobMaster>) mapField.get(coordinatorService);
+
+ Long jobId = jobInformation.jobId;
+ runningJobMasterMap.remove(jobId);
+
+ CompletableFuture<Void> completedFuture =
CompletableFuture.completedFuture(null);
+ PassiveCompletableFuture<Void> completedPassiveFuture =
+ new PassiveCompletableFuture<>(completedFuture);
+
+ Field futureField =
+ CoordinatorService.class.getDeclaredField(
+ "restoreAllJobFromMasterNodeSwitchFuture");
+ futureField.setAccessible(true);
+ futureField.set(coordinatorService, completedPassiveFuture);
+
+ TaskGroupLocation fakeLocation = new TaskGroupLocation(jobId, 0,
0);
+ TaskExecutionState taskExecutionState =
+ new TaskExecutionState(fakeLocation,
ExecutionState.FAILED);
+
+ Assertions.assertThrows(
+ JobNotFoundException.class,
+ () ->
coordinatorService.updateTaskExecutionState(taskExecutionState));
+ } finally {
+ jobInformation.coordinatorServiceTest.shutdown();
+ }
+ }
+
@Test
void testRestoreUsesProvidedJobInfoInitializationTimestamp() throws
Exception {
HazelcastInstanceImpl instance =
