apache · njnu-seafish · Nov 18, 2025 · Nov 18, 2025 · Nov 18, 2025 · Nov 18, 2025
@@ -43,6 +43,7 @@
 import java.sql.CallableStatement;
 import java.sql.Connection;
 import java.sql.SQLException;
+import java.sql.Statement;
 import java.sql.Types;
 import java.util.HashMap;
 import java.util.Map;
@@ -60,6 +61,8 @@ public class ProcedureTask extends AbstractTask {

    private final ProcedureTaskExecutionContext procedureTaskExecutionContext;

+    private volatile Statement sessionStatement;
+
    /**
     * constructor
     *
@@ -105,21 +108,26 @@ public void handle(TaskCallBack taskCallBack) throws TaskException {
            }
            String proceduerSql = formatSql(sqlParamsMap, paramsMap);
            // call method
-            try (CallableStatement stmt = connection.prepareCall(proceduerSql)) {
+            try (CallableStatement tmpStatement = connection.prepareCall(proceduerSql)) {
+                sessionStatement = tmpStatement;
                // set timeout
-                setTimeout(stmt);
+                setTimeout(tmpStatement);

                // outParameterMap
-                Map<Integer, Property> outParameterMap = getOutParameterMap(stmt, sqlParamsMap, paramsMap);
+                Map<Integer, Property> outParameterMap = getOutParameterMap(tmpStatement, sqlParamsMap, paramsMap);

-                stmt.executeUpdate();
+                tmpStatement.executeUpdate();

                // print the output parameters to the log
-                printOutParameter(stmt, outParameterMap);
+                printOutParameter(tmpStatement, outParameterMap);

                setExitStatusCode(EXIT_CODE_SUCCESS);
            }
        } catch (Exception e) {
+            if (exitStatusCode == TaskConstants.EXIT_CODE_KILL) {
+                log.info("procedure task has been killed");
+                return;
+            }
            setExitStatusCode(EXIT_CODE_FAILURE);
            log.error("procedure task error", e);
            throw new TaskException("Execute procedure task failed", e);
@@ -128,7 +136,21 @@ public void handle(TaskCallBack taskCallBack) throws TaskException {

    @Override
    public void cancel() throws TaskException {
-
+        Statement stmt = this.sessionStatement;
+        if (stmt != null) {
+            try {
+                log.debug("Try to cancel this procedure task");
+                setExitStatusCode(TaskConstants.EXIT_CODE_KILL);
+                stmt.cancel();
+                log.debug("this procedure task was canceled");
+            } catch (SQLException ex) {
+                log.warn("Failed to cancel stored procedure (driver/DB may not support it)", ex);
+                throw new TaskException("Cancel procedure task failed", ex);
+            }
+        } else {
+            log.warn(
+                    "Attempted to cancel procedure task, but no active statement exists. Possible reasons: task not started, already completed, or canceled.");
+        }
    }

    private String formatSql(Map<Integer, Property> sqlParamsMap, Map<String, Property> paramsMap) {