Merge pull request #4743 from apache/params-trans

[Feature-3805][mater-worker-ui] The implement of passing variables between tasks

dolphinscheduler-dao/src/main/java/org/apache/dolphinscheduler/dao/mapper/ProcessInstanceMapper.java

@@ -221,4 +221,7 @@ public interface ProcessInstanceMapper extends BaseMapper<ProcessInstance> {
             @Param("processDefinitionId") int processDefinitionId,
             @Param("states") int[] states);
 
+    int updateGlobalParamsById(
+            @Param("globalParams") String globalParams,
+            @Param("id")  int id);
 }

dolphinscheduler-dao/src/main/resources/datasource.properties

@@ -66,4 +66,4 @@ spring.datasource.password=test
 
 # open PSCache, specify count PSCache for every connection
 #spring.datasource.poolPreparedStatements=true
-#spring.datasource.maxPoolPreparedStatementPerConnectionSize=20
+#spring.datasource.maxPoolPreparedStatementPerConnectionSize=20
\ No newline at end of file

dolphinscheduler-dao/src/main/resources/org/apache/dolphinscheduler/dao/mapper/ProcessInstanceMapper.xml

@@ -219,5 +219,9 @@
         </foreach>
         order by id asc
     </select>
-
+    <update id="updateGlobalParamsById">
+        update t_ds_process_instance
+        set global_params = #{globalParams}
+        where id = #{id}
+    </update>
 </mapper>

dolphinscheduler-remote/src/main/java/org/apache/dolphinscheduler/remote/command/TaskExecuteResponseCommand.java

@@ -68,6 +68,10 @@ public class TaskExecuteResponseCommand implements Serializable {
      * varPool string
      */
     private String varPool;
+    /**
+     * task return result
+     */
+    private String result;
 
     public void setVarPool(String varPool) {
         this.varPool = varPool;
@@ -139,4 +143,12 @@ public class TaskExecuteResponseCommand implements Serializable {
                 + ", appIds='" + appIds + '\''
                 + '}';
     }
+
+    public String getResult() {
+        return result;
+    }
+
+    public void setResult(String result) {
+        this.result = result;
+    }
 }

dolphinscheduler-server/src/main/java/org/apache/dolphinscheduler/server/master/processor/TaskResponseProcessor.java

@@ -80,7 +80,9 @@ public class TaskResponseProcessor implements NettyRequestProcessor {
                 responseCommand.getAppIds(),
                 responseCommand.getTaskInstanceId(),
                 responseCommand.getVarPool(),
-                channel);
+                channel,
+                responseCommand.getResult()
+                );
         taskResponseService.addResponse(taskResponseEvent);
     }
 

dolphinscheduler-server/src/main/java/org/apache/dolphinscheduler/server/master/processor/queue/TaskResponseEvent.java

