diff --git a/dolphinscheduler-task-plugin/dolphinscheduler-task-seatunnel/src/main/java/org/apache/dolphinscheduler/plugin/task/seatunnel/Constants.java b/dolphinscheduler-task-plugin/dolphinscheduler-task-seatunnel/src/main/java/org/apache/dolphinscheduler/plugin/task/seatunnel/Constants.java index 5f0f22b0a1ea..ed8b03af6eb8 100644 --- a/dolphinscheduler-task-plugin/dolphinscheduler-task-seatunnel/src/main/java/org/apache/dolphinscheduler/plugin/task/seatunnel/Constants.java +++ b/dolphinscheduler-task-plugin/dolphinscheduler-task-seatunnel/src/main/java/org/apache/dolphinscheduler/plugin/task/seatunnel/Constants.java @@ -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"; + } diff --git a/dolphinscheduler-task-plugin/dolphinscheduler-task-seatunnel/src/main/java/org/apache/dolphinscheduler/plugin/task/seatunnel/SeatunnelTask.java b/dolphinscheduler-task-plugin/dolphinscheduler-task-seatunnel/src/main/java/org/apache/dolphinscheduler/plugin/task/seatunnel/SeatunnelTask.java index 5f9e766b4f00..ac73f301f934 100644 --- a/dolphinscheduler-task-plugin/dolphinscheduler-task-seatunnel/src/main/java/org/apache/dolphinscheduler/plugin/task/seatunnel/SeatunnelTask.java +++ b/dolphinscheduler-task-plugin/dolphinscheduler-task-seatunnel/src/main/java/org/apache/dolphinscheduler/plugin/task/seatunnel/SeatunnelTask.java @@ -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 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 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 args = new ArrayList<>(); diff --git a/dolphinscheduler-task-plugin/dolphinscheduler-task-seatunnel/src/main/java/org/apache/dolphinscheduler/plugin/task/seatunnel/self/SeatunnelEngineTask.java b/dolphinscheduler-task-plugin/dolphinscheduler-task-seatunnel/src/main/java/org/apache/dolphinscheduler/plugin/task/seatunnel/self/SeatunnelEngineTask.java index f870dfdcd9b7..40e140f2f7e2 100644 --- a/dolphinscheduler-task-plugin/dolphinscheduler-task-seatunnel/src/main/java/org/apache/dolphinscheduler/plugin/task/seatunnel/self/SeatunnelEngineTask.java +++ b/dolphinscheduler-task-plugin/dolphinscheduler-task-seatunnel/src/main/java/org/apache/dolphinscheduler/plugin/task/seatunnel/self/SeatunnelEngineTask.java @@ -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 buildOptions() throws Exception { return args; } + @Override + public List getApplicationIds() throws TaskException { + if (StringUtils.isNotEmpty(getAppIds())) { + return Arrays.asList(getAppIds().split(",")); + } + List jobIds = findSeaTunnelJobIdsFromLog(); + if (jobIds != null && !jobIds.isEmpty()) { + setAppIds(String.join(",", jobIds)); + } + return jobIds == null ? Collections.emptyList() : jobIds; + } + + @Override + public void cancelApplication() throws TaskException { + List 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 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); + } + } + } diff --git a/dolphinscheduler-task-plugin/dolphinscheduler-task-seatunnel/src/test/java/org/apache/dolphinscheduler/plugin/task/seatunnel/SeatunnelJobIdParseTest.java b/dolphinscheduler-task-plugin/dolphinscheduler-task-seatunnel/src/test/java/org/apache/dolphinscheduler/plugin/task/seatunnel/SeatunnelJobIdParseTest.java new file mode 100644 index 000000000000..4d874f00f88e --- /dev/null +++ b/dolphinscheduler-task-plugin/dolphinscheduler-task-seatunnel/src/test/java/org/apache/dolphinscheduler/plugin/task/seatunnel/SeatunnelJobIdParseTest.java @@ -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)); + } +}