apache · richardmilles · Aug 7, 2026
@@ -32,4 +32,11 @@ private Constants() {
     public static final String JSON_SUFFIX = "json";
     public static final String CONF_SUFFIX = "conf";
 
+    /**
+     * SeaTunnel Engine client logs job id as: "Start submit job, job id: 123, ..."
+     * or "Submit job finished, job id: 123, job name: ..."
+     */
+    public static final String SEATUNNEL_JOB_ID_REGEX = "(?i)job id:\\s*(\\d+)";
+    public static final String CANCEL_JOB_OPTIONS = "-can";
+
 }
@@ -48,15 +48,21 @@
 import java.util.ArrayList;
 import java.util.Collections;
 import java.util.HashMap;
+import java.util.HashSet;
 import java.util.List;
 import java.util.Map;
+import java.util.Set;
+import java.util.regex.Matcher;
+import java.util.regex.Pattern;
 
 import lombok.extern.slf4j.Slf4j;
 
 @Slf4j
 public class SeatunnelTask extends AbstractRemoteTask {
 
-    private static final String SEATUNNEL_BIN_DIR = "${SEATUNNEL_HOME}/bin/";
+    protected static final String SEATUNNEL_BIN_DIR = "${SEATUNNEL_HOME}/bin/";
+
+    private static final Pattern SEATUNNEL_JOB_ID_PATTERN = Pattern.compile(Constants.SEATUNNEL_JOB_ID_REGEX);
 
     private SeatunnelParameters seatunnelParameters;
 
@@ -121,12 +127,60 @@ public void trackApplicationStatus() throws TaskException {
     public void cancelApplication() throws TaskException {
         // cancel process
         try {
-            shellCommandExecutor.cancelApplication();
+            cancelShellProcess();
         } catch (Exception e) {
             throw new TaskException("cancel application error", e);
         }
     }
 
+    protected void cancelShellProcess() throws Exception {
+        shellCommandExecutor.cancelApplication();
+    }
+
+    /**
+     * Extract SeaTunnel Engine job id from a log line.
+     *
+     * @param line log line
+     * @return job id or null
+     */
+    public static String findSeaTunnelJobId(String line) {
+        if (line == null) {
+            return null;
+        }
+        Matcher matcher = SEATUNNEL_JOB_ID_PATTERN.matcher(line);
+        if (matcher.find()) {
+            return matcher.group(1);
+        }
+        return null;
+    }
+
+    /**
+     * Parse SeaTunnel Engine job ids from the task log file.
+     */
+    protected List<String> findSeaTunnelJobIdsFromLog() {
+        String logPath = taskRequest.getLogPath();
+        if (logPath == null || logPath.isEmpty()) {
+            return Collections.emptyList();
+        }
+        File logFile = new File(logPath);
+        if (!logFile.exists() || !logFile.isFile()) {
+            return Collections.emptyList();
+        }
+        Set<String> jobIds = new HashSet<>();
+        try {
+            for (String line : Files.readAllLines(Paths.get(logPath), StandardCharsets.UTF_8)) {
+                String jobId = findSeaTunnelJobId(line);
+                if (jobId != null) {
+                    jobIds.add(jobId);
+                }
+            }
+        } catch (IOException e) {
+            log.error("Failed to parse SeaTunnel job id from log: {}", logPath, e);
+            return Collections.emptyList();
+        }
+        return new ArrayList<>(jobIds);
+    }
+
     private String buildCommand() throws Exception {
 
         List<String> args = new ArrayList<>();

@@ -18,18 +18,27 @@
 package org.apache.dolphinscheduler.plugin.task.seatunnel.self;
 
 import org.apache.dolphinscheduler.common.utils.JSONUtils;
+import org.apache.dolphinscheduler.plugin.task.api.TaskException;
 import org.apache.dolphinscheduler.plugin.task.api.TaskExecutionContext;
 import org.apache.dolphinscheduler.plugin.task.seatunnel.Constants;
 import org.apache.dolphinscheduler.plugin.task.seatunnel.SeatunnelTask;
 
 import org.apache.commons.lang3.StringUtils;
 
+import java.io.IOException;
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.Collections;
 import java.util.List;
 import java.util.Objects;
 
+import lombok.extern.slf4j.Slf4j;
+
+@Slf4j
 public class SeatunnelEngineTask extends SeatunnelTask {
 
     private SeatunnelEngineParameters seatunnelParameters;
+
     public SeatunnelEngineTask(TaskExecutionContext taskExecutionContext) {
         super(taskExecutionContext);
     }
@@ -55,4 +64,62 @@ public List<String> buildOptions() throws Exception {
         return args;
     }
 
+    @Override
+    public List<String> getApplicationIds() throws TaskException {
+        if (StringUtils.isNotEmpty(getAppIds())) {
+            return Arrays.asList(getAppIds().split(","));
+        }
+        List<String> jobIds = findSeaTunnelJobIdsFromLog();
+        if (jobIds != null && !jobIds.isEmpty()) {
+            setAppIds(String.join(",", jobIds));
+        }
+        return jobIds == null ? Collections.emptyList() : jobIds;
+    }
+
+    @Override
+    public void cancelApplication() throws TaskException {
+        List<String> jobIds = Collections.emptyList();
+        try {
+            jobIds = getApplicationIds();
+        } catch (Exception e) {
+            log.warn("Failed to resolve SeaTunnel job id before cancel, will still kill local process", e);
+        }
+
+        try {
+            // Kill local seatunnel client process tree (relies on SeaTunnel -cj/--close-job default)
+            cancelShellProcess();
+        } catch (Exception e) {
+            throw new TaskException("cancel application error", e);
+        }
+
+        // Also cancel the engine job explicitly. Needed when the client is already gone,
+        // --async was used, or process kill alone did not stop a cluster streaming job.
+        if (jobIds != null && !jobIds.isEmpty()) {
+            for (String jobId : jobIds) {
+                cancelSeaTunnelJob(jobId);
+            }
+        } else {
+            log.warn("SeaTunnel job id not found in logs, skipped seatunnel.sh -can");
+        }
+    }
+
+    void cancelSeaTunnelJob(String jobId) throws TaskException {
+        String seatunnelHome = System.getenv("SEATUNNEL_HOME");
+        if (StringUtils.isBlank(seatunnelHome)) {
+            log.warn("SEATUNNEL_HOME is not set, cannot run seatunnel.sh -can {}", jobId);
+            return;
+        }
+        List<String> args = new ArrayList<>();
+        args.add(seatunnelHome + "/bin/seatunnel.sh");
+        args.add(Constants.CANCEL_JOB_OPTIONS);
+        args.add(jobId);
+        log.info("Cancel SeaTunnel job with args: {}", args);
+        ProcessBuilder processBuilder = new ProcessBuilder(args);
+        try {
+            processBuilder.start();
+        } catch (IOException e) {
+            throw new TaskException("Failed to cancel SeaTunnel job: " + jobId, e);
+        }
+    }
+
 }
@@ -0,0 +1,42 @@
+/*
+ * 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.dolphinscheduler.plugin.task.seatunnel;
+
+import org.junit.jupiter.api.Assertions;
+import org.junit.jupiter.api.Test;
+
+public class SeatunnelJobIdParseTest {
+
+    @Test
+    public void testFindSeaTunnelJobIdFromSubmitStartLog() {
+        String line = "Start submit job, job id: 733584788375666689, with plugin jar []";
+        Assertions.assertEquals("733584788375666689", SeatunnelTask.findSeaTunnelJobId(line));
+    }
+
+    @Test
+    public void testFindSeaTunnelJobIdFromSubmitFinishedLog() {
+        String line = "Submit job finished, job id: 898380162133917698, job name: SeaTunnel";
+        Assertions.assertEquals("898380162133917698", SeatunnelTask.findSeaTunnelJobId(line));
+    }
+
+    @Test
+    public void testFindSeaTunnelJobIdReturnsNullWhenMissing() {
+        Assertions.assertNull(SeatunnelTask.findSeaTunnelJobId("no job here"));
+        Assertions.assertNull(SeatunnelTask.findSeaTunnelJobId(null));
+    }
+}