[Bug][Core] Local mode cancel running JobStatusRunner (#7943)

apache · Oct 30, 2024 · 74791dc · 74791dc
1 parent 8baa012
commit 74791dc
Showing 1 changed file with 11 additions and 6 deletions.
diff --git a/...c/main/java/org/apache/seatunnel/core/starter/seatunnel/command/ClientExecuteCommand.java b/...c/main/java/org/apache/seatunnel/core/starter/seatunnel/command/ClientExecuteCommand.java
@@ -85,7 +85,9 @@ public void execute() throws CommandExecuteException {
         try {
             String clusterName = clientCommandArgs.getClusterName();
             ClientConfig clientConfig = ConfigProvider.locateAndGetClientConfig();
-            if (clientCommandArgs.getMasterType().equals(MasterType.LOCAL)) {
+            //  get running mode
+            boolean isLocalMode = clientCommandArgs.getMasterType().equals(MasterType.LOCAL);
+            if (isLocalMode) {
                 clusterName =
                         creatRandomClusterName(
                                 StringUtils.isNotEmpty(clusterName)
@@ -160,7 +162,7 @@ public void execute() throws CommandExecuteException {
                 // create job proxy
                 ClientJobProxy clientJobProxy = jobExecutionEnv.execute();
                 if (clientCommandArgs.isAsync()) {
-                    if (clientCommandArgs.getMasterType().equals(MasterType.LOCAL)) {
+                    if (isLocalMode) {
                         log.warn("The job is running in local mode, can not use async mode.");
                     } else {
                         return;
@@ -200,10 +202,13 @@ public void execute() throws CommandExecuteException {
                         seaTunnelConfig.getEngineConfig().getPrintJobMetricsInfoInterval(),
                         TimeUnit.SECONDS);
 
-                executorService.schedule(
-                        new JobStatusRunner(engineClient.getJobClient(), jobId),
-                        0,
-                        TimeUnit.SECONDS);
+                if (!isLocalMode) {
+                    // LOCAL mode does not require running the job status runner
+                    executorService.schedule(
+                            new JobStatusRunner(engineClient.getJobClient(), jobId),
+                            0,
+                            TimeUnit.SECONDS);
+                }
 
                 // wait for job complete
                 JobResult jobResult = clientJobProxy.waitForJobCompleteV2();