From db04a5b04df8aa5754741c6dd456579d743af1e9 Mon Sep 17 00:00:00 2001
From: OS <29528966+lenboo@users.noreply.github.com>
Date: Mon, 11 Oct 2021 20:22:58 +0800
Subject: [PATCH] [Feature-6471]Cache Process definition in master (#6485)

* feature-6471 Cache Process definition in master
---
 .../api/service/impl/ExecutorServiceImpl.java |  16 ++-
 .../dolphinscheduler/dao/entity/Command.java  |  41 +++++-
 .../dao/mapper/CommandMapper.java             |   7 -
 .../dao/mapper/CommandMapper.xml              |  10 --
 .../dao/mapper/CommandMapperTest.java         |   4 +-
 .../server/master/config/MasterConfig.java    |  12 ++
 .../master/runner/MasterSchedulerService.java |  28 +++-
 .../src/main/resources/master.properties      |   3 +
 .../service/process/ProcessService.java       | 133 ++++++++----------
 .../service/quartz/ProcessScheduleJob.java    |   1 +
 .../service/process/ProcessServiceTest.java   |  48 ++++---
 sql/dolphinscheduler_h2.sql                   |   4 +
 sql/dolphinscheduler_mysql.sql                |  34 +++--
 sql/dolphinscheduler_postgre.sql              |  63 +++++----
 14 files changed, 235 insertions(+), 169 deletions(-)

diff --git a/dolphinscheduler-api/src/main/java/org/apache/dolphinscheduler/api/service/impl/ExecutorServiceImpl.java b/dolphinscheduler-api/src/main/java/org/apache/dolphinscheduler/api/service/impl/ExecutorServiceImpl.java
index 5042a03714..fe5f9101b1 100644
--- a/dolphinscheduler-api/src/main/java/org/apache/dolphinscheduler/api/service/impl/ExecutorServiceImpl.java
+++ b/dolphinscheduler-api/src/main/java/org/apache/dolphinscheduler/api/service/impl/ExecutorServiceImpl.java
@@ -283,13 +283,13 @@ public class ExecutorServiceImpl extends BaseServiceImpl implements ExecutorServ
 
         switch (executeType) {
             case REPEAT_RUNNING:
-                result = insertCommand(loginUser, processInstanceId, processDefinition.getCode(), CommandType.REPEAT_RUNNING, startParams);
+                result = insertCommand(loginUser, processInstanceId, processDefinition.getCode(), processDefinition.getVersion(), CommandType.REPEAT_RUNNING, startParams);
                 break;
             case RECOVER_SUSPENDED_PROCESS:
-                result = insertCommand(loginUser, processInstanceId, processDefinition.getCode(), CommandType.RECOVER_SUSPENDED_PROCESS, startParams);
+                result = insertCommand(loginUser, processInstanceId, processDefinition.getCode(), processDefinition.getVersion(), CommandType.RECOVER_SUSPENDED_PROCESS, startParams);
                 break;
             case START_FAILURE_TASK_PROCESS:
-                result = insertCommand(loginUser, processInstanceId, processDefinition.getCode(), CommandType.START_FAILURE_TASK_PROCESS, startParams);
+                result = insertCommand(loginUser, processInstanceId, processDefinition.getCode(), processDefinition.getVersion(), CommandType.START_FAILURE_TASK_PROCESS, startParams);
                 break;
             case STOP:
                 if (processInstance.getState() == ExecutionStatus.READY_STOP) {
@@ -409,10 +409,11 @@ public class ExecutorServiceImpl extends BaseServiceImpl implements ExecutorServ
      * @param loginUser           login user
      * @param instanceId          instance id
      * @param processDefinitionCode process definition code
+     * @param version
      * @param commandType         command type
      * @return insert result code
      */
-    private Map<String, Object> insertCommand(User loginUser, Integer instanceId, long processDefinitionCode, CommandType commandType, String startParams) {
+    private Map<String, Object> insertCommand(User loginUser, Integer instanceId, long processDefinitionCode, int processVersion, CommandType commandType, String startParams) {
         Map<String, Object> result = new HashMap<>();
 
         //To add startParams only when repeat running is needed
@@ -427,6 +428,8 @@ public class ExecutorServiceImpl extends BaseServiceImpl implements ExecutorServ
         command.setProcessDefinitionCode(processDefinitionCode);
         command.setCommandParam(JSONUtils.toJsonString(cmdParam));
         command.setExecutorId(loginUser.getId());
+        command.setProcessDefinitionVersion(processVersion);
+        command.setProcessInstanceId(instanceId);
 
         if (!processService.verifyIsNeedCreateCommand(command)) {
             putMsg(result, Status.PROCESS_INSTANCE_EXECUTING_COMMAND, processDefinitionCode);
@@ -545,6 +548,11 @@ public class ExecutorServiceImpl extends BaseServiceImpl implements ExecutorServ
         command.setWorkerGroup(workerGroup);
         command.setEnvironmentCode(environmentCode);
         command.setDryRun(dryRun);
+        ProcessDefinition processDefinition = processService.findProcessDefinitionByCode(processDefineCode);
+        if (processDefinition != null) {
+            command.setProcessDefinitionVersion(processDefinition.getVersion());
+        }
+        command.setProcessInstanceId(0);
 
         Date start = null;
         Date end = null;
diff --git a/dolphinscheduler-dao/src/main/java/org/apache/dolphinscheduler/dao/entity/Command.java b/dolphinscheduler-dao/src/main/java/org/apache/dolphinscheduler/dao/entity/Command.java
index b1ed217537..ae2ff6258a 100644
--- a/dolphinscheduler-dao/src/main/java/org/apache/dolphinscheduler/dao/entity/Command.java
+++ b/dolphinscheduler-dao/src/main/java/org/apache/dolphinscheduler/dao/entity/Command.java
@@ -132,6 +132,12 @@ public class Command {
     @TableField("dry_run")
     private int dryRun;
 
+    @TableField("process_instance_id")
+    private int processInstanceId;
+
+    @TableField("process_definition_version")
+    private int processDefinitionVersion;
+
     public Command() {
         this.taskDependType = TaskDependType.TASK_POST;
         this.failureStrategy = FailureStrategy.CONTINUE;
@@ -152,7 +158,10 @@ public class Command {
             String workerGroup,
             Long environmentCode,
             Priority processInstancePriority,
-            int dryRun) {
+            int dryRun,
+            int processInstanceId,
+            int processDefinitionVersion
+    ) {
         this.commandType = commandType;
         this.executorId = executorId;
         this.processDefinitionCode = processDefinitionCode;
@@ -168,6 +177,8 @@ public class Command {
         this.environmentCode = environmentCode;
         this.processInstancePriority = processInstancePriority;
         this.dryRun = dryRun;
+        this.processInstanceId = processInstanceId;
+        this.processDefinitionVersion = processDefinitionVersion;
     }
 
     public TaskDependType getTaskDependType() {
@@ -298,6 +309,22 @@ public class Command {
         this.dryRun = dryRun;
     }
 
+    public int getProcessInstanceId() {
+        return processInstanceId;
+    }
+
+    public void setProcessInstanceId(int processInstanceId) {
+        this.processInstanceId = processInstanceId;
+    }
+
+    public int getProcessDefinitionVersion() {
+        return processDefinitionVersion;
+    }
+
+    public void setProcessDefinitionVersion(int processDefinitionVersion) {
+        this.processDefinitionVersion = processDefinitionVersion;
+    }
+
     @Override
     public boolean equals(Object o) {
         if (this == o) {
@@ -353,8 +380,13 @@ public class Command {
         if (processInstancePriority != command.processInstancePriority) {
             return false;
         }
+        if (processInstanceId != command.processInstanceId) {
+            return false;
+        }
+        if (processDefinitionVersion != command.getProcessDefinitionVersion()) {
+            return false;
+        }
         return !(updateTime != null ? !updateTime.equals(command.updateTime) : command.updateTime != null);
-
     }
 
     @Override
@@ -375,6 +407,8 @@ public class Command {
         result = 31 * result + (workerGroup != null ? workerGroup.hashCode() : 0);
         result = 31 * result + (environmentCode != null ? environmentCode.hashCode() : 0);
         result = 31 * result + dryRun;
+        result = 31 * result + processInstanceId;
+        result = 31 * result + processDefinitionVersion;
         return result;
     }
 
@@ -397,7 +431,10 @@ public class Command {
                 + ", workerGroup='" + workerGroup + '\''
                 + ", environmentCode='" + environmentCode + '\''
                 + ", dryRun='" + dryRun + '\''
+                + ", processInstanceId='" + processInstanceId + '\''
+                + ", processDefinitionVersion='" + processDefinitionVersion + '\''
                 + '}';
     }
+
 }
 
diff --git a/dolphinscheduler-dao/src/main/java/org/apache/dolphinscheduler/dao/mapper/CommandMapper.java b/dolphinscheduler-dao/src/main/java/org/apache/dolphinscheduler/dao/mapper/CommandMapper.java
index 2bbfb4b7b1..22913845c3 100644
--- a/dolphinscheduler-dao/src/main/java/org/apache/dolphinscheduler/dao/mapper/CommandMapper.java
+++ b/dolphinscheduler-dao/src/main/java/org/apache/dolphinscheduler/dao/mapper/CommandMapper.java
@@ -31,13 +31,6 @@ import java.util.List;
  */
 public interface CommandMapper extends BaseMapper<Command> {
 
-
-    /**
-     * get one command
-     * @return command
-     */
-    Command getOneToRun();
-
     /**
      * count command state
      * @param userId userId
diff --git a/dolphinscheduler-dao/src/main/resources/org/apache/dolphinscheduler/dao/mapper/CommandMapper.xml b/dolphinscheduler-dao/src/main/resources/org/apache/dolphinscheduler/dao/mapper/CommandMapper.xml
index 5b2d6b4d8c..b0ea477431 100644
--- a/dolphinscheduler-dao/src/main/resources/org/apache/dolphinscheduler/dao/mapper/CommandMapper.xml
+++ b/dolphinscheduler-dao/src/main/resources/org/apache/dolphinscheduler/dao/mapper/CommandMapper.xml
@@ -18,16 +18,6 @@
 
 <!DOCTYPE mapper PUBLIC "-//mybatis.org//DTD Mapper 3.0//EN" "http://mybatis.org/dtd/mybatis-3-mapper.dtd" >
 <mapper namespace="org.apache.dolphinscheduler.dao.mapper.CommandMapper">
-    <select id="getOneToRun" resultType="org.apache.dolphinscheduler.dao.entity.Command">
-        select cmd.id, cmd.command_type, cmd.process_definition_code, cmd.command_param, cmd.task_depend_type, cmd.failure_strategy,
-        cmd.warning_type, cmd.warning_group_id, cmd.schedule_time, cmd.start_time, cmd.executor_id, cmd.update_time,
-        cmd.process_instance_priority, cmd.worker_group, cmd.environment_code, cmd.dry_run
-        from t_ds_command cmd
-        join t_ds_process_definition definition on cmd.process_definition_code = definition.code
-        where definition.release_state = 1 AND definition.flag = 1
-        order by cmd.update_time asc
-        limit 1
-    </select>
     <select id="countCommandState" resultType="org.apache.dolphinscheduler.dao.entity.CommandCount">
         select cmd.command_type as command_type, count(1) as count
         from t_ds_command cmd, t_ds_process_definition process
diff --git a/dolphinscheduler-dao/src/test/java/org/apache/dolphinscheduler/dao/mapper/CommandMapperTest.java b/dolphinscheduler-dao/src/test/java/org/apache/dolphinscheduler/dao/mapper/CommandMapperTest.java
index dc9dbeebe7..0266ac0eb2 100644
--- a/dolphinscheduler-dao/src/test/java/org/apache/dolphinscheduler/dao/mapper/CommandMapperTest.java
+++ b/dolphinscheduler-dao/src/test/java/org/apache/dolphinscheduler/dao/mapper/CommandMapperTest.java
@@ -147,7 +147,7 @@ public class CommandMapperTest {
 
         createCommand(CommandType.START_PROCESS, processDefinition.getCode());
 
-        Command actualCommand = commandMapper.getOneToRun();
+        List<Command> actualCommand = commandMapper.queryCommandPage(1,0);
 
         assertNotNull(actualCommand);
     }
@@ -259,6 +259,8 @@ public class CommandMapperTest {
         command.setStartTime(DateUtils.stringToDate("2019-12-29 10:10:00"));
         command.setUpdateTime(DateUtils.stringToDate("2019-12-29 10:10:00"));
         command.setWorkerGroup(Constants.DEFAULT_WORKER_GROUP);
+        command.setProcessInstanceId(0);
+        command.setProcessDefinitionVersion(0);
         commandMapper.insert(command);
 
         return command;
diff --git a/dolphinscheduler-server/src/main/java/org/apache/dolphinscheduler/server/master/config/MasterConfig.java b/dolphinscheduler-server/src/main/java/org/apache/dolphinscheduler/server/master/config/MasterConfig.java
index 6c2e2a1e47..124ecebba9 100644
--- a/dolphinscheduler-server/src/main/java/org/apache/dolphinscheduler/server/master/config/MasterConfig.java
+++ b/dolphinscheduler-server/src/main/java/org/apache/dolphinscheduler/server/master/config/MasterConfig.java
@@ -60,6 +60,9 @@ public class MasterConfig {
     @Value("${master.reserved.memory:0.3}")
     private double masterReservedMemory;
 
+    @Value("${master.cache.process.definition:true}")
+    private boolean masterCacheProcessDefinition;
+
     public int getListenPort() {
         return listenPort;
     }
@@ -150,4 +153,13 @@ public class MasterConfig {
     public void setStateWheelInterval(int stateWheelInterval) {
         this.stateWheelInterval = stateWheelInterval;
     }
+
+    public boolean getMasterCacheProcessDefinition() {
+        return masterCacheProcessDefinition;
+    }
+
+    public void setMasterCacheProcessDefinition(boolean masterCacheProcessDefinition) {
+        this.masterCacheProcessDefinition = masterCacheProcessDefinition;
+    }
+
 }
\ No newline at end of file
diff --git a/dolphinscheduler-server/src/main/java/org/apache/dolphinscheduler/server/master/runner/MasterSchedulerService.java b/dolphinscheduler-server/src/main/java/org/apache/dolphinscheduler/server/master/runner/MasterSchedulerService.java
index faa4eb081c..803ba09bcc 100644
--- a/dolphinscheduler-server/src/main/java/org/apache/dolphinscheduler/server/master/runner/MasterSchedulerService.java
+++ b/dolphinscheduler-server/src/main/java/org/apache/dolphinscheduler/server/master/runner/MasterSchedulerService.java
@@ -23,6 +23,7 @@ import org.apache.dolphinscheduler.common.thread.ThreadUtils;
 import org.apache.dolphinscheduler.common.utils.NetUtils;
 import org.apache.dolphinscheduler.common.utils.OSUtils;
 import org.apache.dolphinscheduler.dao.entity.Command;
+import org.apache.dolphinscheduler.dao.entity.ProcessDefinition;
 import org.apache.dolphinscheduler.dao.entity.ProcessInstance;
 import org.apache.dolphinscheduler.dao.entity.TaskInstance;
 import org.apache.dolphinscheduler.remote.NettyRemotingClient;
@@ -34,6 +35,7 @@ import org.apache.dolphinscheduler.server.master.registry.ServerNodeManager;
 import org.apache.dolphinscheduler.service.alert.ProcessAlertManager;
 import org.apache.dolphinscheduler.service.process.ProcessService;
 
+import java.util.HashMap;
 import java.util.List;
 import java.util.concurrent.ConcurrentHashMap;
 import java.util.concurrent.ThreadPoolExecutor;
@@ -92,11 +94,26 @@ public class MasterSchedulerService extends Thread {
      */
     private ThreadPoolExecutor masterExecService;
 
-
+    /**
+     * process instance execution list
+     */
     private ConcurrentHashMap<Integer, WorkflowExecuteThread> processInstanceExecMaps;
+    /**
+     * process timeout check list
+     */
     ConcurrentHashMap<Integer, ProcessInstance> processTimeoutCheckList = new ConcurrentHashMap<>();
+
+    /**
+     * task time out checkout list
+     */
     ConcurrentHashMap<Integer, TaskInstance> taskTimeoutCheckList = new ConcurrentHashMap<>();
 
+    /**
+     * key:code-version
+     * value: processDefinition
+     */
+    HashMap<String, ProcessDefinition> processDefinitionCacheMaps = new HashMap<>();
+
     private StateWheelExecuteThread stateWheelExecuteThread;
 
     /**
@@ -112,7 +129,6 @@ public class MasterSchedulerService extends Thread {
                 taskTimeoutCheckList,
                 this.processInstanceExecMaps,
                 masterConfig.getStateWheelInterval() * Constants.SLEEP_TIME_MILLIS);
-
     }
 
     @Override
@@ -165,7 +181,6 @@ public class MasterSchedulerService extends Thread {
      */
     private void scheduleProcess() throws Exception {
 
-        int activeCount = masterExecService.getActiveCount();
         // make sure to scan and delete command  table in one transaction
         Command command = findOneCommand();
         if (command != null) {
@@ -173,7 +188,12 @@ public class MasterSchedulerService extends Thread {
             try {
                 ProcessInstance processInstance = processService.handleCommand(logger,
                         getLocalAddress(),
-                        this.masterConfig.getMasterExecThreads() - activeCount, command);
+                        command,
+                        processDefinitionCacheMaps);
+                if (!masterConfig.getMasterCacheProcessDefinition()
+                        && processDefinitionCacheMaps.size() > 0) {
+                    processDefinitionCacheMaps.clear();
+                }
                 if (processInstance != null) {
                     WorkflowExecuteThread workflowExecuteThread = new WorkflowExecuteThread(
                             processInstance
diff --git a/dolphinscheduler-server/src/main/resources/master.properties b/dolphinscheduler-server/src/main/resources/master.properties
index cc45622132..fbe8f6e2a6 100644
--- a/dolphinscheduler-server/src/main/resources/master.properties
+++ b/dolphinscheduler-server/src/main/resources/master.properties
@@ -39,6 +39,9 @@
 # master commit task interval, the unit is millisecond
 #master.task.commit.interval=1000
 
+# master cache process definition, default: true
+#master.cache.process.definition=true
+
 # master max cpuload avg, only higher than the system cpu load average, master server can schedule. default value -1: the number of cpu cores * 2
 #master.max.cpuload.avg=-1
 
diff --git a/dolphinscheduler-service/src/main/java/org/apache/dolphinscheduler/service/process/ProcessService.java b/dolphinscheduler-service/src/main/java/org/apache/dolphinscheduler/service/process/ProcessService.java
index 4ca47c8fca..8100d9fbf9 100644
--- a/dolphinscheduler-service/src/main/java/org/apache/dolphinscheduler/service/process/ProcessService.java
+++ b/dolphinscheduler-service/src/main/java/org/apache/dolphinscheduler/service/process/ProcessService.java
@@ -208,13 +208,12 @@ public class ProcessService {
      *
      * @param logger logger
      * @param host host
-     * @param validThreadNum validThreadNum
      * @param command found command
+     * @param processDefinitionCacheMaps
      * @return process instance
      */
-    @Transactional(rollbackFor = Exception.class)
-    public ProcessInstance handleCommand(Logger logger, String host, int validThreadNum, Command command) {
-        ProcessInstance processInstance = constructProcessInstance(command, host);
+    public ProcessInstance handleCommand(Logger logger, String host, Command command, HashMap<String, ProcessDefinition> processDefinitionCacheMaps) {
+        ProcessInstance processInstance = constructProcessInstance(command, host, processDefinitionCacheMaps);
         // cannot construct process instance, return null
         if (processInstance == null) {
             logger.error("scan command, command parameter is error: {}", command);
@@ -235,7 +234,6 @@ public class ProcessService {
      * @param command command
      * @param message message
      */
-    @Transactional(rollbackFor = Exception.class)
     public void moveToErrorCommand(Command command, String message) {
         ErrorCommand errorCommand = new ErrorCommand(command, message);
         this.errorCommandMapper.insert(errorCommand);
@@ -286,15 +284,6 @@ public class ProcessService {
         return result;
     }
 
-    /**
-     * find one command from queue list
-     *
-     * @return command
-     */
-    public Command findOneCommand() {
-        return commandMapper.getOneToRun();
-    }
-
     /**
      * get command page
      *
@@ -547,7 +536,9 @@ public class ProcessService {
                 processInstance.getWorkerGroup(),
                 processInstance.getEnvironmentCode(),
                 processInstance.getProcessInstancePriority(),
-                processInstance.getDryRun()
+                processInstance.getDryRun(),
+                processInstance.getId(),
+                processInstance.getProcessDefinitionVersion()
             );
             saveCommand(command);
             return;
@@ -744,90 +735,74 @@ public class ProcessService {
     /**
      * construct process instance according to one command.
      *
-     * @param command command
-     * @param host host
+     * @param command                    command
+     * @param host                       host
+     * @param processDefinitionCacheMaps
      * @return process instance
      */
-    private ProcessInstance constructProcessInstance(Command command, String host) {
+    private ProcessInstance constructProcessInstance(Command command, String host, HashMap<String, ProcessDefinition> processDefinitionCacheMaps) {
         ProcessInstance processInstance;
+        ProcessDefinition processDefinition;
         CommandType commandType = command.getCommandType();
-        Map<String, String> cmdParam = JSONUtils.toMap(command.getCommandParam());
-
-        ProcessDefinition processDefinition = getProcessDefinitionByCommand(command.getProcessDefinitionCode(), cmdParam);
+        String key = String.format("%d-%d", command.getProcessDefinitionCode(), command.getProcessDefinitionVersion());
+        if (processDefinitionCacheMaps.containsKey(key)) {
+            processDefinition = processDefinitionCacheMaps.get(key);
+        } else {
+            processDefinition = this.findProcessDefinition(command.getProcessDefinitionCode(), command.getProcessDefinitionVersion());
+            if (processDefinition != null) {
+                processDefinitionCacheMaps.put(key, processDefinition);
+            }
+        }
         if (processDefinition == null) {
             logger.error("cannot find the work process define! define code : {}", command.getProcessDefinitionCode());
             return null;
         }
-
+        Map<String, String> cmdParam = JSONUtils.toMap(command.getCommandParam());
+        int processInstanceId = command.getProcessInstanceId();
+        if (processInstanceId == 0) {
+            processInstance = generateNewProcessInstance(processDefinition, command, cmdParam);
+        } else {
+            processInstance = this.findProcessInstanceDetailById(processInstanceId);
+            if (processInstance == null) {
+                return processInstance;
+            }
+        }
         if (cmdParam != null) {
-            int processInstanceId = 0;
-            // recover from failure or pause tasks
-            if (cmdParam.containsKey(Constants.CMD_PARAM_RECOVER_PROCESS_ID_STRING)) {
-                String processId = cmdParam.get(Constants.CMD_PARAM_RECOVER_PROCESS_ID_STRING);
-                processInstanceId = Integer.parseInt(processId);
-                if (processInstanceId == 0) {
-                    logger.error("command parameter is error, [ ProcessInstanceId ] is 0");
-                    return null;
-                }
-            } else if (cmdParam.containsKey(Constants.CMD_PARAM_SUB_PROCESS)) {
-                // sub process map
-                String pId = cmdParam.get(Constants.CMD_PARAM_SUB_PROCESS);
-                processInstanceId = Integer.parseInt(pId);
-            } else if (cmdParam.containsKey(Constants.CMD_PARAM_RECOVERY_WAITING_THREAD)) {
-                // waiting thread command
-                String pId = cmdParam.get(Constants.CMD_PARAM_RECOVERY_WAITING_THREAD);
-                processInstanceId = Integer.parseInt(pId);
+            CommandType commandTypeIfComplement = getCommandTypeIfComplement(processInstance, command);
+            // reset global params while repeat running is needed by cmdParam
+            if (commandTypeIfComplement == CommandType.REPEAT_RUNNING) {
+                setGlobalParamIfCommanded(processDefinition, cmdParam);
             }
 
-            if (processInstanceId == 0) {
-                processInstance = generateNewProcessInstance(processDefinition, command, cmdParam);
-            } else {
-                processInstance = this.findProcessInstanceDetailById(processInstanceId);
-                if (processInstance == null) {
-                    return processInstance;
-                }
-                CommandType commandTypeIfComplement = getCommandTypeIfComplement(processInstance, command);
-
-                // reset global params while repeat running is needed by cmdParam
-                if (commandTypeIfComplement == CommandType.REPEAT_RUNNING) {
-                    setGlobalParamIfCommanded(processDefinition, cmdParam);
-                }
-
-                // Recalculate global parameters after rerun.
-                processInstance.setGlobalParams(ParameterUtils.curingGlobalParams(
+            // Recalculate global parameters after rerun.
+            processInstance.setGlobalParams(ParameterUtils.curingGlobalParams(
                     processDefinition.getGlobalParamMap(),
                     processDefinition.getGlobalParamList(),
                     commandTypeIfComplement,
                     processInstance.getScheduleTime()));
-                processInstance.setProcessDefinition(processDefinition);
-            }
-            //reset command parameter
-            if (processInstance.getCommandParam() != null) {
-                Map<String, String> processCmdParam = JSONUtils.toMap(processInstance.getCommandParam());
-                for (Map.Entry<String, String> entry : processCmdParam.entrySet()) {
-                    if (!cmdParam.containsKey(entry.getKey())) {
-                        cmdParam.put(entry.getKey(), entry.getValue());
-                    }
+            processInstance.setProcessDefinition(processDefinition);
+        }
+        //reset command parameter
+        if (processInstance.getCommandParam() != null) {
+            Map<String, String> processCmdParam = JSONUtils.toMap(processInstance.getCommandParam());
+            for (Map.Entry<String, String> entry : processCmdParam.entrySet()) {
+                if (!cmdParam.containsKey(entry.getKey())) {
+                    cmdParam.put(entry.getKey(), entry.getValue());
                 }
             }
-            // reset command parameter if sub process
-            if (cmdParam.containsKey(Constants.CMD_PARAM_SUB_PROCESS)) {
-                processInstance.setCommandParam(command.getCommandParam());
-            }
-        } else {
-            // generate one new process instance
-            processInstance = generateNewProcessInstance(processDefinition, command, cmdParam);
+        }
+        // reset command parameter if sub process
+        if (cmdParam.containsKey(Constants.CMD_PARAM_SUB_PROCESS)) {
+            processInstance.setCommandParam(command.getCommandParam());
         }
         if (Boolean.FALSE.equals(checkCmdParam(command, cmdParam))) {
             logger.error("command parameter check failed!");
             return null;
         }
-
         if (command.getScheduleTime() != null) {
             processInstance.setScheduleTime(command.getScheduleTime());
         }
         processInstance.setHost(host);
-
         ExecutionStatus runStatus = ExecutionStatus.RUNNING_EXECUTION;
         int runTime = processInstance.getRunTimes();
         switch (commandType) {
@@ -846,7 +821,7 @@ public class ProcessService {
                     initTaskInstance(this.findTaskInstanceById(taskId));
                 }
                 cmdParam.put(Constants.CMD_PARAM_RECOVERY_START_NODE_STRING,
-                    String.join(Constants.COMMA, convertIntListToString(failedList)));
+                        String.join(Constants.COMMA, convertIntListToString(failedList)));
                 processInstance.setCommandParam(JSONUtils.toJsonString(cmdParam));
                 processInstance.setRunTimes(runTime + 1);
                 break;
@@ -859,7 +834,7 @@ public class ProcessService {
                 cmdParam.remove(Constants.CMD_PARAM_RECOVERY_START_NODE_STRING);
                 List<Integer> suspendedNodeList = this.findTaskIdByInstanceState(processInstance.getId(), ExecutionStatus.PAUSE);
                 List<Integer> stopNodeList = findTaskIdByInstanceState(processInstance.getId(),
-                    ExecutionStatus.KILL);
+                        ExecutionStatus.KILL);
                 suspendedNodeList.addAll(stopNodeList);
                 for (Integer taskId : suspendedNodeList) {
                     // initialize the pause state
@@ -1274,7 +1249,7 @@ public class ProcessService {
             }
         }
         String processParam = getSubWorkFlowParam(instanceMap, parentProcessInstance, fatherParams);
-
+        int subProcessInstanceId = childInstance == null ? 0 : childInstance.getId();
         return new Command(
             commandType,
             TaskDependType.TASK_POST,
@@ -1288,7 +1263,9 @@ public class ProcessService {
             task.getWorkerGroup(),
             task.getEnvironmentCode(),
             parentProcessInstance.getProcessInstancePriority(),
-            parentProcessInstance.getDryRun()
+            parentProcessInstance.getDryRun(),
+            subProcessInstanceId,
+            parentProcessInstance.getProcessDefinitionVersion()
         );
     }
 
@@ -1886,6 +1863,8 @@ public class ProcessService {
         //2 insert into recover command
         Command cmd = new Command();
         cmd.setProcessDefinitionCode(processDefinition.getCode());
+        cmd.setProcessDefinitionVersion(processDefinition.getVersion());
+        cmd.setProcessInstanceId(processInstance.getId());
         cmd.setCommandParam(String.format("{\"%s\":%d}", Constants.CMD_PARAM_RECOVER_PROCESS_ID_STRING, processInstance.getId()));
         cmd.setExecutorId(processInstance.getExecutorId());
         cmd.setCommandType(CommandType.RECOVER_TOLERANCE_FAULT_PROCESS);
diff --git a/dolphinscheduler-service/src/main/java/org/apache/dolphinscheduler/service/quartz/ProcessScheduleJob.java b/dolphinscheduler-service/src/main/java/org/apache/dolphinscheduler/service/quartz/ProcessScheduleJob.java
index 1de5c56fd0..3c9f165e47 100644
--- a/dolphinscheduler-service/src/main/java/org/apache/dolphinscheduler/service/quartz/ProcessScheduleJob.java
+++ b/dolphinscheduler-service/src/main/java/org/apache/dolphinscheduler/service/quartz/ProcessScheduleJob.java
@@ -101,6 +101,7 @@ public class ProcessScheduleJob implements Job {
         command.setWorkerGroup(workerGroup);
         command.setWarningType(schedule.getWarningType());
         command.setProcessInstancePriority(schedule.getProcessInstancePriority());
+        command.setProcessDefinitionVersion(processDefinition.getVersion());
 
         getProcessService().createCommand(command);
     }
diff --git a/dolphinscheduler-service/src/test/java/org/apache/dolphinscheduler/service/process/ProcessServiceTest.java b/dolphinscheduler-service/src/test/java/org/apache/dolphinscheduler/service/process/ProcessServiceTest.java
index e13558dbc1..21125639e8 100644
--- a/dolphinscheduler-service/src/test/java/org/apache/dolphinscheduler/service/process/ProcessServiceTest.java
+++ b/dolphinscheduler-service/src/test/java/org/apache/dolphinscheduler/service/process/ProcessServiceTest.java
@@ -82,7 +82,6 @@ import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
 import com.fasterxml.jackson.databind.JsonNode;
-import com.google.common.collect.Lists;
 
 /**
  * process service test
@@ -119,6 +118,8 @@ public class ProcessServiceTest {
     @Mock
     private ResourceMapper resourceMapper;
 
+    private HashMap<String, ProcessDefinition> processDefinitionCacheMaps = new HashMap<>();
+
     @Test
     public void testCreateSubCommand() {
         ProcessInstance parentInstance = new ProcessInstance();
@@ -240,56 +241,65 @@ public class ProcessServiceTest {
 
         //cannot construct process instance, return null;
         String host = "127.0.0.1";
-        int validThreadNum = 1;
         Command command = new Command();
         command.setProcessDefinitionCode(222);
         command.setCommandType(CommandType.REPEAT_RUNNING);
         command.setCommandParam("{\"" + CMD_PARAM_RECOVER_PROCESS_ID_STRING + "\":\"111\",\""
             + CMD_PARAM_SUB_PROCESS_DEFINE_ID + "\":\"222\"}");
-        Assert.assertNull(processService.handleCommand(logger, host, validThreadNum, command));
+        Assert.assertNull(processService.handleCommand(logger, host, command, processDefinitionCacheMaps));
 
+        int definitionVersion = 1;
+        long definitionCode = 123;
+        int processInstanceId = 222;
         //there is not enough thread for this command
         Command command1 = new Command();
-        command1.setProcessDefinitionCode(123);
+        command1.setProcessDefinitionCode(definitionCode);
+        command1.setProcessDefinitionVersion(definitionVersion);
         command1.setCommandParam("{\"ProcessInstanceId\":222}");
         command1.setCommandType(CommandType.START_PROCESS);
         ProcessDefinition processDefinition = new ProcessDefinition();
         processDefinition.setId(123);
         processDefinition.setName("test");
-        processDefinition.setVersion(1);
-        processDefinition.setCode(11L);
+        processDefinition.setVersion(definitionVersion);
+        processDefinition.setCode(definitionCode);
         processDefinition.setGlobalParams("[{\"prop\":\"startParam1\",\"direct\":\"IN\",\"type\":\"VARCHAR\",\"value\":\"\"}]");
         ProcessInstance processInstance = new ProcessInstance();
-        processInstance.setId(222);
-        processInstance.setProcessDefinitionCode(11L);
-        processInstance.setProcessDefinitionVersion(1);
-        Mockito.when(processDefineMapper.queryByCode(command1.getProcessDefinitionCode())).thenReturn(processDefinition);
+        processInstance.setId(processInstanceId);
+        processInstance.setProcessDefinitionCode(definitionCode);
+        processInstance.setProcessDefinitionVersion(definitionVersion);
         Mockito.when(processDefineLogMapper.queryByDefinitionCodeAndVersion(processInstance.getProcessDefinitionCode(),
             processInstance.getProcessDefinitionVersion())).thenReturn(new ProcessDefinitionLog(processDefinition));
         Mockito.when(processInstanceMapper.queryDetailById(222)).thenReturn(processInstance);
-        Assert.assertNotNull(processService.handleCommand(logger, host, validThreadNum, command1));
+        Assert.assertNotNull(processService.handleCommand(logger, host, command1, processDefinitionCacheMaps));
 
         Command command2 = new Command();
         command2.setCommandParam("{\"ProcessInstanceId\":222,\"StartNodeIdList\":\"n1,n2\"}");
-        command2.setProcessDefinitionCode(123);
+        command2.setProcessDefinitionCode(definitionCode);
+        command2.setProcessDefinitionVersion(definitionVersion);
         command2.setCommandType(CommandType.RECOVER_SUSPENDED_PROCESS);
+        command2.setProcessInstanceId(processInstanceId);
 
-        Assert.assertNotNull(processService.handleCommand(logger, host, validThreadNum, command2));
+        Assert.assertNotNull(processService.handleCommand(logger, host, command2, processDefinitionCacheMaps));
 
         Command command3 = new Command();
-        command3.setProcessDefinitionCode(123);
+        command3.setProcessDefinitionCode(definitionCode);
+        command3.setProcessDefinitionVersion(definitionVersion);
+        command3.setProcessInstanceId(processInstanceId);
         command3.setCommandParam("{\"WaitingThreadInstanceId\":222}");
         command3.setCommandType(CommandType.START_FAILURE_TASK_PROCESS);
-        Assert.assertNotNull(processService.handleCommand(logger, host, validThreadNum, command3));
+        Assert.assertNotNull(processService.handleCommand(logger, host, command3, processDefinitionCacheMaps));
 
         Command command4 = new Command();
-        command4.setProcessDefinitionCode(123);
+        command4.setProcessDefinitionCode(definitionCode);
+        command4.setProcessDefinitionVersion(definitionVersion);
         command4.setCommandParam("{\"WaitingThreadInstanceId\":222,\"StartNodeIdList\":\"n1,n2\"}");
         command4.setCommandType(CommandType.REPEAT_RUNNING);
-        Assert.assertNotNull(processService.handleCommand(logger, host, validThreadNum, command4));
+        command4.setProcessInstanceId(processInstanceId);
+        Assert.assertNotNull(processService.handleCommand(logger, host, command4, processDefinitionCacheMaps));
 
         Command command5 = new Command();
-        command5.setProcessDefinitionCode(123);
+        command5.setProcessDefinitionCode(definitionCode);
+        command5.setProcessDefinitionVersion(definitionVersion);
         HashMap<String, String> startParams = new HashMap<>();
         startParams.put("startParam1", "testStartParam1");
         HashMap<String, String> commandParams = new HashMap<>();
@@ -297,7 +307,7 @@ public class ProcessServiceTest {
         command5.setCommandParam(JSONUtils.toJsonString(commandParams));
         command5.setCommandType(CommandType.START_PROCESS);
         command5.setDryRun(Constants.DRY_RUN_FLAG_NO);
-        ProcessInstance processInstance1 = processService.handleCommand(logger, host, validThreadNum, command5);
+        ProcessInstance processInstance1 = processService.handleCommand(logger, host, command5, processDefinitionCacheMaps);
         Assert.assertTrue(processInstance1.getGlobalParams().contains("\"testStartParam1\""));
     }
 
diff --git a/sql/dolphinscheduler_h2.sql b/sql/dolphinscheduler_h2.sql
index ea4c3485f2..ccd900e4b8 100644
--- a/sql/dolphinscheduler_h2.sql
+++ b/sql/dolphinscheduler_h2.sql
@@ -328,6 +328,8 @@ CREATE TABLE t_ds_command
     worker_group              varchar(64),
     environment_code          bigint(20) DEFAULT '-1',
     dry_run                   int NULL DEFAULT 0,
+    process_instance_id       int(11) DEFAULT 0,
+    process_definition_version int(11) DEFAULT 0,
     PRIMARY KEY (id),
     KEY                       priority_id_index (process_instance_priority, id)
 );
@@ -381,6 +383,8 @@ CREATE TABLE t_ds_error_command
     environment_code          bigint(20) DEFAULT '-1',
     message                   text,
     dry_run                   int NULL DEFAULT 0,
+    process_instance_id       int(11) DEFAULT 0,
+    process_definition_version int(11) DEFAULT 0,
     PRIMARY KEY (id)
 );
 
diff --git a/sql/dolphinscheduler_mysql.sql b/sql/dolphinscheduler_mysql.sql
index 82f41a2e22..82ed90f2de 100644
--- a/sql/dolphinscheduler_mysql.sql
+++ b/sql/dolphinscheduler_mysql.sql
@@ -317,22 +317,24 @@ CREATE TABLE `t_ds_alertgroup`(
 -- ----------------------------
 DROP TABLE IF EXISTS `t_ds_command`;
 CREATE TABLE `t_ds_command` (
-  `id` int(11) NOT NULL AUTO_INCREMENT COMMENT 'key',
-  `command_type` tinyint(4) DEFAULT NULL COMMENT 'Command type: 0 start workflow, 1 start execution from current node, 2 resume fault-tolerant workflow, 3 resume pause process, 4 start execution from failed node, 5 complement, 6 schedule, 7 rerun, 8 pause, 9 stop, 10 resume waiting thread',
-  `process_definition_code` bigint(20) DEFAULT NULL COMMENT 'process definition code',
-  `command_param` text COMMENT 'json command parameters',
-  `task_depend_type` tinyint(4) DEFAULT NULL COMMENT 'Node dependency type: 0 current node, 1 forward, 2 backward',
-  `failure_strategy` tinyint(4) DEFAULT '0' COMMENT 'Failed policy: 0 end, 1 continue',
-  `warning_type` tinyint(4) DEFAULT '0' COMMENT 'Alarm type: 0 is not sent, 1 process is sent successfully, 2 process is sent failed, 3 process is sent successfully and all failures are sent',
-  `warning_group_id` int(11) DEFAULT NULL COMMENT 'warning group',
-  `schedule_time` datetime DEFAULT NULL COMMENT 'schedule time',
-  `start_time` datetime DEFAULT NULL COMMENT 'start time',
-  `executor_id` int(11) DEFAULT NULL COMMENT 'executor id',
-  `update_time` datetime DEFAULT NULL COMMENT 'update time',
+  `id`                        int(11)    NOT NULL AUTO_INCREMENT COMMENT 'key',
+  `command_type`              tinyint(4) DEFAULT NULL COMMENT 'Command type: 0 start workflow, 1 start execution from current node, 2 resume fault-tolerant workflow, 3 resume pause process, 4 start execution from failed node, 5 complement, 6 schedule, 7 rerun, 8 pause, 9 stop, 10 resume waiting thread',
+  `process_definition_code`   bigint(20) DEFAULT NULL COMMENT 'process definition code',
+  `command_param`             text COMMENT 'json command parameters',
+  `task_depend_type`          tinyint(4) DEFAULT NULL COMMENT 'Node dependency type: 0 current node, 1 forward, 2 backward',
+  `failure_strategy`          tinyint(4) DEFAULT '0' COMMENT 'Failed policy: 0 end, 1 continue',
+  `warning_type`              tinyint(4) DEFAULT '0' COMMENT 'Alarm type: 0 is not sent, 1 process is sent successfully, 2 process is sent failed, 3 process is sent successfully and all failures are sent',
+  `warning_group_id`          int(11) DEFAULT NULL COMMENT 'warning group',
+  `schedule_time`             datetime DEFAULT NULL COMMENT 'schedule time',
+  `start_time`                datetime DEFAULT NULL COMMENT 'start time',
+  `executor_id`               int(11) DEFAULT NULL COMMENT 'executor id',
+  `update_time`               datetime DEFAULT NULL COMMENT 'update time',
   `process_instance_priority` int(11) DEFAULT NULL COMMENT 'process instance priority: 0 Highest,1 High,2 Medium,3 Low,4 Lowest',
-  `worker_group` varchar(64)  COMMENT 'worker group',
-  `environment_code` bigint(20) DEFAULT '-1' COMMENT 'environment code',
-  `dry_run` int NULL DEFAULT 0 COMMENT 'dry run flag：0 normal, 1 dry run',
+  `worker_group`              varchar(64)  COMMENT 'worker group',
+  `environment_code`          bigint(20) DEFAULT '-1' COMMENT 'environment code',
+  `dry_run`                   int NULL DEFAULT 0 COMMENT 'dry run flag：0 normal, 1 dry run',
+  `process_instance_id`       int(11) DEFAULT 0 COMMENT 'process instance id',
+  `process_definition_version` int(11) DEFAULT 0 COMMENT 'process definition version',
   PRIMARY KEY (`id`),
   KEY `priority_id_index` (`process_instance_priority`,`id`) USING BTREE
 ) ENGINE=InnoDB AUTO_INCREMENT=1 DEFAULT CHARSET=utf8;
@@ -384,6 +386,8 @@ CREATE TABLE `t_ds_error_command` (
   `environment_code` bigint(20) DEFAULT '-1' COMMENT 'environment code',
   `message` text COMMENT 'message',
   `dry_run` int NULL DEFAULT NULL COMMENT 'dry run flag: 0 normal, 1 dry run',
+  `process_instance_id` int(11) DEFAULT 0 COMMENT 'process instance id: 0',
+  `process_definition_version` int(11) DEFAULT 0 COMMENT 'process definition version',
   PRIMARY KEY (`id`) USING BTREE
 ) ENGINE=InnoDB DEFAULT CHARSET=utf8 ROW_FORMAT=DYNAMIC;
 
diff --git a/sql/dolphinscheduler_postgre.sql b/sql/dolphinscheduler_postgre.sql
index 40a35354cb..5d33d036f9 100644
--- a/sql/dolphinscheduler_postgre.sql
+++ b/sql/dolphinscheduler_postgre.sql
@@ -240,21 +240,23 @@ CREATE TABLE t_ds_alertgroup(
 DROP TABLE IF EXISTS t_ds_command;
 CREATE TABLE t_ds_command (
   id int NOT NULL  ,
-  command_type int DEFAULT NULL ,
-  process_definition_code bigint NOT NULL ,
-  command_param text ,
-  task_depend_type int DEFAULT NULL ,
-  failure_strategy int DEFAULT '0' ,
-  warning_type int DEFAULT '0' ,
-  warning_group_id int DEFAULT NULL ,
-  schedule_time timestamp DEFAULT NULL ,
-  start_time timestamp DEFAULT NULL ,
-  executor_id int DEFAULT NULL ,
-  update_time timestamp DEFAULT NULL ,
+  command_type              int DEFAULT NULL ,
+  process_definition_code   bigint NOT NULL ,
+  command_param             text ,
+  task_depend_type          int DEFAULT NULL ,
+  failure_strategy          int DEFAULT '0' ,
+  warning_type              int DEFAULT '0' ,
+  warning_group_id          int DEFAULT NULL ,
+  schedule_time             timestamp DEFAULT NULL ,
+  start_time                timestamp DEFAULT NULL ,
+  executor_id               int DEFAULT NULL ,
+  update_time               timestamp DEFAULT NULL ,
   process_instance_priority int DEFAULT NULL ,
-  worker_group varchar(64),
-  environment_code bigint DEFAULT '-1',
-  dry_run int DEFAULT '0' ,
+  worker_group              varchar(64),
+  environment_code          bigint DEFAULT '-1',
+  dry_run                   int DEFAULT '0' ,
+  process_instance_id       int DEFAULT 0,
+  process_definition_version int DEFAULT 0,
   PRIMARY KEY (id)
 ) ;
 
@@ -284,23 +286,24 @@ CREATE TABLE t_ds_datasource (
 
 DROP TABLE IF EXISTS t_ds_error_command;
 CREATE TABLE t_ds_error_command (
-  id int NOT NULL ,
-  command_type int DEFAULT NULL ,
-  executor_id int DEFAULT NULL ,
-  process_definition_code bigint NOT NULL ,
-  command_param text ,
-  task_depend_type int DEFAULT NULL ,
-  failure_strategy int DEFAULT '0' ,
-  warning_type int DEFAULT '0' ,
-  warning_group_id int DEFAULT NULL ,
-  schedule_time timestamp DEFAULT NULL ,
-  start_time timestamp DEFAULT NULL ,
-  update_time timestamp DEFAULT NULL ,
+  id int NOT NULL  ,
+  command_type              int DEFAULT NULL ,
+  process_definition_code   bigint NOT NULL ,
+  command_param             text ,
+  task_depend_type          int DEFAULT NULL ,
+  failure_strategy          int DEFAULT '0' ,
+  warning_type              int DEFAULT '0' ,
+  warning_group_id          int DEFAULT NULL ,
+  schedule_time             timestamp DEFAULT NULL ,
+  start_time                timestamp DEFAULT NULL ,
+  executor_id               int DEFAULT NULL ,
+  update_time               timestamp DEFAULT NULL ,
   process_instance_priority int DEFAULT NULL ,
-  worker_group varchar(64),
-  environment_code bigint DEFAULT '-1',
-  message text ,
-  dry_ru int DEFAULT '0' ,
+  worker_group              varchar(64),
+  environment_code          bigint DEFAULT '-1',
+  dry_run                   int DEFAULT '0' ,
+  process_instance_id       int DEFAULT 0,
+  process_definition_version int DEFAULT 0,
   PRIMARY KEY (id)
 );
 --