@@ -92,6 +92,10 @@ public class TaskResponseEvent {
      * channel
      */
     private Channel channel;
+    /**
+     * task return result
+     */
+    private String result;
     
     public static TaskResponseEvent newAck(ExecutionStatus state,
                                            Date startTime,
@@ -118,7 +122,8 @@ public class TaskResponseEvent {
                                               String appIds,
                                               int taskInstanceId,
                                               String varPool,
-                                              Channel channel) {
+                                              Channel channel,
+                                              String result) {
         TaskResponseEvent event = new TaskResponseEvent();
         event.setState(state);
         event.setEndTime(endTime);
@@ -128,6 +133,7 @@ public class TaskResponseEvent {
         event.setEvent(Event.RESULT);
         event.setVarPool(varPool);
         event.setChannel(channel);
+        event.setResult(result);
         return event;
     }
 
@@ -226,4 +232,12 @@ public class TaskResponseEvent {
     public void setChannel(Channel channel) {
         this.channel = channel;
     }
+
+    public String getResult() {
+        return result;
+    }
+
+    public void setResult(String result) {
+        this.result = result;
+    }
 }

dolphinscheduler-server/src/main/java/org/apache/dolphinscheduler/server/master/processor/queue/TaskResponseService.java

@@ -165,7 +165,8 @@ public class TaskResponseService {
                             taskResponseEvent.getProcessId(),
                             taskResponseEvent.getAppIds(),
                             taskResponseEvent.getTaskInstanceId(),
-                            taskResponseEvent.getVarPool()
+                            taskResponseEvent.getVarPool(),
+                                taskResponseEvent.getResult()
                         );
                     }
                     // if taskInstance is null (maybe deleted) . retry will be meaningless . so response success

dolphinscheduler-server/src/main/java/org/apache/dolphinscheduler/server/master/runner/MasterExecThread.java

@@ -22,11 +22,13 @@ import static org.apache.dolphinscheduler.common.Constants.CMDPARAM_COMPLEMENT_D
 import static org.apache.dolphinscheduler.common.Constants.CMD_PARAM_RECOVERY_START_NODE_STRING;
 import static org.apache.dolphinscheduler.common.Constants.CMD_PARAM_START_NODE_NAMES;
 import static org.apache.dolphinscheduler.common.Constants.DEFAULT_WORKER_GROUP;
+import static org.apache.dolphinscheduler.common.Constants.LOCAL_PARAMS;
 import static org.apache.dolphinscheduler.common.Constants.SEC_2_MINUTES_TIME_UNIT;
 
 import org.apache.dolphinscheduler.common.Constants;
 import org.apache.dolphinscheduler.common.enums.CommandType;
 import org.apache.dolphinscheduler.common.enums.DependResult;
+import org.apache.dolphinscheduler.common.enums.Direct;
 import org.apache.dolphinscheduler.common.enums.ExecutionStatus;
 import org.apache.dolphinscheduler.common.enums.FailureStrategy;
 import org.apache.dolphinscheduler.common.enums.Flag;
@@ -36,6 +38,7 @@ import org.apache.dolphinscheduler.common.graph.DAG;
 import org.apache.dolphinscheduler.common.model.TaskNode;
 import org.apache.dolphinscheduler.common.model.TaskNodeRelation;
 import org.apache.dolphinscheduler.common.process.ProcessDag;
+import org.apache.dolphinscheduler.common.process.Property;
 import org.apache.dolphinscheduler.common.thread.Stopper;
 import org.apache.dolphinscheduler.common.thread.ThreadUtils;
 import org.apache.dolphinscheduler.common.utils.CollectionUtils;
@@ -67,6 +70,7 @@ import java.util.ArrayList;
 import java.util.Arrays;
 import java.util.Collection;
 import java.util.Date;
+import java.util.HashMap;
 import java.util.Iterator;
 import java.util.List;
 import java.util.Map;
@@ -74,6 +78,7 @@ import java.util.Set;
 import java.util.concurrent.ConcurrentHashMap;
 import java.util.concurrent.ExecutorService;
 import java.util.concurrent.Future;
+import java.util.stream.Collectors;
 
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
@@ -491,7 +496,8 @@ public class MasterExecThread implements Runnable {
      */
     private TaskInstance createTaskInstance(ProcessInstance processInstance, String nodeName,
                                             TaskNode taskNode) {
-
+        //update processInstance for update the globalParams
+        this.processInstance = this.processService.findProcessInstanceById(this.processInstance.getId());
         TaskInstance taskInstance = findTaskIfExists(nodeName);
         if (taskInstance == null) {
             taskInstance = new TaskInstance();
@@ -540,13 +546,57 @@ public class MasterExecThread implements Runnable {
             } else {
                 taskInstance.setWorkerGroup(taskWorkerGroup);
             }
-
+            //get process global
+            setProcessGlobal(taskNode, taskInstance);
             // delay execution time
             taskInstance.setDelayTime(taskNode.getDelayTime());
         }
         return taskInstance;
     }
 
+    private void setProcessGlobal(TaskNode taskNode, TaskInstance taskInstance) {
+        String globalParams = this.processInstance.getGlobalParams();
+        if (StringUtils.isNotEmpty(globalParams)) {
+            Map<String, String> globalMap = getGlobalParamMap(globalParams);
+            if (globalMap != null && globalMap.size() != 0) {
+                setGlobalMapToTask(taskNode, taskInstance, globalMap);
+            }
+        }
+    }
+
+    private void setGlobalMapToTask(TaskNode taskNode, TaskInstance taskInstance, Map<String, String> globalMap) {
+        // the param save in localParams
+        Map<String, Object> result = JSONUtils.toMap(taskNode.getParams(), String.class, Object.class);
+        Object localParams = result.get(LOCAL_PARAMS);
+        if (localParams != null) {
+            List<Property> allParam = JSONUtils.toList(JSONUtils.toJsonString(localParams), Property.class);
+            for (Property info : allParam) {
+                if (info.getDirect().equals(Direct.IN)) {
+                    String paramName = info.getProp();
+                    String value = globalMap.get(paramName);
+                    if (StringUtils.isNotEmpty(value)) {
+                        info.setValue(value);
+                    }
+                }
+            }
+            result.put(LOCAL_PARAMS, allParam);
+            taskNode.setParams(JSONUtils.toJsonString(result));
+            // task instance node json
+            taskInstance.setTaskJson(JSONUtils.toJsonString(taskNode));
+        }
+    }
+
+    public Map<String, String> getGlobalParamMap(String globalParams) {
+        List<Property> propList;
+        Map<String,String> globalParamMap = new HashMap<>();
+        if (StringUtils.isNotEmpty(globalParams)) {
+            propList = JSONUtils.toList(globalParams, Property.class);
+            globalParamMap = propList.stream().collect(Collectors.toMap(Property::getProp, Property::getValue));
+        }
+
+        return globalParamMap;
+    }
+
     private void submitPostNode(String parentNodeName) {
         Set<String> submitTaskNodeList = DagHelper.parsePostNodes(parentNodeName, skipTaskNodeList, dag, completeTaskList);
         List<TaskInstance> taskInstances = new ArrayList<>();
@@ -951,6 +1001,7 @@ public class MasterExecThread implements Runnable {
                         task.getName(), task.getId(), task.getState());
                 // node success , post node submit
                 if (task.getState() == ExecutionStatus.SUCCESS) {
+                    processInstance = processService.findProcessInstanceById(processInstance.getId());
                     processInstance.setVarPool(task.getVarPool());
                     processService.updateProcessInstance(processInstance);
                     completeTaskList.put(task.getName(), task);

dolphinscheduler-server/src/main/java/org/apache/dolphinscheduler/server/worker/runner/TaskExecuteThread.java

@@ -174,6 +174,7 @@ public class TaskExecuteThread implements Runnable, Delayed {
             responseCommand.setProcessId(task.getProcessId());
             responseCommand.setAppIds(task.getAppIds());
             responseCommand.setVarPool(task.getVarPool());
+            responseCommand.setResult(task.getResultString());
             logger.info("task instance id : {},task final status : {}", taskExecutionContext.getTaskInstanceId(), task.getExitStatus());
         } catch (Exception e) {
             logger.error("task scheduler failure", e);

dolphinscheduler-server/src/main/java/org/apache/dolphinscheduler/server/worker/task/AbstractCommandExecutor.java

@@ -54,6 +54,7 @@ import java.util.regex.Pattern;
 
 import org.slf4j.Logger;
 
+
 /**
  * abstract command executor
  */
@@ -84,6 +85,11 @@ public abstract class AbstractCommandExecutor {
      */
     protected final List<String> logBuffer;
 
+    /**
+     * SHELL result string
+     */
+    protected String taskResultString;
+
     /**
      * taskExecutionContext
      */
@@ -104,6 +110,10 @@ public abstract class AbstractCommandExecutor {
         this.taskExecutionContextCacheManager = SpringApplicationContext.getBean(TaskExecutionContextCacheManagerImpl.class);
     }
 
+    protected AbstractCommandExecutor(List<String> logBuffer) {
+        this.logBuffer = logBuffer;
+    }
+
     /**
      * build process
      *
@@ -223,6 +233,7 @@ public abstract class AbstractCommandExecutor {
         return varPool.toString();
     }
 
+
     /**
      * cancel application
      *
@@ -355,6 +366,7 @@ public abstract class AbstractCommandExecutor {
                             varPool.append("$VarPool$");
                         } else {
                             logBuffer.add(line);
+                            taskResultString = line;
                             lastFlushTime = flush(lastFlushTime);
                         }
                     }
@@ -561,4 +573,12 @@ public abstract class AbstractCommandExecutor {
     protected abstract String commandInterpreter();
 
     protected abstract void createCommandFileIfNotExists(String execCommand, String commandFile) throws IOException;
+
+    public String getTaskResultString() {
+        return taskResultString;
+    }
+
+    public void setTaskResultString(String taskResultString) {
+        this.taskResultString = taskResultString;
+    }
 }
\ No newline at end of file

dolphinscheduler-server/src/main/java/org/apache/dolphinscheduler/server/worker/task/AbstractTask.java

@@ -63,6 +63,11 @@ public abstract class AbstractTask {
      */
     protected int processId;
 
+    /**
+     * SHELL result string
+     */
+    protected String resultString;
+
     /**
      * other resource manager appId , for example : YARN etc
      */
@@ -167,6 +172,14 @@ public abstract class AbstractTask {
         this.processId = processId;
     }
 
+    public String getResultString() {
+        return resultString;
+    }
+
+    public void setResultString(String resultString) {
+        this.resultString = resultString;
+    }
+
     /**
      * get task parameters
      *

dolphinscheduler-server/src/main/java/org/apache/dolphinscheduler/server/worker/task/ShellCommandExecutor.java

@@ -56,6 +56,9 @@ public class ShellCommandExecutor extends AbstractCommandExecutor {
         super(logHandler,taskExecutionContext,logger);
     }
 
+    public ShellCommandExecutor(List<String> logBuffer) {
+        super(logBuffer);
+    }
 
     @Override
     protected String buildCommandFilePath() {

dolphinscheduler-server/src/main/java/org/apache/dolphinscheduler/server/worker/task/shell/ShellTask.java

@@ -21,6 +21,7 @@ import static java.util.Calendar.DAY_OF_MONTH;
 
 import org.apache.dolphinscheduler.common.Constants;
 import org.apache.dolphinscheduler.common.enums.CommandType;
+import org.apache.dolphinscheduler.common.enums.Direct;
 import org.apache.dolphinscheduler.common.process.Property;
 import org.apache.dolphinscheduler.common.task.AbstractParameters;
 import org.apache.dolphinscheduler.common.task.shell.ShellParameters;
@@ -34,6 +35,8 @@ import org.apache.dolphinscheduler.server.worker.task.AbstractTask;
 import org.apache.dolphinscheduler.server.worker.task.CommandExecuteResult;
 import org.apache.dolphinscheduler.server.worker.task.ShellCommandExecutor;
 
+import org.slf4j.Logger;
+
 import java.io.File;
 import java.nio.file.Files;
 import java.nio.file.Path;
@@ -41,13 +44,13 @@ import java.nio.file.StandardOpenOption;
 import java.nio.file.attribute.FileAttribute;
 import java.nio.file.attribute.PosixFilePermission;
 import java.nio.file.attribute.PosixFilePermissions;
+import java.util.ArrayList;
 import java.util.Date;
 import java.util.HashMap;
+import java.util.List;
 import java.util.Map;
 import java.util.Set;
 
-import org.slf4j.Logger;
-
 /**
  * shell task
  */
@@ -102,6 +105,7 @@ public class ShellTask extends AbstractTask {
             setExitStatusCode(commandExecuteResult.getExitStatusCode());
             setAppIds(commandExecuteResult.getAppIds());
             setProcessId(commandExecuteResult.getProcessId());
+            setResult(shellCommandExecutor.getTaskResultString());
         } catch (Exception e) {
             logger.error("shell task error", e);
             setExitStatusCode(Constants.EXIT_CODE_FAILURE);
@@ -183,4 +187,17 @@ public class ShellTask extends AbstractTask {
         }
         return ParameterUtils.convertParameterPlaceholders(script, ParamUtils.convert(paramsMap));
     }
+
+    public void setResult(String result) {
+        Map<String, Property> localParams = shellParameters.getLocalParametersMap();
+        List<Map<String, String>> outProperties = new ArrayList<>();
+        Map<String, String> p = new HashMap<>();
+        localParams.forEach((k,v) -> {
+            if (v.getDirect() == Direct.OUT) {
+                p.put(k, result);
+            }
+        });
+        outProperties.add(p);
+        resultString = JSONUtils.toJsonString(outProperties);
+    }
 }

dolphinscheduler-server/src/main/java/org/apache/dolphinscheduler/server/worker/task/sql/SqlTask.java

@@ -20,6 +20,7 @@ package org.apache.dolphinscheduler.server.worker.task.sql;
 import org.apache.dolphinscheduler.common.Constants;
 import org.apache.dolphinscheduler.common.enums.CommandType;
 import org.apache.dolphinscheduler.common.enums.DbType;
+import org.apache.dolphinscheduler.common.enums.Direct;
 import org.apache.dolphinscheduler.common.enums.TaskTimeoutStrategy;
 import org.apache.dolphinscheduler.common.process.Property;
 import org.apache.dolphinscheduler.common.task.AbstractParameters;
@@ -148,7 +149,7 @@ public class SqlTask extends AbstractTask {
                     logger);
 
             // execute sql task
-            executeFuncAndSql(mainSqlBinds, preStatementSqlBinds, postStatementSqlBinds, createFuncs);
+            executeFuncAndSql(mainSqlBinds, preStatementSqlBinds, postStatementSqlBinds, createFuncs, sqlParameters.getLocalParams());
 
             setExitStatusCode(Constants.EXIT_CODE_SUCCESS);
 
@@ -237,7 +238,8 @@ public class SqlTask extends AbstractTask {
     public void executeFuncAndSql(SqlBinds mainSqlBinds,
                                   List<SqlBinds> preStatementsBinds,
                                   List<SqlBinds> postStatementsBinds,
-                                  List<String> createFuncs) {
+                                  List<String> createFuncs,
+                                  List<Property> properties) {
         Connection connection = null;
         PreparedStatement stmt = null;
         ResultSet resultSet = null;
@@ -253,18 +255,21 @@ public class SqlTask extends AbstractTask {
             preSql(connection, preStatementsBinds);
             stmt = prepareStatementAndBind(connection, mainSqlBinds);
 
+            String result = null;
             // decide whether to executeQuery or executeUpdate based on sqlType
             if (sqlParameters.getSqlType() == SqlType.QUERY.ordinal()) {
                 // query statements need to be convert to JsonArray and inserted into Alert to send
                 resultSet = stmt.executeQuery();
-                resultProcess(resultSet);
+                result = resultProcess(resultSet);
 
             } else if (sqlParameters.getSqlType() == SqlType.NON_QUERY.ordinal()) {
                 // non query statement
-                stmt.executeUpdate();
+                String updateResult = String.valueOf(stmt.executeUpdate());
+                result = setNonQuerySqlReturn(updateResult, properties);
             }
 
             postSql(connection, postStatementsBinds);
+            this.setResultString(result);
 
         } catch (Exception e) {
             logger.error("execute sql error", e);
@@ -274,13 +279,28 @@ public class SqlTask extends AbstractTask {
         }
     }
 
+    public String setNonQuerySqlReturn(String updateResult, List<Property> properties) {
+        String result = null;
+        for (Property info :properties) {
+            if (Direct.OUT == info.getDirect()) {
+                List<Map<String,String>> updateRL = new ArrayList<>();
+                Map<String,String> updateRM = new HashMap<>();
+                updateRM.put(info.getProp(),updateResult);
+                updateRL.add(updateRM);
+                result = JSONUtils.toJsonString(updateRL);
+                break;
+            }
+        }
+        return result;
+    }
+
     /**
      * result process
      *
      * @param resultSet resultSet
      * @throws Exception Exception
      */
-    private void resultProcess(ResultSet resultSet) throws Exception {
+    private String resultProcess(ResultSet resultSet) throws Exception {
         ArrayNode resultJSONArray = JSONUtils.createArrayNode();
         ResultSetMetaData md = resultSet.getMetaData();
         int num = md.getColumnCount();
@@ -297,13 +317,13 @@ public class SqlTask extends AbstractTask {
         }
         String result = JSONUtils.toJsonString(resultJSONArray);
         logger.debug("execute sql : {}", result);
-
         try {
             sendAttachment(sqlParameters.getGroupId(), StringUtils.isNotEmpty(sqlParameters.getTitle()) ? sqlParameters.getTitle() : taskExecutionContext.getTaskName() + " query result sets",
                     JSONUtils.toJsonString(resultJSONArray));
         } catch (Exception e) {
             logger.warn("sql task sendAttachment error! msg : {} ", e.getMessage());
         }
+        return result;
     }
 
     /**

dolphinscheduler-server/src/test/java/org/apache/dolphinscheduler/server/master/processor/queue/TaskResponseServiceTest.java

@@ -70,7 +70,8 @@ public class TaskResponseServiceTest {
             "ids",
             22,
             "varPol",
-            channel);
+            channel,
+                "[{\"id\":70000,\"database_name\":\"yuul\",\"status\":-1,\"create_time\":1601202829000,\"update_time\":1601202829000,\"table_name3\":\"\",\"table_name4\":\"\"}]");
 
         taskInstance = new TaskInstance();
         taskInstance.setId(22);

dolphinscheduler-server/src/test/java/org/apache/dolphinscheduler/server/worker/processor/TaskCallbackServiceTest.java

@@ -121,6 +121,11 @@ public class TaskCallbackServiceTest {
         ackCommand.setStartTime(new Date());
         taskCallbackService.sendAck(1, ackCommand.convert2Command());
 
+        TaskExecuteResponseCommand responseCommand = new TaskExecuteResponseCommand();
+        String result = responseCommand.getResult();
+        responseCommand.setResult("return string");
+        taskCallbackService.sendResult(1, responseCommand.convert2Command());
+
         Stopper.stop();
 
         nettyRemotingServer.close();

dolphinscheduler-server/src/test/java/org/apache/dolphinscheduler/server/worker/task/AbstractCommandExecutorTest.java

+/*
+ * 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.server.worker.task;
+
+import org.apache.dolphinscheduler.service.bean.SpringApplicationContext;
+
+import org.junit.Before;
+import org.junit.Test;
+import org.junit.runner.RunWith;
+import org.powermock.core.classloader.annotations.PrepareForTest;
+import org.powermock.modules.junit4.PowerMockRunner;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+@RunWith(PowerMockRunner.class)
+@PrepareForTest({SpringApplicationContext.class})
+public class AbstractCommandExecutorTest {
+
+    private static final Logger logger = LoggerFactory.getLogger(AbstractCommandExecutorTest.class);
+
+    private ShellCommandExecutor shellCommandExecutor;
+
+    @Before
+    public void before() throws Exception {
+        System.setProperty("log4j2.disable.jmx", Boolean.TRUE.toString());
+        shellCommandExecutor = new ShellCommandExecutor(null);
+    }
+
+    @Test
+    public void testSetTaskResultString() {
+        shellCommandExecutor.setTaskResultString("shellReturn");
+    }
+
+    @Test
+    public void testGetTaskResultString() {
+        logger.info(shellCommandExecutor.getTaskResultString());
+    }
+}
\ No newline at end of file

dolphinscheduler-server/src/test/java/org/apache/dolphinscheduler/server/worker/task/ShellTaskReturnTest.java

+/*
+ * 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.server.worker.task;
+
+import static org.mockito.ArgumentMatchers.anyString;
+
+import org.apache.dolphinscheduler.server.entity.TaskExecutionContext;
+import org.apache.dolphinscheduler.server.worker.task.CommandExecuteResult;
+import org.apache.dolphinscheduler.server.worker.task.ShellCommandExecutor;
+import org.apache.dolphinscheduler.server.worker.task.shell.ShellTask;
+import org.apache.dolphinscheduler.server.worker.task.shell.ShellTaskTest;
+
+import java.nio.file.Files;
+import java.nio.file.Paths;
+import java.util.Date;
+import java.util.HashMap;
+import java.util.Map;
+
+import org.junit.Before;
+import org.junit.Test;
+import org.junit.runner.RunWith;
+import org.powermock.api.mockito.PowerMockito;
+import org.powermock.core.classloader.annotations.PrepareForTest;
+import org.powermock.modules.junit4.PowerMockRunner;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+/**
+ * shell task return test.
+ */
+@RunWith(PowerMockRunner.class)
+@PrepareForTest({ShellTask.class})
+public class ShellTaskReturnTest {
+    private static final Logger logger = LoggerFactory.getLogger(ShellTaskTest.class);
+
+    private ShellTask shellTask;
+    private ShellCommandExecutor shellCommandExecutor;
+    private TaskExecutionContext taskExecutionContext;
+    private CommandExecuteResult commandExecuteResult;
+
+    @Before
+    public void before() throws Exception {
+        System.setProperty("log4j2.disable.jmx", Boolean.TRUE.toString());
+        shellCommandExecutor = PowerMockito.mock(ShellCommandExecutor.class);
+        PowerMockito.whenNew(ShellCommandExecutor.class).withAnyArguments().thenReturn(shellCommandExecutor);
+        taskExecutionContext = new TaskExecutionContext();
+        taskExecutionContext.setTaskInstanceId(1);
+        taskExecutionContext.setTaskName("kris test");
+        taskExecutionContext.setTaskType("SHELL");
+        taskExecutionContext.setHost("127.0.0.1:1234");
+        taskExecutionContext.setExecutePath("/tmp");
+        taskExecutionContext.setLogPath("/log");
+        taskExecutionContext.setTaskJson(
+                "{\"conditionResult\":\"{\\\"successNode\\\":[\\\"\\\"],"
+                        + "\\\"failedNode\\\":[\\\"\\\"]}\",\"conditionsTask\":false,"
+                        + "\"depList\":[],\"dependence\":\"{}\",\"forbidden\":false,\"id\":\""
+                        + "tasks-16849\",\"maxRetryTimes\":0,\"name\":\"shell test 001\","
+                        + "\"params\":\"{\\\"rawScript\\\":\\\"#!/bin/sh\\\\necho $[yyyy-MM-dd HH:mm:ss +3]\\\\necho \\\\\\\" ?? "
+                        + "${time1} \\\\\\\"\\\\necho \\\\\\\" ????? ${time2}\\\\\\\"\\\\n\\\","
+                        + "\\\"localParams\\\":[{\\\"prop\\\":\\\"time1\\\",\\\"direct\\\":\\\"IN\\\",\\\"type\\\":"
+                        + "\\\"VARCHAR\\\",\\\"value\\\":\\\"$[yyyy-MM-dd HH:mm:ss]\\\"},"
+                        + "{\\\"prop\\\":\\\"time2\\\",\\\"direct\\\":\\\"IN\\\",\\\"type\\\":\\\"VARCHAR\\\",\\\"value\\\":\\\"${time_gb}\\\"}"
+                        + "],\\\"resourceList\\\":[]}\",\"preTasks\":\"[]\",\"retryInterval\":1,"
+                        + "\"runFlag\":\"NORMAL\",\"taskInstancePriority\":\"MEDIUM\",\"taskTimeoutParameter\":"
+                        + "{\"enable\":false,\"interval\":0},\"timeout\":\"{\\\"enable\\\":false,\\\"strategy\\\":\\\"\\\"}\","
+                        + "\"type\":\"SHELL\",\"workerGroup\":\"default\"}");
+        taskExecutionContext.setProcessInstanceId(1);
+        taskExecutionContext.setGlobalParams("[{\"direct\":\"IN\",\"prop\":\"time_gb\",\"type\":\"VARCHAR\",\"value\":\"2020-12-16 17:18:33\"}]");
+        taskExecutionContext.setExecutorId(1);
+        taskExecutionContext.setCmdTypeIfComplement(5);
+        taskExecutionContext.setTenantCode("roo");
+        taskExecutionContext.setScheduleTime(new Date());
+        taskExecutionContext.setQueue("default");
+        taskExecutionContext.setTaskParams(
+                "{\"rawScript\":\"#!/bin/sh\\necho $[yyyy-MM-dd HH:mm:ss +3]\\necho \\\" ?? ${time1} \\\"\\necho \\\" ????? ${time2}\\\"\\n\",\"localParams\":"
+                        +
+                        "[{\"prop\":\"time1\",\"direct\":\"OUT\",\"type\":\"VARCHAR\",\"value\":\"$[yyyy-MM-dd HH:mm:ss]\"},{\"prop\":\"time2\",\"direct\":\"IN\",\"type\":\"VARCHAR"
+                        + "\",\"value\":\"${time_gb}\"}],\"resourceList\":[]}");
+        Map<String, String> definedParams = new HashMap<>();
+        definedParams.put("time_gb", "2020-12-16 00:00:00");
+        taskExecutionContext.setDefinedParams(definedParams);
+        PowerMockito.mockStatic(Files.class);
+        PowerMockito.when(Files.exists(Paths.get(anyString()))).thenReturn(true);
+        commandExecuteResult = new CommandExecuteResult();
+        commandExecuteResult.setAppIds("appId");
+        commandExecuteResult.setExitStatusCode(0);
+        commandExecuteResult.setProcessId(1);
+    }
+
+    @Test
+    public void testShellReturnString() {
+        shellTask = new ShellTask(taskExecutionContext, logger);
+        shellTask.init();
+        try {
+            PowerMockito.when(shellCommandExecutor.run(anyString())).thenReturn(commandExecuteResult);
+        } catch (Exception e) {
+            e.printStackTrace();
+        }
+        shellTask.setResult("shell return string");
+        logger.info("shell return string:{}", shellTask.getResultString());
+    }
+
+    @Test
+    public void testSetTaskResultString() {
+        shellCommandExecutor.setTaskResultString("shellReturn");
+    }
+
+    @Test
+    public void testGetTaskResultString() {
+        logger.info(shellCommandExecutor.getTaskResultString());
+    }
+}

dolphinscheduler-server/src/test/java/org/apache/dolphinscheduler/server/worker/task/TaskManagerTest.java

@@ -17,13 +17,22 @@
 
 package org.apache.dolphinscheduler.server.worker.task;
 
+import org.apache.dolphinscheduler.common.process.Property;
+import org.apache.dolphinscheduler.common.task.sql.SqlParameters;
+import org.apache.dolphinscheduler.common.utils.JSONUtils;
 import org.apache.dolphinscheduler.common.utils.LoggerUtils;
+import org.apache.dolphinscheduler.server.entity.SQLTaskExecutionContext;
 import org.apache.dolphinscheduler.server.entity.TaskExecutionContext;
 import org.apache.dolphinscheduler.server.worker.cache.impl.TaskExecutionContextCacheManagerImpl;
+import org.apache.dolphinscheduler.server.worker.task.shell.ShellTask;
+import org.apache.dolphinscheduler.server.worker.task.sql.SqlTask;
 import org.apache.dolphinscheduler.service.alert.AlertClientService;
 import org.apache.dolphinscheduler.service.bean.SpringApplicationContext;
 
 import java.util.Date;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
 
 import org.junit.Assert;
 import org.junit.Before;
@@ -116,4 +125,70 @@ public class TaskManagerTest {
         taskExecutionContext.setTaskType("XXX");
         TaskManager.newTask(taskExecutionContext,taskLogger,alertClientService);
     }
+
+    @Test
+    public void testShellTaskReturnString() {
+        taskExecutionContext.setTaskInstanceId(1);
+        taskExecutionContext.setTaskName("kris test");
+        taskExecutionContext.setTaskType("SHELL");
+        taskExecutionContext.setHost("127.0.0.1:1234");
+        taskExecutionContext.setExecutePath("/tmp");
+        taskExecutionContext.setLogPath("/log");
+        taskExecutionContext.setTaskJson(
+                "{\"conditionResult\":\"{\\\"successNode\\\":[\\\"\\\"],\\\"failedNode\\\":[\\\"\\\"]}\","
+                        + "\"conditionsTask\":false,\"depList\":[],\"dependence\":\"{}\",\"forbidden\":false,\"id\":\""
+                        + "tasks-16849\",\"maxRetryTimes\":0,\"name\":\"shell test 001\","
+                        + "\"params\":\"{\\\"rawScript\\\":\\\"#!/bin/sh\\\\necho $[yyyy-MM-dd HH:mm:ss +3]\\\\necho \\\\\\\" ?? "
+                        + "${time1} \\\\\\\"\\\\necho \\\\\\\" ????? ${time2}\\\\\\\"\\\\n\\\","
+                        + "\\\"localParams\\\":[{\\\"prop\\\":\\\"time1\\\",\\\"direct\\\":\\\"OUT\\\",\\\"type\\\":"
+                        + "\\\"VARCHAR\\\",\\\"value\\\":\\\"$[yyyy-MM-dd HH:mm:ss]\\\"},"
+                        + "{\\\"prop\\\":\\\"time2\\\",\\\"direct\\\":\\\"IN\\\",\\\"type\\\":\\\"VARCHAR\\\",\\\"value\\\":\\\"${time_gb}\\\"}"
+                        + "],\\\"resourceList\\\":[]}\",\"preTasks\":\"[]\",\"retryInterval\":1,\"runFlag\":\"NORMAL\",\"taskInstancePriority\":\"MEDIUM\",\"taskTimeoutParameter\":"
+                        + "{\"enable\":false,\"interval\":0},\"timeout\":\"{\\\"enable\\\":false,\\\"strategy\\\":\\\"\\\"}\",\"type\":\"SHELL\",\"workerGroup\":\"default\"}");
+        taskExecutionContext.setProcessInstanceId(1);
+        taskExecutionContext.setGlobalParams("[{\"direct\":\"IN\",\"prop\":\"time_gb\",\"type\":\"VARCHAR\",\"value\":\"2020-12-16 17:18:33\"}]");
+        taskExecutionContext.setExecutorId(1);
+        taskExecutionContext.setCmdTypeIfComplement(5);
+        taskExecutionContext.setTenantCode("roo");
+        taskExecutionContext.setScheduleTime(new Date());
+        taskExecutionContext.setQueue("default");
+        taskExecutionContext.setTaskParams(
+                "{\"rawScript\":\"#!/bin/sh\\necho $[yyyy-MM-dd HH:mm:ss +3]\\necho \\\" ?? ${time1} \\\"\\necho \\\" ????? ${time2}\\\"\\n\",\"localParams\":"
+                        +
+                        "[{\"prop\":\"time1\",\"direct\":\"OUT\",\"type\":\"VARCHAR\",\"value\":\"$[yyyy-MM-dd HH:mm:ss]\"},{\"prop\":\"time2\",\"direct\":\"IN\",\"type\":\"VARCHAR"
+                        + "\",\"value\":\"${time_gb}\"}],\"resourceList\":[]}");
+        Map<String, String> definedParams = new HashMap<>();
+        definedParams.put("time_gb", "2020-12-16 00:00:00");
+        taskExecutionContext.setDefinedParams(definedParams);
+        ShellTask shellTask = (ShellTask) TaskManager.newTask(taskExecutionContext,taskLogger,alertClientService);
+        shellTask.setResultString("shell return");
+        String shellReturn = shellTask.getResultString();
+        shellTask.init();
+        shellTask.setResult(shellReturn);
+        Assert.assertSame(shellReturn, "shell return");
+    }
+
+    @Test
+    public void testSqlTaskReturnString() {
+        String params = "{\"user\":\"root\",\"password\":\"123456\",\"address\":\"jdbc:mysql://127.0.0.1:3306\","
+                + "\"database\":\"test\",\"jdbcUrl\":\"jdbc:mysql://127.0.0.1:3306/test\"}";
+        taskExecutionContext = new TaskExecutionContext();
+        taskExecutionContext.setTaskParams("{\"localParams\":[{\"prop\":\"ret\", \"direct\":\"OUT\", \"type\":\"VARCHAR\", \"value\":\"\"}],"
+                + "\"type\":\"POSTGRESQL\",\"datasource\":1,\"sql\":\"insert into tb_1 values('1','2')\","
+                + "\"sqlType\":1}");
+        taskExecutionContext.setExecutePath("/tmp");
+        taskExecutionContext.setTaskAppId("1");
+        taskExecutionContext.setTenantCode("root");
+        taskExecutionContext.setStartTime(new Date());
+        taskExecutionContext.setTaskTimeout(10000);
+        taskExecutionContext.setLogPath("/tmp/dx");
+
+        SQLTaskExecutionContext sqlTaskExecutionContext = new SQLTaskExecutionContext();
+        sqlTaskExecutionContext.setConnectionParams(params);
+        taskExecutionContext.setSqlTaskExecutionContext(sqlTaskExecutionContext);
+        SqlTask sqlTask = new SqlTask(taskExecutionContext, logger, null);
+        SqlParameters sqlParameters = JSONUtils.parseObject(taskExecutionContext.getTaskParams(), SqlParameters.class);
+        List<Property> properties = sqlParameters.getLocalParams();
+        sqlTask.setNonQuerySqlReturn("sql return", properties);
+    }
 }

dolphinscheduler-server/src/test/java/org/apache/dolphinscheduler/server/worker/task/shell/ShellTaskTest.java

@@ -19,12 +19,15 @@ package org.apache.dolphinscheduler.server.worker.task.shell;
 
 import static org.mockito.ArgumentMatchers.anyString;
 
+import org.apache.dolphinscheduler.common.utils.ParameterUtils;
 import org.apache.dolphinscheduler.server.entity.TaskExecutionContext;
 import org.apache.dolphinscheduler.server.worker.task.CommandExecuteResult;
 import org.apache.dolphinscheduler.server.worker.task.ShellCommandExecutor;
+import org.apache.dolphinscheduler.service.bean.SpringApplicationContext;
 
 import java.nio.file.Files;
 import java.nio.file.Paths;
+import java.sql.DriverManager;
 import java.util.Date;
 import java.util.HashMap;
 import java.util.Map;
@@ -42,7 +45,7 @@ import org.slf4j.LoggerFactory;
  * shell task test.
  */
 @RunWith(PowerMockRunner.class)
-@PrepareForTest({ShellTask.class})
+@PrepareForTest(value = {ShellTask.class, DriverManager.class, SpringApplicationContext.class, ParameterUtils.class})
 public class ShellTaskTest {
 
     private static final Logger logger = LoggerFactory.getLogger(ShellTaskTest.class);
@@ -57,6 +60,7 @@ public class ShellTaskTest {
         System.setProperty("log4j2.disable.jmx", Boolean.TRUE.toString());
         shellCommandExecutor = PowerMockito.mock(ShellCommandExecutor.class);
         PowerMockito.whenNew(ShellCommandExecutor.class).withAnyArguments().thenReturn(shellCommandExecutor);
+        shellCommandExecutor.setTaskResultString("shellReturn");
         taskExecutionContext = new TaskExecutionContext();
         taskExecutionContext.setTaskInstanceId(1);
         taskExecutionContext.setTaskName("kris test");
@@ -68,7 +72,7 @@ public class ShellTaskTest {
             "{\"conditionResult\":\"{\\\"successNode\\\":[\\\"\\\"],\\\"failedNode\\\":[\\\"\\\"]}\",\"conditionsTask\":false,\"depList\":[],\"dependence\":\"{}\",\"forbidden\":false,\"id\":\""
                 +
                 "tasks-16849\",\"maxRetryTimes\":0,\"name\":\"shell test 001\",\"params\":\"{\\\"rawScript\\\":\\\"#!/bin/sh\\\\necho $[yyyy-MM-dd HH:mm:ss +3]\\\\necho \\\\\\\" ?? "
-                + "${time1} \\\\\\\"\\\\necho \\\\\\\" ????? ${time2}\\\\\\\"\\\\n\\\",\\\"localParams\\\":[{\\\"prop\\\":\\\"time1\\\",\\\"direct\\\":\\\"IN\\\",\\\"type\\\":"
+                + "${time1} \\\\\\\"\\\\necho \\\\\\\" ????? ${time2}\\\\\\\"\\\\n\\\",\\\"localParams\\\":[{\\\"prop\\\":\\\"time1\\\",\\\"direct\\\":\\\"OUT\\\",\\\"type\\\":"
                 + "\\\"VARCHAR\\\",\\\"value\\\":\\\"$[yyyy-MM-dd HH:mm:ss]\\\"},{\\\"prop\\\":\\\"time2\\\",\\\"direct\\\":\\\"IN\\\",\\\"type\\\":\\\"VARCHAR\\\",\\\"value\\\":\\\"${time_gb}\\\"}"
                 + "],\\\"resourceList\\\":[]}\",\"preTasks\":\"[]\",\"retryInterval\":1,\"runFlag\":\"NORMAL\",\"taskInstancePriority\":\"MEDIUM\",\"taskTimeoutParameter\":"
                 + "{\"enable\":false,\"interval\":0},\"timeout\":\"{\\\"enable\\\":false,\\\"strategy\\\":\\\"\\\"}\",\"type\":\"SHELL\",\"workerGroup\":\"default\"}");
@@ -82,7 +86,7 @@ public class ShellTaskTest {
         taskExecutionContext.setTaskParams(
             "{\"rawScript\":\"#!/bin/sh\\necho $[yyyy-MM-dd HH:mm:ss +3]\\necho \\\" ?? ${time1} \\\"\\necho \\\" ????? ${time2}\\\"\\n\",\"localParams\":"
                 +
-                "[{\"prop\":\"time1\",\"direct\":\"IN\",\"type\":\"VARCHAR\",\"value\":\"$[yyyy-MM-dd HH:mm:ss]\"},{\"prop\":\"time2\",\"direct\":\"IN\",\"type\":\"VARCHAR"
+                "[{\"prop\":\"time1\",\"direct\":\"OUT\",\"type\":\"VARCHAR\",\"value\":\"$[yyyy-MM-dd HH:mm:ss]\"},{\"prop\":\"time2\",\"direct\":\"IN\",\"type\":\"VARCHAR"
                 + "\",\"value\":\"${time_gb}\"}],\"resourceList\":[]}");
         Map<String, String> definedParams = new HashMap<>();
         definedParams.put("time_gb", "2020-12-16 00:00:00");
@@ -111,4 +115,13 @@ public class ShellTaskTest {
         PowerMockito.when(shellCommandExecutor.run(anyString())).thenReturn(commandExecuteResult);
         shellTask.handle();
     }
+
+    @Test
+    public void testSetResult() {
+        shellTask = new ShellTask(taskExecutionContext, logger);
+        shellTask.init();
+        String r = "return";
+        shellTask.setResult(r);
+    }
+
 }

dolphinscheduler-server/src/test/java/org/apache/dolphinscheduler/server/worker/task/sql/SqlTaskTest.java

@@ -19,10 +19,12 @@ package org.apache.dolphinscheduler.server.worker.task.sql;
 
 import org.apache.dolphinscheduler.common.Constants;
 import org.apache.dolphinscheduler.common.utils.ParameterUtils;
+import org.apache.dolphinscheduler.dao.AlertDao;
 import org.apache.dolphinscheduler.server.entity.SQLTaskExecutionContext;
 import org.apache.dolphinscheduler.server.entity.TaskExecutionContext;
 import org.apache.dolphinscheduler.server.worker.task.TaskProps;
 import org.apache.dolphinscheduler.service.alert.AlertClientService;
+import org.apache.dolphinscheduler.service.bean.SpringApplicationContext;
 
 import java.sql.Connection;
 import java.sql.DriverManager;
@@ -44,7 +46,7 @@ import org.slf4j.LoggerFactory;
  *  sql task test
  */
 @RunWith(PowerMockRunner.class)
-@PrepareForTest(value = {SqlTask.class, DriverManager.class})
+@PrepareForTest(value = {SqlTask.class, DriverManager.class, SpringApplicationContext.class, ParameterUtils.class})
 public class SqlTaskTest {
 
     private static final Logger logger = LoggerFactory.getLogger(SqlTaskTest.class);
@@ -70,7 +72,9 @@ public class SqlTaskTest {
         props.setTaskStartTime(new Date());
         props.setTaskTimeout(0);
         props.setTaskParams(
-                "{\"localParams\":[],\"type\":\"POSTGRESQL\",\"datasource\":1,\"sql\":\"insert into tb_1 values('1','2')\",\"sqlType\":1}");
+                "{\"localParams\":[{\"prop\":\"ret\", \"direct\":\"OUT\", \"type\":\"VARCHAR\", \"value\":\"\"}],"
+                        + "\"type\":\"POSTGRESQL\",\"datasource\":1,\"sql\":\"insert into tb_1 values('1','2')\","
+                        + "\"sqlType\":1}");
 
         taskExecutionContext = PowerMockito.mock(TaskExecutionContext.class);
         PowerMockito.when(taskExecutionContext.getTaskParams()).thenReturn(props.getTaskParams());
@@ -85,6 +89,8 @@ public class SqlTaskTest {
         sqlTaskExecutionContext.setConnectionParams(CONNECTION_PARAMS);
         PowerMockito.when(taskExecutionContext.getSqlTaskExecutionContext()).thenReturn(sqlTaskExecutionContext);
 
+        PowerMockito.mockStatic(SpringApplicationContext.class);
+        PowerMockito.when(SpringApplicationContext.getBean(Mockito.any())).thenReturn(new AlertDao());
         alertClientService = PowerMockito.mock(AlertClientService.class);
         sqlTask = new SqlTask(taskExecutionContext, logger, alertClientService);
         sqlTask.init();
@@ -95,7 +101,7 @@ public class SqlTaskTest {
         Assert.assertNotNull(sqlTask.getParameters());
     }
 
-    @Test(expected = Exception.class)
+    @Test
     public void testHandle() throws Exception {
         Connection connection = PowerMockito.mock(Connection.class);
         PowerMockito.mockStatic(DriverManager.class);

dolphinscheduler-service/src/main/java/org/apache/dolphinscheduler/service/process/ProcessService.java

@@ -24,6 +24,7 @@ import static org.apache.dolphinscheduler.common.Constants.CMD_PARAM_RECOVER_PRO
 import static org.apache.dolphinscheduler.common.Constants.CMD_PARAM_SUB_PROCESS;
 import static org.apache.dolphinscheduler.common.Constants.CMD_PARAM_SUB_PROCESS_DEFINE_ID;
 import static org.apache.dolphinscheduler.common.Constants.CMD_PARAM_SUB_PROCESS_PARENT_INSTANCE_ID;
+import static org.apache.dolphinscheduler.common.Constants.LOCAL_PARAMS;
 import static org.apache.dolphinscheduler.common.Constants.YYYY_MM_DD_HH_MM_SS;
 
 import static java.util.stream.Collectors.toSet;
@@ -32,6 +33,7 @@ import org.apache.dolphinscheduler.common.Constants;
 import org.apache.dolphinscheduler.common.enums.AuthorizationType;
 import org.apache.dolphinscheduler.common.enums.CommandType;
 import org.apache.dolphinscheduler.common.enums.CycleEnum;
+import org.apache.dolphinscheduler.common.enums.Direct;
 import org.apache.dolphinscheduler.common.enums.ExecutionStatus;
 import org.apache.dolphinscheduler.common.enums.FailureStrategy;
 import org.apache.dolphinscheduler.common.enums.Flag;
@@ -89,6 +91,7 @@ import java.util.Date;
 import java.util.EnumMap;
 import java.util.HashMap;
 import java.util.HashSet;
+import java.util.Iterator;
 import java.util.List;
 import java.util.Map;
 import java.util.Objects;
@@ -103,6 +106,8 @@ import org.springframework.stereotype.Component;
 import org.springframework.transaction.annotation.Transactional;
 
 import com.cronutils.model.Cron;
+import com.fasterxml.jackson.databind.JsonNode;
+import com.fasterxml.jackson.databind.node.ArrayNode;
 import com.fasterxml.jackson.databind.node.ObjectNode;
 
 /**
@@ -425,7 +430,7 @@ public class ProcessService {
      * recursive query sub process definition id by parent id.
      *
      * @param parentId parentId
-     * @param ids ids
+     * @param ids      ids
      */
     public void recurseFindSubProcessId(int parentId, List<Integer> ids) {
         ProcessDefinition processDefinition = processDefineMapper.selectById(parentId);
@@ -456,7 +461,7 @@ public class ProcessService {
      * create recovery waiting thread  command and delete origin command at the same time.
      * if the recovery command is exists, only update the field update_time
      *
-     * @param originCommand originCommand
+     * @param originCommand   originCommand
      * @param processInstance processInstance
      */
     public void createRecoveryWaitingThreadCommand(Command originCommand, ProcessInstance processInstance) {
@@ -508,7 +513,7 @@ public class ProcessService {
     /**
      * get schedule time from command
      *
-     * @param command command
+     * @param command  command
      * @param cmdParam cmdParam map
      * @return date
      */
@@ -524,8 +529,8 @@ public class ProcessService {
      * generate a new work process instance from command.
      *
      * @param processDefinition processDefinition
-     * @param command command
-     * @param cmdParam cmdParam map
+     * @param command           command
+     * @param cmdParam          cmdParam map
      * @return process instance
      */
     private ProcessInstance generateNewProcessInstance(ProcessDefinition processDefinition,
@@ -601,7 +606,7 @@ public class ProcessService {
      * use definition creator's tenant.
      *
      * @param tenantId tenantId
-     * @param userId userId
+     * @param userId   userId
      * @return tenant
      */
     public Tenant getTenantForProcess(int tenantId, int userId) {
@@ -624,7 +629,7 @@ public class ProcessService {
     /**
      * check command parameters is valid
      *
-     * @param command command
+     * @param command  command
      * @param cmdParam cmdParam map
      * @return whether command param is valid
      */
@@ -644,7 +649,7 @@ public class ProcessService {
      * construct process instance according to one command.
      *
      * @param command command
-     * @param host host
+     * @param host    host
      * @return process instance
      */
     private ProcessInstance constructProcessInstance(Command command, String host) {
@@ -686,11 +691,6 @@ public class ProcessService {
             } else {
                 processInstance = this.findProcessInstanceDetailById(processInstanceId);
                 // Recalculate global parameters after rerun.
-                processInstance.setGlobalParams(ParameterUtils.curingGlobalParams(
-                    processDefinition.getGlobalParamMap(),
-                    processDefinition.getGlobalParamList(),
-                    getCommandTypeIfComplement(processInstance, command),
-                    processInstance.getScheduleTime()));
             }
             processDefinition = processDefineMapper.selectById(processInstance.getProcessDefinitionId());
             processInstance.setProcessDefinition(processDefinition);
@@ -807,7 +807,7 @@ public class ProcessService {
      * return complement data if the process start with complement data
      *
      * @param processInstance processInstance
-     * @param command command
+     * @param command         command
      * @return command type
      */
     private CommandType getCommandTypeIfComplement(ProcessInstance processInstance, Command command) {
@@ -822,8 +822,8 @@ public class ProcessService {
      * initialize complement data parameters
      *
      * @param processDefinition processDefinition
-     * @param processInstance processInstance
-     * @param cmdParam cmdParam
+     * @param processInstance   processInstance
+     * @param cmdParam          cmdParam
      */
     private void initComplementDataParam(ProcessDefinition processDefinition,
                                          ProcessInstance processInstance,
@@ -895,7 +895,7 @@ public class ProcessService {
      * only the keys doesn't in sub process global would be joined.
      *
      * @param parentGlobalParams parentGlobalParams
-     * @param subGlobalParams subGlobalParams
+     * @param subGlobalParams    subGlobalParams
      * @return global params join
      */
     private String joinGlobalParams(String parentGlobalParams, String subGlobalParams) {
@@ -965,7 +965,7 @@ public class ProcessService {
      * set map {parent instance id, task instance id, 0(child instance id)}
      *
      * @param parentInstance parentInstance
-     * @param parentTask parentTask
+     * @param parentTask     parentTask
      * @return process instance map
      */
     private ProcessInstanceMap setProcessInstanceMap(ProcessInstance parentInstance, TaskInstance parentTask) {
@@ -994,7 +994,7 @@ public class ProcessService {
      * find previous task work process map.
      *
      * @param parentProcessInstance parentProcessInstance
-     * @param parentTask parentTask
+     * @param parentTask            parentTask
      * @return process instance map
      */
     private ProcessInstanceMap findPreviousTaskProcessMap(ProcessInstance parentProcessInstance,
@@ -1020,7 +1020,7 @@ public class ProcessService {
      * create sub work process command
      *
      * @param parentProcessInstance parentProcessInstance
-     * @param task task
+     * @param task                  task
      */
     public void createSubWorkProcess(ProcessInstance parentProcessInstance, TaskInstance task) {
         if (!task.isSubProcess()) {
@@ -1119,7 +1119,7 @@ public class ProcessService {
      * update sub process definition
      *
      * @param parentProcessInstance parentProcessInstance
-     * @param childDefinitionId childDefinitionId
+     * @param childDefinitionId     childDefinitionId
      */
     private void updateSubProcessDefinitionByParent(ProcessInstance parentProcessInstance, int childDefinitionId) {
         ProcessDefinition fatherDefinition = this.findProcessDefineById(parentProcessInstance.getProcessDefinitionId());
@@ -1133,7 +1133,7 @@ public class ProcessService {
     /**
      * submit task to mysql
      *
-     * @param taskInstance taskInstance
+     * @param taskInstance    taskInstance
      * @param processInstance processInstance
      * @return task instance
      */
@@ -1187,16 +1187,16 @@ public class ProcessService {
      * return stop if work process state is ready stop
      * if all of above are not satisfied, return submit success
      *
-     * @param taskInstance taskInstance
+     * @param taskInstance         taskInstance
      * @param processInstanceState processInstanceState
      * @return process instance state
      */
     public ExecutionStatus getSubmitTaskState(TaskInstance taskInstance, ExecutionStatus processInstanceState) {
         ExecutionStatus state = taskInstance.getState();
+        // running, delayed or killed
+        // the task already exists in task queue
+        // return state
         if (
-                // running, delayed or killed
-                // the task already exists in task queue
-                // return state
                 state == ExecutionStatus.RUNNING_EXECUTION
                         || state == ExecutionStatus.DELAY_EXECUTION
                         || state == ExecutionStatus.KILL
@@ -1363,7 +1363,7 @@ public class ProcessService {
      * get id list by task state
      *
      * @param instanceId instanceId
-     * @param state state
+     * @param state      state
      * @return task instance states
      */
     public List<Integer> findTaskIdByInstanceState(int instanceId, ExecutionStatus state) {
@@ -1418,7 +1418,7 @@ public class ProcessService {
      * find work process map by parent process id and parent task id.
      *
      * @param parentWorkProcessId parentWorkProcessId
-     * @param parentTaskId parentTaskId
+     * @param parentTaskId        parentTaskId
      * @return process instance map
      */
     public ProcessInstanceMap findWorkProcessMapByParent(Integer parentWorkProcessId, Integer parentTaskId) {
@@ -1440,7 +1440,7 @@ public class ProcessService {
      * find sub process instance
      *
      * @param parentProcessId parentProcessId
-     * @param parentTaskId parentTaskId
+     * @param parentTaskId    parentTaskId
      * @return process instance
      */
     public ProcessInstance findSubProcessInstance(Integer parentProcessId, Integer parentTaskId) {
@@ -1472,12 +1472,12 @@ public class ProcessService {
     /**
      * change task state
      *
-     * @param state state
-     * @param startTime startTime
-     * @param host host
+     * @param state       state
+     * @param startTime   startTime
+     * @param host        host
      * @param executePath executePath
-     * @param logPath logPath
-     * @param taskInstId taskInstId
+     * @param logPath     logPath
+     * @param taskInstId  taskInstId
      */
     public void changeTaskState(TaskInstance taskInstance, ExecutionStatus state, Date startTime, String host,
                                 String executePath,
@@ -1505,12 +1505,12 @@ public class ProcessService {
      * update the process instance
      *
      * @param processInstanceId processInstanceId
-     * @param processJson processJson
-     * @param globalParams globalParams
-     * @param scheduleTime scheduleTime
-     * @param flag flag
-     * @param locations locations
-     * @param connects connects
+     * @param processJson       processJson
+     * @param globalParams      globalParams
+     * @param scheduleTime      scheduleTime
+     * @param flag              flag
+     * @param locations         locations
+     * @param connects          connects
      * @return update process instance result
      */
     public int updateProcessInstance(Integer processInstanceId, String processJson,
@@ -1531,25 +1531,85 @@ public class ProcessService {
     /**
      * change task state
      *
-     * @param state state
-     * @param endTime endTime
+     * @param state      state
+     * @param endTime    endTime
      * @param taskInstId taskInstId
-     * @param varPool varPool
+     * @param varPool    varPool
      */
     public void changeTaskState(TaskInstance taskInstance, ExecutionStatus state,
                                 Date endTime,
                                 int processId,
                                 String appIds,
                                 int taskInstId,
-                                String varPool) {
+                                String varPool,
+                                String result) {
         taskInstance.setPid(processId);
         taskInstance.setAppLink(appIds);
         taskInstance.setState(state);
         taskInstance.setEndTime(endTime);
         taskInstance.setVarPool(varPool);
+        changeOutParam(result, taskInstance);
         saveTaskInstance(taskInstance);
     }
 
+    public void changeOutParam(String result, TaskInstance taskInstance) {
+        if (StringUtils.isEmpty(result)) {
+            return;
+        }
+        List<Map<String, String>> workerResultParam = getListMapByString(result);
+        if (CollectionUtils.isEmpty(workerResultParam)) {
+            return;
+        }
+        //if the result more than one line,just get the first .
+        Map<String, String> row = workerResultParam.get(0);
+        if (row == null || row.size() == 0) {
+            return;
+        }
+        TaskNode taskNode = JSONUtils.parseObject(taskInstance.getTaskJson(), TaskNode.class);
+        Map<String, Object> taskParams = JSONUtils.toMap(taskNode.getParams(), String.class, Object.class);
+        Object localParams = taskParams.get(LOCAL_PARAMS);
+        if (localParams == null) {
+            return;
+        }
+        ProcessInstance processInstance = this.processInstanceMapper.queryDetailById(taskInstance.getProcessInstanceId());
+        List<Property> params4Property = JSONUtils.toList(processInstance.getGlobalParams(), Property.class);
+        Map<String, Property> allParamMap = params4Property.stream().collect(Collectors.toMap(Property::getProp, Property -> Property));
+
+        List<Property> allParam = JSONUtils.toList(JSONUtils.toJsonString(localParams), Property.class);
+        for (Property info : allParam) {
+            if (info.getDirect() == Direct.OUT) {
+                String paramName = info.getProp();
+                Property property = allParamMap.get(paramName);
+                if (property == null) {
+                    continue;
+                }
+                String value = row.get(paramName);
+                if (StringUtils.isNotEmpty(value)) {
+                    property.setValue(value);
+                    info.setValue(value);
+                }
+            }
+        }
+        taskParams.put(LOCAL_PARAMS, allParam);
+        taskNode.setParams(JSONUtils.toJsonString(taskParams));
+        // task instance node json
+        taskInstance.setTaskJson(JSONUtils.toJsonString(taskNode));
+        String params4ProcessString = JSONUtils.toJsonString(params4Property);
+        int updateCount = this.processInstanceMapper.updateGlobalParamsById(params4ProcessString, processInstance.getId());
+        logger.info("updateCount:{}, params4Process:{}, processInstanceId:{}", updateCount, params4ProcessString, processInstance.getId());
+    }
+
+    public List<Map<String, String>> getListMapByString(String json) {
+        List<Map<String, String>> allParams = new ArrayList<>();
+        ArrayNode paramsByJson = JSONUtils.parseArray(json);
+        Iterator<JsonNode> listIterator = paramsByJson.iterator();
+        while (listIterator.hasNext()) {
+            Map<String, String> param = JSONUtils.toMap(listIterator.next().toString(), String.class, String.class);
+            allParams.add(param);
+        }
+        return allParams;
+    }
+
     /**
      * convert integer list to string list
      *
@@ -1642,7 +1702,7 @@ public class ProcessService {
      * update process instance state by id
      *
      * @param processInstanceId processInstanceId
-     * @param executionStatus executionStatus
+     * @param executionStatus   executionStatus
      * @return update process result
      */
     public int updateProcessInstanceState(Integer processInstanceId, ExecutionStatus executionStatus) {
@@ -1679,7 +1739,7 @@ public class ProcessService {
     /**
      * find tenant code by resource name
      *
-     * @param resName resource name
+     * @param resName      resource name
      * @param resourceType resource type
      * @return tenant code
      */
@@ -1703,9 +1763,9 @@ public class ProcessService {
     /**
      * get dependency cycle by work process define id and scheduler fire time
      *
-     * @param masterId masterId
+     * @param masterId            masterId
      * @param processDefinitionId processDefinitionId
-     * @param scheduledFireTime the time the task schedule is expected to trigger
+     * @param scheduledFireTime   the time the task schedule is expected to trigger
      * @return CycleDependency
      * @throws Exception if error throws Exception
      */
@@ -1718,8 +1778,8 @@ public class ProcessService {
     /**
      * get dependency cycle list by work process define id list and scheduler fire time
      *
-     * @param masterId masterId
-     * @param ids ids
+     * @param masterId          masterId
+     * @param ids               ids
      * @param scheduledFireTime the time the task schedule is expected to trigger
      * @return CycleDependency list
      * @throws Exception if error throws Exception
@@ -1814,8 +1874,8 @@ public class ProcessService {
      * find last running process instance
      *
      * @param definitionId process definition id
-     * @param startTime start time
-     * @param endTime end time
+     * @param startTime    start time
+     * @param endTime      end time
      * @return process instance
      */
     public ProcessInstance findLastRunningProcess(int definitionId, Date startTime, Date endTime) {
@@ -1915,7 +1975,7 @@ public class ProcessService {
     /**
      * list unauthorized udf function
      *
-     * @param userId user id
+     * @param userId     user id
      * @param needChecks data source id array
      * @return unauthorized udf function list
      */

dolphinscheduler-service/src/test/java/org/apache/dolphinscheduler/service/process/ProcessServiceTest.java

@@ -442,4 +442,26 @@ public class ProcessServiceTest {
         Assert.assertEquals(expect, processService.changeJson(newProcessData,oldJson));
 
     }
+
+    @Test
+    public void testChangeOutParam() {
+        String result = "[{\"d\":\"20210203\"}]";
+        TaskInstance taskInstance = new TaskInstance();
+        taskInstance.setProcessInstanceId(62);
+        taskInstance.setTaskJson("{\"id\":\"tasks-86175\",\"name\":\"wew\",\"desc\":null,\"type\":\"SHELL\",\"runFlag\":\"NORMAL\",\"loc\":null,\"maxRetryTimes\":0,"
+                + "\"retryInterval\":1,\"params\":{\"rawScript\":\"echo 20210203\",\"localParams\":[{\"prop\":\"d\",\"direct\":\"OUT\",\"type\":\"VARCHAR\",\"value\":\"\"}],"
+                + "\"resourceList\":[]},\"preTasks\":[],\"extras\":null,\"depList\":[],\"dependence\":{},\"conditionResult\":{\"successNode\":[\"\"],\"failedNode\":[\"\"]},"
+                + "\"taskInstancePriority\":\"MEDIUM\",\"workerGroup\":\"default\",\"workerGroupId\":null,"
+                + "\"timeout\":{\"strategy\":\"\",\"interval\":null,\"enable\":false},\"delayTime\":0}");
+        ProcessInstance processInstance = new ProcessInstance();
+        processInstance.setId(62);
+        processInstance.setGlobalParams("[{\"prop\":\"sql2\",\"direct\":null,\"type\":null,\"value\":\"\"},{\"prop\":\"out\",\"direct\":null,\"type\":null,\"value\":\"\"},"
+                + "{\"prop\":\"d\",\"direct\":\"IN\",\"type\":\"VARCHAR\",\"value\":\"\"}]");
+        String params4ProcessString = "[{\"prop\":\"sql2\",\"direct\":null,\"type\":null,\"value\":\"\"},{\"prop\":\"out\",\"direct\":null,\"type\":null,\"value\":\"\"},"
+                + "{\"prop\":\"d\",\"direct\":\"IN\",\"type\":\"VARCHAR\",\"value\":\"20210203\"}]";
+        Mockito.when(processInstanceMapper.queryDetailById(taskInstance.getProcessInstanceId())).thenReturn(processInstance);
+        Mockito.when(this.processInstanceMapper.updateGlobalParamsById(params4ProcessString, processInstance.getId())).thenReturn(1);
+        processService.changeOutParam(result,taskInstance);
+    }
+
 }

dolphinscheduler-ui/src/js/conf/home/pages/dag/_source/dag.vue

@@ -178,7 +178,7 @@
         :title="$t('Set the DAG diagram name')"
         :visible.sync="dialogVisible"
         width="auto">
-        <m-udp @onUdp="onUdpDialog" @close="closeDialog"></m-udp>
+        <m-udp ref="mUdp" @onUdp="onUdpDialog" @close="closeDialog"></m-udp>
       </el-dialog>
       <el-dialog
         :title="$t('Please set the parameters before starting')"
@@ -268,7 +268,7 @@
     },
     methods: {
       ...mapActions('dag', ['saveDAGchart', 'updateInstance', 'updateDefinition', 'getTaskState', 'switchProcessDefinitionVersion', 'getProcessDefinitionVersionsPage', 'deleteProcessDefinitionVersion']),
-      ...mapMutations('dag', ['addTasks', 'cacheTasks', 'resetParams', 'setIsEditDag', 'setName', 'addConnects']),
+      ...mapMutations('dag', ['addTasks', 'cacheTasks', 'resetParams', 'setIsEditDag', 'setName', 'addConnects', 'resetLocalParam']),
       startRunning (item, startNodeList, sourceType) {
         this.startData = item
         this.startNodeList = startNodeList
@@ -377,7 +377,7 @@
 
             // remove tip state dom
             $('.w').find('.state-p').html('')
-
+            const newTask = []
             data.forEach(v1 => {
               idArr.forEach(v2 => {
                 if (v2.name === v1.name) {
@@ -387,6 +387,12 @@
                   taskList.forEach(item => {
                     if (item.name === v1.name) {
                       depState = item.state
+                      const params = item.taskJson ? JSON.parse(item.taskJson).params : ''
+                      let localParam = params.localParams || []
+                      newTask.push({
+                        id: v2.id,
+                        localParam
+                      })
                     }
                   })
                   dom.attr('data-state-id', v1.stateId)
@@ -403,6 +409,9 @@
                 findComponentDownward(this.$root, `${this.type}-details`)._reset()
               }
             }
+            if (!isReset) {
+              this.resetLocalParam(newTask)
+            }
             resolve()
           })
         })
@@ -550,7 +559,11 @@
           this.$message.warning(`${i18n.$t('Failed to create node to save')}`)
           return
         }
+
         this.dialogVisible = true
+        this.$nextTick(() => {
+          this.$refs.mUdp.reloadParam()
+        })
       },
       /**
        * Return to the previous child node

dolphinscheduler-ui/src/js/conf/home/pages/dag/_source/formModel/tasks/_source/localParams.vue

@@ -21,7 +21,7 @@
          :key="item.id"
          @click="_getIndex($index)">
       <el-input
-              :disabled="isDetails"
+              :disabled="isDetails || item.ifFixed"
               type="text"
               size="small"
               v-model="localParamsList[$index].prop"
@@ -68,7 +68,7 @@
               @blur="_handleValue()"
               :style="inputStyle">
       </el-input>
-      <span class="lt-add">
+      <span class="lt-add" v-show="!item.ifFixed">
         <a href="javascript:" style="color:red;" @click="!isDetails && _removeUdp($index)" >
           <em class="el-icon-delete" :class="_isDetails" data-toggle="tooltip" :title="$t('delete')" ></em>
         </a>

dolphinscheduler-ui/src/js/conf/home/pages/dag/_source/formModel/tasks/shell.vue

@@ -46,7 +46,7 @@
                 ref="refLocalParams"
                 @on-local-params="_onLocalParams"
                 :udp-list="localParams"
-                :hide="false">
+                :hide="true">
         </m-local-params>
       </div>
     </m-list-box>

dolphinscheduler-ui/src/js/conf/home/pages/dag/_source/udp/udp.vue

@@ -142,7 +142,12 @@
         return true
       },
       _accuStore () {
-        this.store.commit('dag/setGlobalParams', _.cloneDeep(this.udpList))
+        const udp = _.cloneDeep(this.udpList)
+        udp.forEach(u => {
+          delete u.ifFixed
+        })
+        this.store.commit('dag/setGlobalParams', udp)
+
         this.store.commit('dag/setName', _.cloneDeep(this.name))
         this.store.commit('dag/setTimeout', _.cloneDeep(this.timeout))
         this.store.commit('dag/setTenantId', _.cloneDeep(this.tenantId))
@@ -191,6 +196,46 @@
        */
       close () {
         this.$emit('close')
+      },
+      /**
+       * reload localParam
+       */
+      reloadParam () {
+        const dag = _.cloneDeep(this.store.state.dag)
+        let fixedParam = []
+        const tasks = this.store.state.dag.tasks
+        for (const task of tasks) {
+          const localParam = task.params ? task.params.localParams : []
+          localParam.forEach(l => {
+            if (!fixedParam.some(f => { return f.prop === l.prop })) {
+              fixedParam.push(Object.assign({
+                ifFixed: true
+              }, l))
+            }
+          })
+        }
+
+        let globalParams = _.cloneDeep(dag.globalParams)
+
+        globalParams = globalParams.map(g => {
+          if (fixedParam.some(f => { return g.prop === f.prop })) {
+            fixedParam = fixedParam.filter(f => { return g.prop !== f.prop })
+            return Object.assign(g, {
+              ifFixed: true
+            })
+          } else {
+            return g
+          }
+        })
+        let udpList = [...fixedParam, ...globalParams].sort(s => {
+          if (s.ifFixed) {
+            return -1
+          } else {
+            return 1
+          }
+        })
+        this.udpList = udpList
+        this.udpListCache = udpList
       }
     },
     watch: {
@@ -203,8 +248,7 @@
     },
     created () {
       const dag = _.cloneDeep(this.store.state.dag)
-      this.udpList = dag.globalParams
-      this.udpListCache = dag.globalParams
+
       this.name = dag.name
       this.originalName = dag.name
       this.description = dag.description

dolphinscheduler-ui/src/js/conf/home/store/dag/mutations.js

@@ -162,5 +162,16 @@ export default {
     } else {
       state.cacheTasks[payload.id] = payload
     }
+  },
+  resetLocalParam (state, payload) {
+    const tasks = state.tasks
+    tasks.forEach((task, index) => {
+      payload.forEach(p => {
+        if (p.id === task.id) {
+          tasks[index].params.localParams = p.localParam
+        }
+      })
+    })
+    state.tasks = tasks
   }
 }

pom.xml

@@ -925,6 +925,8 @@
                         <include>**/server/worker/task/sqoop/SqoopTaskTest.java</include>
                         <include>**/server/worker/task/shell/ShellTaskTest.java</include>
                         <include>**/server/worker/task/TaskManagerTest.java</include>
+                        <include>**/server/worker/task/AbstractCommandExecutorTest.java</include>
+                        <include>**/server/worker/task/ShellTaskReturnTest.java</include>
                         <include>**/server/worker/EnvFileTest.java</include>
                         <include>**/server/worker/runner/TaskExecuteThreadTest.java</include>
                         <include>**/server/worker/runner/WorkerManagerThreadTest.java</include>