Browse Source

[DS-14566][master] fix serial_wait command_param does not replace global_params (#14606)

This closes #14566

Co-authored-by: yangxin3 <yangxin3@newhope.cn>
Co-authored-by: Eric Gao <ericgao.apache@gmail.com>
Co-authored-by: xiangzihao <460888207@qq.com>
3.2.1-prepare
KingsleyY 1 year ago committed by GitHub
parent
commit
4a80e3a021
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
  1. 11
      dolphinscheduler-master/src/main/java/org/apache/dolphinscheduler/server/master/runner/WorkflowExecuteRunnable.java
  2. 4
      dolphinscheduler-service/src/main/java/org/apache/dolphinscheduler/service/process/ProcessServiceImpl.java

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

@ -96,6 +96,7 @@ import org.apache.dolphinscheduler.service.queue.PeerTaskInstancePriorityQueue;
import org.apache.dolphinscheduler.service.utils.DagHelper; import org.apache.dolphinscheduler.service.utils.DagHelper;
import org.apache.commons.collections4.CollectionUtils; import org.apache.commons.collections4.CollectionUtils;
import org.apache.commons.collections4.MapUtils;
import org.apache.commons.lang3.StringUtils; import org.apache.commons.lang3.StringUtils;
import org.apache.commons.lang3.tuple.Pair; import org.apache.commons.lang3.tuple.Pair;
@ -784,6 +785,16 @@ public class WorkflowExecuteRunnable implements IWorkflowExecuteRunnable {
return; return;
} }
Map<String, Object> cmdParam = new HashMap<>(); Map<String, Object> cmdParam = new HashMap<>();
// write the parameters of the nextProcessInstance to command
if (StringUtils.isNotEmpty(nextProcessInstance.getCommandParam())) {
Map<String, String> commandStartParamsMap = JSONUtils.toMap(nextProcessInstance.getCommandParam());
if (MapUtils.isNotEmpty(commandStartParamsMap)) {
Map<String, String> paramsMap = JSONUtils.toMap(commandStartParamsMap.get(CMD_PARAM_START_PARAMS));
if (MapUtils.isNotEmpty(paramsMap)) {
cmdParam.put(CMD_PARAM_START_PARAMS, JSONUtils.toJsonString(paramsMap));
}
}
}
cmdParam.put(CMD_PARAM_RECOVER_PROCESS_ID_STRING, nextInstanceId); cmdParam.put(CMD_PARAM_RECOVER_PROCESS_ID_STRING, nextInstanceId);
Command command = new Command(); Command command = new Command();
command.setCommandType(CommandType.RECOVER_SERIAL_WAIT); command.setCommandType(CommandType.RECOVER_SERIAL_WAIT);

4
dolphinscheduler-service/src/main/java/org/apache/dolphinscheduler/service/process/ProcessServiceImpl.java

@ -744,6 +744,7 @@ public class ProcessServiceImpl implements ProcessService {
* @param host host * @param host host
* @return process instance * @return process instance
*/ */
@Override
public @Nullable ProcessInstance constructProcessInstance(Command command, public @Nullable ProcessInstance constructProcessInstance(Command command,
String host) throws CronParseException, CodeGenerateException { String host) throws CronParseException, CodeGenerateException {
ProcessInstance processInstance; ProcessInstance processInstance;
@ -774,7 +775,8 @@ public class ProcessServiceImpl implements ProcessService {
CommandType commandTypeIfComplement = getCommandTypeIfComplement(processInstance, command); CommandType commandTypeIfComplement = getCommandTypeIfComplement(processInstance, command);
// reset global params while repeat running and recover tolerance fault process is needed by cmdParam // reset global params while repeat running and recover tolerance fault process is needed by cmdParam
if (commandTypeIfComplement == CommandType.REPEAT_RUNNING || if (commandTypeIfComplement == CommandType.REPEAT_RUNNING ||
commandTypeIfComplement == CommandType.RECOVER_TOLERANCE_FAULT_PROCESS) { commandTypeIfComplement == CommandType.RECOVER_TOLERANCE_FAULT_PROCESS ||
commandTypeIfComplement == CommandType.RECOVER_SERIAL_WAIT) {
setGlobalParamIfCommanded(processDefinition, cmdParam); setGlobalParamIfCommanded(processDefinition, cmdParam);
} }

Loading…
Cancel
Save