[Hotfix][Zeta] Fix master active bug

seatunnel-engine/seatunnel-engine-server/src/main/java/org/apache/seatunnel/engine/server/CoordinatorService.java

@@ -128,7 +128,7 @@ public class CoordinatorService {
     /** If this node is a master node */
     private volatile boolean isActive = false;
 
-    private final ExecutorService executorService;
+    private ExecutorService executorService;
 
     private final SeaTunnelServer seaTunnelServer;
 
@@ -278,13 +278,18 @@ private void restoreJobFromMasterActiveSwitch(@NonNull Long jobId, @NonNull JobI
         }
 
         if (jobStatus.ordinal() < JobStatus.RUNNING.ordinal()) {
-            logger.info(
-                    String.format(
-                            "The restore %s is state %s, cancel job and submit it again.",
-                            jobFullName, jobStatus));
-            jobMaster.cancelJob();
-            jobMaster.getJobMasterCompleteFuture().join();
-            submitJob(jobId, jobInfo.getJobImmutableInformation()).join();
+            CompletableFuture.runAsync(
+                    () -> {
+                        logger.info(
+                                String.format(
+                                        "The restore %s is state %s, cancel job and submit it again.",
+                                        jobFullName, jobStatus));
+                        jobMaster.cancelJob();
+                        jobMaster.getJobMasterCompleteFuture().join();
+                        submitJob(jobId, jobInfo.getJobImmutableInformation()).join();
+                    },
+                    executorService);
+
             return;
         }
 
@@ -345,6 +350,13 @@ private void checkNewActiveMaster() {
             if (!isActive && this.seaTunnelServer.isMasterNode()) {
                 logger.info(
                         "This node become a new active master node, begin init coordinator service");
+                if (this.executorService.isShutdown()) {
+                    this.executorService =
+                            Executors.newCachedThreadPool(
+                                    new ThreadFactoryBuilder()
+                                            .setNameFormat("seatunnel-coordinator-service-%d")
+                                            .build());
+                }
                 initCoordinatorService();
                 isActive = true;
             } else if (isActive && !this.seaTunnelServer.isMasterNode()) {
@@ -522,6 +534,11 @@ public JobDAGInfo getJobInfo(long jobId) {
      * TaskGroup's state.
      */
     public void updateTaskExecutionState(TaskExecutionState taskExecutionState) {
+        logger.info(
+                String.format(
+                        "Received task end from execution %s, state %s",
+                        taskExecutionState.getTaskGroupLocation(),
+                        taskExecutionState.getExecutionState()));
         TaskGroupLocation taskGroupLocation = taskExecutionState.getTaskGroupLocation();
         JobMaster runningJobMaster = runningJobMasterMap.get(taskGroupLocation.getJobId());
         if (runningJobMaster == null) {

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

@@ -211,7 +211,9 @@ private boolean checkTaskGroupIsExecuting(TaskGroupLocation taskGroupLocation) {
                 LOGGER.warning(
                         "The node:"
                                 + worker.toString()
-                                + " running the taskGroup no longer exists, return false.");
+                                + " running the taskGroup "
+                                + taskGroupLocation
+                                + " no longer exists, return false.");
                 return false;
             }
             InvocationFuture<Object> invoke =
@@ -226,7 +228,9 @@ private boolean checkTaskGroupIsExecuting(TaskGroupLocation taskGroupLocation) {
                 return (Boolean) invoke.get();
             } catch (InterruptedException | ExecutionException e) {
                 LOGGER.warning(
-                        "Execution of CheckTaskGroupIsExecutingOperation failed, checkTaskGroupIsExecuting return false. ",
+                        "Execution of CheckTaskGroupIsExecutingOperation "
+                                + taskGroupLocation
+                                + " failed, checkTaskGroupIsExecuting return false. ",
                         e);
             }
         }