[Fix-17316][Task-API] Add check process status after killing task (#17320)

Author: njnu-seafish

deploy/kubernetes/dolphinscheduler/README.md

@@ -155,6 +155,7 @@ Please refer to the [Quick Start in Kubernetes](../../../docs/docs/en/guide/inst
 | conf.common."resource.manager.httpaddress.port" | int | `8088` | resourcemanager port, the default value is 8088 if not specified |
 | conf.common."resource.storage.type" | string | `"S3"` | resource storage type: HDFS, S3, OSS, GCS, ABS, NONE |
 | conf.common."resource.storage.upload.base.path" | string | `"/dolphinscheduler"` | resource store on HDFS/S3 path, resource file will store to this base path, self configuration, please make sure the directory exists on hdfs and have read write permissions. "/dolphinscheduler" is recommended |
+| conf.common."shell.kill.wait.timeout" | int | `10` | If the shell process is still active after this timeout value (in seconds), then will use kill -9 to kill it |
 | conf.common."sudo.enable" | bool | `true` | use sudo or not, if set true, executing user is tenant user and deploy user needs sudo permissions; if set false, executing user is the deploy user and doesn't need sudo permissions |
 | conf.common."support.hive.oneSession" | bool | `false` | Whether hive SQL is executed in the same session |
 | conf.common."task.resource.limit.state" | bool | `false` | Task resource limit state |

deploy/kubernetes/dolphinscheduler/values.yaml

@@ -345,6 +345,9 @@ conf:
     # -- development state
     development.state: false
 
+    # -- If the shell process is still active after this timeout value (in seconds), then will use kill -9 to kill it
+    shell.kill.wait.timeout: 10
+
     # -- set path of conda.sh
     conda.path: /opt/anaconda3/etc/profile.d/conda.sh
 

dolphinscheduler-api-test/dolphinscheduler-api-test-case/src/test/resources/docker/file-manage/common.properties

@@ -99,6 +99,9 @@ sudo.enable=true
 # development state
 development.state=false
 
+# If the shell process is still active after this timeout value (in seconds), then will use kill -9 to kill it
+shell.kill.wait.timeout=10
+
 # set path of conda.sh
 conda.path=/opt/anaconda3/etc/profile.d/conda.sh
 
@@ -111,4 +114,5 @@ ml.mlflow.preset_repository=https://github.com/apache/dolphinscheduler-mlflow
 ml.mlflow.preset_repository_version="main"
 
 # way to collect applicationId: log(original regex match), aop
-appId.collect: log
+appId.collect=log
+

dolphinscheduler-common/src/main/java/org/apache/dolphinscheduler/common/constants/Constants.java

@@ -69,6 +69,11 @@ public final class Constants {
      */
     public static final String DEVELOPMENT_STATE = "development.state";
 
+    /**
+     * shell.kill.wait.timeout: this property defines the wait timeout in seconds before using SIGKILL.
+     */
+    public static final String SHELL_KILL_WAIT_TIMEOUT = "shell.kill.wait.timeout";
+
     /**
      * sudo enable
      */

dolphinscheduler-common/src/main/resources/common.properties

@@ -84,6 +84,9 @@ dolphin.scheduler.network.interface.restrict=docker0
 # development state
 development.state=false
 
+# If the shell process is still active after this timeout value (in seconds), then will use kill -9 to kill it
+shell.kill.wait.timeout=10
+
 # set path of conda.sh
 conda.path=/opt/anaconda3/etc/profile.d/conda.sh
 

dolphinscheduler-common/src/test/resources/common.properties

@@ -148,6 +148,9 @@ dolphin.scheduler.network.interface.restrict=docker0
 # development state
 development.state=false
 
+# If the shell process is still active after this timeout value (in seconds), then will use kill -9 to kill it
+shell.kill.wait.timeout=10
+
 # set path of conda.sh
 conda.path=/opt/anaconda3/etc/profile.d/conda.sh
 

dolphinscheduler-e2e/dolphinscheduler-e2e-case/src/test/resources/docker/file-manage/common.properties

@@ -99,6 +99,9 @@ sudo.enable=true
 # development state
 development.state=false
 
+# If the shell process is still active after this timeout value (in seconds), then will use kill -9 to kill it
+shell.kill.wait.timeout=10
+
 # set path of conda.sh
 conda.path=/opt/anaconda3/etc/profile.d/conda.sh
 
@@ -111,4 +114,4 @@ ml.mlflow.preset_repository=https://github.com/apache/dolphinscheduler-mlflow
 ml.mlflow.preset_repository_version="main"
 
 # way to collect applicationId: log(original regex match), aop
-appId.collect: log
+appId.collect=log

dolphinscheduler-task-plugin/dolphinscheduler-task-api/src/main/java/org/apache/dolphinscheduler/plugin/task/api/utils/ProcessUtils.java

@@ -17,11 +17,14 @@
 
 package org.apache.dolphinscheduler.plugin.task.api.utils;
 
+import static org.apache.dolphinscheduler.common.constants.Constants.SLEEP_TIME_MILLIS;
 import static org.apache.dolphinscheduler.plugin.task.api.TaskConstants.APPID_COLLECT;
 import static org.apache.dolphinscheduler.plugin.task.api.TaskConstants.COMMA;
 import static org.apache.dolphinscheduler.plugin.task.api.TaskConstants.DEFAULT_COLLECT_WAY;
 import static org.apache.dolphinscheduler.plugin.task.api.TaskConstants.TASK_TYPE_SET_K8S;
 
+import org.apache.dolphinscheduler.common.constants.Constants;
+import org.apache.dolphinscheduler.common.thread.ThreadUtils;
 import org.apache.dolphinscheduler.common.utils.OSUtils;
 import org.apache.dolphinscheduler.common.utils.PropertyUtils;
 import org.apache.dolphinscheduler.plugin.task.api.K8sTaskExecutionContext;
@@ -46,8 +49,10 @@
 import java.util.Map;
 import java.util.Objects;
 import java.util.ServiceLoader;
+import java.util.concurrent.TimeUnit;
 import java.util.regex.Matcher;
 import java.util.regex.Pattern;
+import java.util.stream.Collectors;
 
 import lombok.NonNull;
 import lombok.extern.slf4j.Slf4j;
@@ -56,6 +61,10 @@
 @Slf4j
 public final class ProcessUtils {
 
+    // If the shell process is still active after this timeout value (in seconds), then will use kill -9 to kill it
+    private static final Integer SHELL_KILL_WAIT_TIMEOUT =
+            PropertyUtils.getInt(Constants.SHELL_KILL_WAIT_TIMEOUT, 10);
+
     private ProcessUtils() {
         throw new IllegalStateException("Utility class");
     }
@@ -77,13 +86,18 @@ private ProcessUtils() {
     /**
      * Expression of PID recognition in Windows scene
      */
-    private static final Pattern WINDOWSPATTERN = Pattern.compile("(\\d+)");
+    private static final Pattern WINDOWSPATTERN = Pattern.compile("\\((\\d+)\\)");
 
     /**
      * Expression of PID recognition in Linux scene
      */
     private static final Pattern LINUXPATTERN = Pattern.compile("\\((\\d+)\\)");
 
+    /**
+     * PID recognition pattern
+     */
+    private static final Pattern PID_PATTERN = Pattern.compile("\\s+");
+
     /**
      * Terminate the task process, support multi-level signal processing and fallback strategy
      * @param request Task execution context
@@ -100,29 +114,32 @@ public static boolean kill(@NonNull TaskExecutionContext request) {
 
             // Get all child processes
             String pids = getPidsStr(processId);
-            String[] pidArray = pids.split("\\s+");
+            String[] pidArray = PID_PATTERN.split(pids);
             if (pidArray.length == 0) {
                 log.warn("No valid PIDs found for process: {}", processId);
                 return true;
             }
 
+            // Convert PID string to list of integers
+            List<Integer> pidList = Arrays.stream(pidArray).map(Integer::parseInt).collect(Collectors.toList());
+
             // 1. Try to terminate gracefully (SIGINT)
-            boolean gracefulKillSuccess = sendKillSignal("SIGINT", pids, request.getTenantCode());
+            boolean gracefulKillSuccess = sendKillSignal("SIGINT", pidList, request.getTenantCode());
             if (gracefulKillSuccess) {
                 log.info("Successfully killed process tree using SIGINT, processId: {}", processId);
                 return true;
             }
 
             // 2. Try to terminate forcefully (SIGTERM)
-            boolean termKillSuccess = sendKillSignal("SIGTERM", pids, request.getTenantCode());
+            boolean termKillSuccess = sendKillSignal("SIGTERM", pidList, request.getTenantCode());
             if (termKillSuccess) {
                 log.info("Successfully killed process tree using SIGTERM, processId: {}", processId);
                 return true;
             }
 
             // 3. As a last resort, use `kill -9`
             log.warn("SIGINT & SIGTERM failed, using SIGKILL as a last resort for processId: {}", processId);
-            boolean forceKillSuccess = sendKillSignal("SIGKILL", pids, request.getTenantCode());
+            boolean forceKillSuccess = sendKillSignal("SIGKILL", pidList, request.getTenantCode());
             if (forceKillSuccess) {
                 log.info("Successfully sent SIGKILL signal to process tree, processId: {}", processId);
             } else {
@@ -139,23 +156,100 @@ public static boolean kill(@NonNull TaskExecutionContext request) {
     /**
      * Send a kill signal to a process group
      * @param signal Signal type (SIGINT, SIGTERM, SIGKILL)
-     * @param pids Process ID list
+     * @param pidList Process ID list
      * @param tenantCode Tenant code
      */
-    private static boolean sendKillSignal(String signal, String pids, String tenantCode) {
+    private static boolean sendKillSignal(String signal, List<Integer> pidList, String tenantCode) {
+        if (pidList == null || pidList.isEmpty()) {
+            log.info("No process needs to be killed.");
+            return true;
+        }
+
+        List<Integer> alivePidList = getAlivePidList(pidList, tenantCode);
+        if (alivePidList.isEmpty()) {
+            log.info("All processes already terminated.");
+            return true;
+        }
+
+        String pids = alivePidList.stream()
+                .map(String::valueOf)
+                .collect(Collectors.joining(" "));
+
         try {
+            // 1. Send the kill signal
             String killCmd = String.format("kill -s %s %s", signal, pids);
             killCmd = OSUtils.getSudoCmd(tenantCode, killCmd);
             log.info("Sending {} to process group: {}, command: {}", signal, pids, killCmd);
             OSUtils.exeCmd(killCmd);
 
-            return true;
+            // 2. Wait for the processes to terminate with a timeout-based polling mechanism
+            // Max wait time
+            long timeoutMillis = TimeUnit.SECONDS.toMillis(SHELL_KILL_WAIT_TIMEOUT);
+
+            long startTime = System.currentTimeMillis();
+            while (!alivePidList.isEmpty() && (System.currentTimeMillis() - startTime < timeoutMillis)) {
+                // Remove if process is no longer alive
+                alivePidList.removeIf(pid -> !isProcessAlive(pid, tenantCode));
+                if (!alivePidList.isEmpty()) {
+                    // Wait for a short interval before checking process statuses again, to avoid excessive CPU usage
+                    // from tight-loop polling.
+                    ThreadUtils.sleep(SLEEP_TIME_MILLIS);
+                }
+            }
+
+            // 3. Return final result based on whether all processes were terminated
+            if (alivePidList.isEmpty()) {
+                // All processes have been successfully terminated
+                log.debug("Kill command: {}, kill succeeded", killCmd);
+                return true;
+            } else {
+                String remainingPids = alivePidList.stream()
+                        .map(String::valueOf)
+                        .collect(Collectors.joining(" "));
+                log.info("Kill command: {}, timed out, still running PIDs: {}", killCmd, remainingPids);
+                return false;
+            }
         } catch (Exception e) {
             log.error("Error sending {} to process: {}", signal, pids, e);
             return false;
         }
     }
 
+    /**
+     * Returns a list of process IDs that are still running.
+     * This method filters the provided list of PIDs by checking whether each process is still active
+     *
+     * @param pidList   the list of process IDs to check
+     * @param tenantCode the tenant identifier used for permission control or logging context
+     * @return a new list containing only the PIDs of processes that are still running;
+     *         returns an empty list if none are alive
+     */
+    private static List<Integer> getAlivePidList(List<Integer> pidList, String tenantCode) {
+        return pidList.stream()
+                .filter(pid -> isProcessAlive(pid, tenantCode))
+                .collect(Collectors.toList());
+    }
+
+    /**
+     * Check if a process with the specified PID is alive.
+     *
+     * @param pid the process ID to check
+     * @return true if the process exists and is running, false otherwise
+     */
+    private static boolean isProcessAlive(int pid, String tenantCode) {
+        try {
+            // Use kill -0 to check if the process exists; it does not actually send a signal
+            String checkCmd = String.format("kill -0 %d", pid);
+            checkCmd = OSUtils.getSudoCmd(tenantCode, checkCmd);
+            OSUtils.exeCmd(checkCmd);
+            // If the command executes successfully, the process exists
+            return true;
+        } catch (Exception e) {
+            // If the command fails, the process does not exist
+            return false;
+        }
+    }
+
     /**
      * get pids str.
      *
@@ -249,6 +343,7 @@ public static void cancelApplication(TaskExecutionContext taskExecutionContext)
                 }
                 ApplicationManager applicationManager = applicationManagerMap.get(ResourceManagerType.YARN);
                 applicationManager.killApplication(new YarnApplicationManagerContext(executePath, tenantCode, appIds));
+                log.info("yarn application [{}] is killed or already finished", appIds);
             }
         } catch (Exception e) {
             log.error("Cancel application failed: {}", e.getMessage());