Browse Source

[Feature-#5565][Master Worker-Server] Global Param passed by sense dependencies (#5603)

* add globalParams new plan with varPool

* add unit test

* add python task varPoolParams


Co-authored-by: wangxj <wangxj31>
2.0.7-release
wangxj3 4 years ago committed by GitHub
parent
commit
6df58e8d78
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
  1. 3
      dolphinscheduler-common/src/main/java/org/apache/dolphinscheduler/common/enums/DataType.java
  2. 182
      dolphinscheduler-common/src/main/java/org/apache/dolphinscheduler/common/task/AbstractParameters.java
  3. 60
      dolphinscheduler-common/src/main/java/org/apache/dolphinscheduler/common/task/shell/ShellParameters.java
  4. 54
      dolphinscheduler-common/src/main/java/org/apache/dolphinscheduler/common/task/sql/SqlParameters.java
  5. 3
      dolphinscheduler-common/src/main/java/org/apache/dolphinscheduler/common/utils/VarPoolUtils.java
  6. 2
      dolphinscheduler-common/src/test/java/org/apache/dolphinscheduler/common/task/EntityTestUtils.java
  7. 30
      dolphinscheduler-common/src/test/java/org/apache/dolphinscheduler/common/task/SqlParametersTest.java
  8. 12
      dolphinscheduler-remote/src/main/java/org/apache/dolphinscheduler/remote/command/TaskExecuteResponseCommand.java
  9. 14
      dolphinscheduler-server/src/main/java/org/apache/dolphinscheduler/server/builder/TaskExecutionContextBuilder.java
  10. 13
      dolphinscheduler-server/src/main/java/org/apache/dolphinscheduler/server/entity/TaskExecutionContext.java
  11. 3
      dolphinscheduler-server/src/main/java/org/apache/dolphinscheduler/server/master/processor/TaskResponseProcessor.java
  12. 15
      dolphinscheduler-server/src/main/java/org/apache/dolphinscheduler/server/master/processor/queue/TaskResponseEvent.java
  13. 3
      dolphinscheduler-server/src/main/java/org/apache/dolphinscheduler/server/master/processor/queue/TaskResponseService.java
  14. 90
      dolphinscheduler-server/src/main/java/org/apache/dolphinscheduler/server/master/runner/MasterExecThread.java
  15. 8
      dolphinscheduler-server/src/main/java/org/apache/dolphinscheduler/server/utils/ParamUtils.java
  16. 7
      dolphinscheduler-server/src/main/java/org/apache/dolphinscheduler/server/worker/runner/TaskExecuteThread.java
  17. 13
      dolphinscheduler-server/src/main/java/org/apache/dolphinscheduler/server/worker/task/AbstractCommandExecutor.java
  18. 28
      dolphinscheduler-server/src/main/java/org/apache/dolphinscheduler/server/worker/task/AbstractTask.java
  19. 1
      dolphinscheduler-server/src/main/java/org/apache/dolphinscheduler/server/worker/task/datax/DataxTask.java
  20. 1
      dolphinscheduler-server/src/main/java/org/apache/dolphinscheduler/server/worker/task/flink/FlinkTask.java
  21. 1
      dolphinscheduler-server/src/main/java/org/apache/dolphinscheduler/server/worker/task/http/HttpTask.java
  22. 1
      dolphinscheduler-server/src/main/java/org/apache/dolphinscheduler/server/worker/task/mr/MapReduceTask.java
  23. 1
      dolphinscheduler-server/src/main/java/org/apache/dolphinscheduler/server/worker/task/procedure/ProcedureTask.java
  24. 3
      dolphinscheduler-server/src/main/java/org/apache/dolphinscheduler/server/worker/task/python/PythonTask.java
  25. 17
      dolphinscheduler-server/src/main/java/org/apache/dolphinscheduler/server/worker/task/shell/ShellTask.java
  26. 1
      dolphinscheduler-server/src/main/java/org/apache/dolphinscheduler/server/worker/task/spark/SparkTask.java
  27. 11
      dolphinscheduler-server/src/main/java/org/apache/dolphinscheduler/server/worker/task/sql/SqlTask.java
  28. 1
      dolphinscheduler-server/src/main/java/org/apache/dolphinscheduler/server/worker/task/sqoop/SqoopTask.java
  29. 48
      dolphinscheduler-server/src/test/java/org/apache/dolphinscheduler/server/master/MasterExecThreadTest.java
  30. 33
      dolphinscheduler-server/src/test/java/org/apache/dolphinscheduler/server/master/ParamsTest.java
  31. 3
      dolphinscheduler-server/src/test/java/org/apache/dolphinscheduler/server/master/processor/queue/TaskResponseServiceTest.java
  32. 42
      dolphinscheduler-server/src/test/java/org/apache/dolphinscheduler/server/utils/ParamUtilsTest.java
  33. 2
      dolphinscheduler-server/src/test/java/org/apache/dolphinscheduler/server/worker/processor/TaskCallbackServiceTest.java
  34. 6
      dolphinscheduler-server/src/test/java/org/apache/dolphinscheduler/server/worker/processor/TaskKillProcessorTest.java
  35. 3
      dolphinscheduler-server/src/test/java/org/apache/dolphinscheduler/server/worker/runner/TaskExecuteThreadTest.java
  36. 53
      dolphinscheduler-server/src/test/java/org/apache/dolphinscheduler/server/worker/task/AbstractCommandExecutorTest.java
  37. 11
      dolphinscheduler-server/src/test/java/org/apache/dolphinscheduler/server/worker/task/ShellTaskReturnTest.java
  38. 5
      dolphinscheduler-server/src/test/java/org/apache/dolphinscheduler/server/worker/task/TaskManagerTest.java
  39. 77
      dolphinscheduler-server/src/test/java/org/apache/dolphinscheduler/server/worker/task/TaskParamsTest.java
  40. 12
      dolphinscheduler-server/src/test/java/org/apache/dolphinscheduler/server/worker/task/shell/ShellTaskTest.java
  41. 2
      dolphinscheduler-server/src/test/java/org/apache/dolphinscheduler/server/worker/task/sql/SqlTaskTest.java
  42. 55
      dolphinscheduler-service/src/main/java/org/apache/dolphinscheduler/service/process/ProcessService.java
  43. 15
      dolphinscheduler-service/src/test/java/org/apache/dolphinscheduler/service/process/ProcessServiceTest.java
  44. 2
      pom.xml

3
dolphinscheduler-common/src/main/java/org/apache/dolphinscheduler/common/enums/DataType.java

@ -30,6 +30,7 @@ public enum DataType {
* 6 time, "HH:MM:SS"
* 7 time stamp
* 8 Boolean
* 9 list <String>
*/
VARCHAR,INTEGER,LONG,FLOAT,DOUBLE,DATE,TIME,TIMESTAMP,BOOLEAN
VARCHAR,INTEGER,LONG,FLOAT,DOUBLE,DATE,TIME,TIMESTAMP,BOOLEAN,LIST
}

182
dolphinscheduler-common/src/main/java/org/apache/dolphinscheduler/common/task/AbstractParameters.java

@ -16,55 +16,163 @@
*/
package org.apache.dolphinscheduler.common.task;
import org.apache.dolphinscheduler.common.enums.Direct;
import org.apache.dolphinscheduler.common.process.Property;
import org.apache.dolphinscheduler.common.process.ResourceInfo;
import org.apache.dolphinscheduler.common.utils.CollectionUtils;
import org.apache.dolphinscheduler.common.utils.JSONUtils;
import org.apache.dolphinscheduler.common.utils.StringUtils;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.Iterator;
import java.util.LinkedHashMap;
import java.util.List;
import java.util.Map;
import com.fasterxml.jackson.databind.JsonNode;
import com.fasterxml.jackson.databind.node.ArrayNode;
/**
* job params related class
*/
public abstract class AbstractParameters implements IParameters {
@Override
public abstract boolean checkParameters();
@Override
public abstract List<ResourceInfo> getResourceFilesList();
/**
* local parameters
*/
public List<Property> localParams;
/**
* get local parameters list
* @return Property list
*/
public List<Property> getLocalParams() {
return localParams;
}
public void setLocalParams(List<Property> localParams) {
this.localParams = localParams;
}
/**
* get local parameters map
* @return parameters map
*/
public Map<String,Property> getLocalParametersMap() {
if (localParams != null) {
Map<String,Property> localParametersMaps = new LinkedHashMap<>();
for (Property property : localParams) {
localParametersMaps.put(property.getProp(),property);
@Override
public abstract boolean checkParameters();
@Override
public abstract List<ResourceInfo> getResourceFilesList();
/**
* local parameters
*/
public List<Property> localParams;
/**
* var pool
*/
public List<Property> varPool;
/**
* get local parameters list
*
* @return Property list
*/
public List<Property> getLocalParams() {
return localParams;
}
public void setLocalParams(List<Property> localParams) {
this.localParams = localParams;
}
/**
* get local parameters map
*
* @return parameters map
*/
public Map<String, Property> getLocalParametersMap() {
if (localParams != null) {
Map<String, Property> localParametersMaps = new LinkedHashMap<>();
for (Property property : localParams) {
localParametersMaps.put(property.getProp(), property);
}
return localParametersMaps;
}
return null;
}
/**
* get varPool map
*
* @return parameters map
*/
public Map<String, Property> getVarPoolMap() {
if (varPool != null) {
Map<String, Property> varPoolMap = new LinkedHashMap<>();
for (Property property : varPool) {
varPoolMap.put(property.getProp(), property);
}
return varPoolMap;
}
return null;
}
public List<Property> getVarPool() {
return varPool;
}
public void setVarPool(String varPool) {
if (StringUtils.isEmpty(varPool)) {
this.varPool = new ArrayList<>();
} else {
this.varPool = JSONUtils.toList(varPool, Property.class);
}
}
public void dealOutParam(String result) {
if (CollectionUtils.isEmpty(localParams)) {
return;
}
List<Property> outProperty = getOutProperty(localParams);
if (CollectionUtils.isEmpty(outProperty)) {
return;
}
if (StringUtils.isEmpty(result)) {
varPool.addAll(outProperty);
return;
}
Map<String, String> taskResult = getMapByString(result);
if (taskResult == null || taskResult.size() == 0) {
return;
}
for (Property info : outProperty) {
info.setValue(taskResult.get(info.getProp()));
varPool.add(info);
}
}
public List<Property> getOutProperty(List<Property> params) {
if (CollectionUtils.isEmpty(params)) {
return new ArrayList<>();
}
List<Property> result = new ArrayList<>();
for (Property info : params) {
if (info.getDirect() == Direct.OUT) {
result.add(info);
}
}
return result;
}
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;
}
/**
* shell's result format is key=value$VarPool$key=value$VarPool$
* @param result
* @return
*/
public static Map<String, String> getMapByString(String result) {
String[] formatResult = result.split("\\$VarPool\\$");
Map<String, String> format = new HashMap<>();
for (String info : formatResult) {
if (StringUtils.isNotEmpty(info) && info.contains("=")) {
String[] keyValue = info.split("=");
format.put(keyValue[0], keyValue[1]);
}
}
return localParametersMaps;
}
return null;
}
return format;
}
}

60
dolphinscheduler-common/src/main/java/org/apache/dolphinscheduler/common/task/shell/ShellParameters.java

@ -14,52 +14,52 @@
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package org.apache.dolphinscheduler.common.task.shell;
package org.apache.dolphinscheduler.common.task.shell;
import org.apache.dolphinscheduler.common.process.ResourceInfo;
import org.apache.dolphinscheduler.common.task.AbstractParameters;
import java.util.List;
import java.util.stream.Collectors;
/**
* shell parameters
*/
public class ShellParameters extends AbstractParameters {
/**
* shell script
*/
private String rawScript;
/**
* shell script
*/
private String rawScript;
/**
* resource list
*/
private List<ResourceInfo> resourceList;
/**
* resource list
*/
private List<ResourceInfo> resourceList;
public String getRawScript() {
return rawScript;
}
public String getRawScript() {
return rawScript;
}
public void setRawScript(String rawScript) {
this.rawScript = rawScript;
}
public void setRawScript(String rawScript) {
this.rawScript = rawScript;
}
public List<ResourceInfo> getResourceList() {
return resourceList;
}
public List<ResourceInfo> getResourceList() {
return resourceList;
}
public void setResourceList(List<ResourceInfo> resourceList) {
this.resourceList = resourceList;
}
public void setResourceList(List<ResourceInfo> resourceList) {
this.resourceList = resourceList;
}
@Override
public boolean checkParameters() {
return rawScript != null && !rawScript.isEmpty();
}
@Override
public boolean checkParameters() {
return rawScript != null && !rawScript.isEmpty();
}
@Override
public List<ResourceInfo> getResourceFilesList() {
return resourceList;
}
@Override
public List<ResourceInfo> getResourceFilesList() {
return resourceList;
}
}

54
dolphinscheduler-common/src/main/java/org/apache/dolphinscheduler/common/task/sql/SqlParameters.java

@ -17,12 +17,19 @@
package org.apache.dolphinscheduler.common.task.sql;
import org.apache.dolphinscheduler.common.enums.DataType;
import org.apache.dolphinscheduler.common.process.Property;
import org.apache.dolphinscheduler.common.process.ResourceInfo;
import org.apache.dolphinscheduler.common.task.AbstractParameters;
import org.apache.dolphinscheduler.common.utils.CollectionUtils;
import org.apache.dolphinscheduler.common.utils.JSONUtils;
import org.apache.dolphinscheduler.common.utils.StringUtils;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.Set;
/**
* Sql/Hql parameter
@ -218,6 +225,53 @@ public class SqlParameters extends AbstractParameters {
return new ArrayList<>();
}
@Override
public void dealOutParam(String result) {
if (CollectionUtils.isEmpty(localParams)) {
return;
}
List<Property> outProperty = getOutProperty(localParams);
if (CollectionUtils.isEmpty(outProperty)) {
return;
}
if (StringUtils.isEmpty(result)) {
varPool.addAll(outProperty);
return;
}
List<Map<String, String>> sqlResult = getListMapByString(result);
if (CollectionUtils.isEmpty(sqlResult)) {
return;
}
//if sql return more than one line
if (sqlResult.size() > 1) {
Map<String, List<String>> sqlResultFormat = new HashMap<>();
//init sqlResultFormat
Set<String> keySet = sqlResult.get(0).keySet();
for (String key : keySet) {
sqlResultFormat.put(key, new ArrayList<>());
}
for (Map<String, String> info : sqlResult) {
for (String key : info.keySet()) {
sqlResultFormat.get(key).add(String.valueOf(info.get(key)));
}
}
for (Property info : outProperty) {
if (info.getType() == DataType.LIST) {
info.setValue(JSONUtils.toJsonString(sqlResultFormat.get(info.getProp())));
varPool.add(info);
}
}
} else {
//result only one line
Map<String, String> firstRow = sqlResult.get(0);
for (Property info : outProperty) {
info.setValue(String.valueOf(firstRow.get(info.getProp())));
varPool.add(info);
}
}
}
@Override
public String toString() {
return "SqlParameters{"

3
dolphinscheduler-common/src/main/java/org/apache/dolphinscheduler/common/utils/VarPoolUtils.java

@ -21,7 +21,6 @@ import org.apache.dolphinscheduler.common.model.TaskNode;
import java.text.ParseException;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.Map;
public class VarPoolUtils {
@ -71,7 +70,7 @@ public class VarPoolUtils {
if (kvs.length == 2) {
propToValue.put(kvs[0], kvs[1]);
} else {
throw new ParseException(kv, 2);
return;
}
}
}

2
dolphinscheduler-common/src/test/java/org/apache/dolphinscheduler/common/task/EntityTestUtils.java

@ -32,7 +32,7 @@ public class EntityTestUtils {
static {
OBJECT_MAP.put("java.lang.Long", 1L);
OBJECT_MAP.put("java.lang.String", "test");
OBJECT_MAP.put("java.lang.String", "[{\"direct\":\"OUT\",\"prop\":\"percentage5\",\"type\":\"VARCHAR\",\"value\":\"qwe\"}]");
OBJECT_MAP.put("java.lang.Integer", 1);
OBJECT_MAP.put("int", 1);
OBJECT_MAP.put("long", 1L);

30
dolphinscheduler-common/src/test/java/org/apache/dolphinscheduler/common/task/SqlParametersTest.java

@ -17,9 +17,17 @@
package org.apache.dolphinscheduler.common.task;
import static org.junit.Assert.assertNotNull;
import org.apache.dolphinscheduler.common.enums.DataType;
import org.apache.dolphinscheduler.common.enums.Direct;
import org.apache.dolphinscheduler.common.process.Property;
import org.apache.dolphinscheduler.common.task.sql.SqlParameters;
import org.apache.dolphinscheduler.common.utils.CollectionUtils;
import java.util.ArrayList;
import java.util.List;
import org.junit.Assert;
import org.junit.Test;
@ -39,6 +47,14 @@ public class SqlParametersTest {
@Test
public void testSqlParameters() {
List<Property> properties = new ArrayList<>();
Property property = new Property();
property.setProp("test1");
property.setDirect(Direct.OUT);
property.setType(DataType.VARCHAR);
property.setValue("test1");
properties.add(property);
SqlParameters sqlParameters = new SqlParameters();
Assert.assertTrue(CollectionUtils.isEmpty(sqlParameters.getResourceFilesList()));
@ -66,6 +82,18 @@ public class SqlParametersTest {
Assert.assertEquals(groupId, sqlParameters.getGroupId());
Assert.assertEquals(limit, sqlParameters.getLimit());
Assert.assertTrue(sqlParameters.checkParameters());
String sqlResult = "[{\"id\":6,\"test1\":\"6\"},{\"id\":70002,\"test1\":\"+1\"}]";
String sqlResult1 = "[{\"id\":6,\"test1\":\"6\"}]";
sqlParameters.setLocalParams(properties);
sqlParameters.varPool = new ArrayList<>();
sqlParameters.dealOutParam(sqlResult1);
assertNotNull(sqlParameters.getVarPool().get(0));
property.setType(DataType.LIST);
properties.clear();
properties.add(property);
sqlParameters.setLocalParams(properties);
sqlParameters.dealOutParam(sqlResult);
assertNotNull(sqlParameters.getVarPool().get(0));
}
}

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

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

14
dolphinscheduler-server/src/main/java/org/apache/dolphinscheduler/server/builder/TaskExecutionContextBuilder.java

@ -21,8 +21,15 @@ import static org.apache.dolphinscheduler.common.Constants.SEC_2_MINUTES_TIME_UN
import org.apache.dolphinscheduler.common.enums.TaskTimeoutStrategy;
import org.apache.dolphinscheduler.common.enums.TimeoutFlag;
import org.apache.dolphinscheduler.dao.entity.*;
import org.apache.dolphinscheduler.server.entity.*;
import org.apache.dolphinscheduler.dao.entity.ProcessDefinition;
import org.apache.dolphinscheduler.dao.entity.ProcessInstance;
import org.apache.dolphinscheduler.dao.entity.TaskDefinition;
import org.apache.dolphinscheduler.dao.entity.TaskInstance;
import org.apache.dolphinscheduler.server.entity.DataxTaskExecutionContext;
import org.apache.dolphinscheduler.server.entity.ProcedureTaskExecutionContext;
import org.apache.dolphinscheduler.server.entity.SQLTaskExecutionContext;
import org.apache.dolphinscheduler.server.entity.SqoopTaskExecutionContext;
import org.apache.dolphinscheduler.server.entity.TaskExecutionContext;
/**
* TaskExecutionContext builder
@ -41,7 +48,7 @@ public class TaskExecutionContextBuilder {
* @param taskInstance taskInstance
* @return TaskExecutionContextBuilder
*/
public TaskExecutionContextBuilder buildTaskInstanceRelatedInfo(TaskInstance taskInstance){
public TaskExecutionContextBuilder buildTaskInstanceRelatedInfo(TaskInstance taskInstance) {
taskExecutionContext.setTaskInstanceId(taskInstance.getId());
taskExecutionContext.setTaskName(taskInstance.getName());
taskExecutionContext.setFirstSubmitTime(taskInstance.getFirstSubmitTime());
@ -52,6 +59,7 @@ public class TaskExecutionContextBuilder {
taskExecutionContext.setHost(taskInstance.getHost());
taskExecutionContext.setResources(taskInstance.getResources());
taskExecutionContext.setDelayTime(taskInstance.getDelayTime());
taskExecutionContext.setVarPool(taskInstance.getVarPool());
return this;
}

13
dolphinscheduler-server/src/main/java/org/apache/dolphinscheduler/server/entity/TaskExecutionContext.java

@ -216,6 +216,11 @@ public class TaskExecutionContext implements Serializable {
*/
private SqoopTaskExecutionContext sqoopTaskExecutionContext;
/**
* taskInstance varPool
*/
private String varPool;
/**
* procedure TaskExecutionContext
*/
@ -556,4 +561,12 @@ public class TaskExecutionContext implements Serializable {
+ ", procedureTaskExecutionContext=" + procedureTaskExecutionContext
+ '}';
}
public String getVarPool() {
return varPool;
}
public void setVarPool(String varPool) {
this.varPool = varPool;
}
}

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

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

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

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

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

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

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

@ -22,7 +22,6 @@ 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;
@ -47,7 +46,6 @@ import org.apache.dolphinscheduler.common.utils.JSONUtils;
import org.apache.dolphinscheduler.common.utils.OSUtils;
import org.apache.dolphinscheduler.common.utils.ParameterUtils;
import org.apache.dolphinscheduler.common.utils.StringUtils;
import org.apache.dolphinscheduler.common.utils.VarPoolUtils;
import org.apache.dolphinscheduler.dao.entity.ProcessInstance;
import org.apache.dolphinscheduler.dao.entity.ProjectUser;
import org.apache.dolphinscheduler.dao.entity.Schedule;
@ -60,7 +58,6 @@ import org.apache.dolphinscheduler.service.process.ProcessService;
import org.apache.dolphinscheduler.service.quartz.cron.CronUtils;
import org.apache.dolphinscheduler.service.queue.PeerTaskInstancePriorityQueue;
import java.text.ParseException;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.Collection;
@ -470,8 +467,6 @@ public class MasterExecThread implements Runnable {
* @return TaskInstance
*/
private TaskInstance createTaskInstance(ProcessInstance processInstance, TaskNode taskNode) {
//update processInstance for update the globalParams
this.processInstance = this.processService.findProcessInstanceById(this.processInstance.getId());
TaskInstance taskInstance = findTaskIfExists(taskNode.getCode(), taskNode.getVersion());
if (taskInstance == null) {
taskInstance = new TaskInstance();
@ -503,6 +498,9 @@ public class MasterExecThread implements Runnable {
// retry task instance interval
taskInstance.setRetryInterval(taskNode.getRetryInterval());
//set task param
taskInstance.setTaskParams(taskNode.getTaskParams());
// task instance priority
if (taskNode.getTaskInstancePriority() == null) {
taskInstance.setTaskInstancePriority(Priority.MEDIUM);
@ -518,54 +516,74 @@ public class MasterExecThread implements Runnable {
} else {
taskInstance.setWorkerGroup(taskWorkerGroup);
}
taskInstance.setTaskParams(globalParamToTaskParams(taskNode.getTaskParams()));
// delay execution time
taskInstance.setDelayTime(taskNode.getDelayTime());
}
//get pre task ,get all the task varPool to this task
Set<String> preTask = dag.getPreviousNodes(taskInstance.getName());
getPreVarPool(taskInstance, preTask);
return taskInstance;
}
private String globalParamToTaskParams(String params) {
String globalParams = this.processInstance.getGlobalParams();
if (StringUtils.isBlank(globalParams)) {
return params;
}
Map<String, String> globalMap = processService.getGlobalParamMap(globalParams);
if (globalMap == null || globalMap.size() == 0) {
return params;
}
// the process global param save in localParams
Map<String, Object> result = JSONUtils.toMap(params, 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) {
String paramName = info.getProp();
if (StringUtils.isNotEmpty(paramName) && propToValue.containsKey(paramName)) {
info.setValue((String) propToValue.get(paramName));
public void getPreVarPool(TaskInstance taskInstance, Set<String> preTask) {
Map<String,Property> allProperty = new HashMap<>();
Map<String,TaskInstance> allTaskInstance = new HashMap<>();
if (CollectionUtils.isNotEmpty(preTask)) {
for (String preTaskName : preTask) {
TaskInstance preTaskInstance = completeTaskList.get(preTaskName);
if (preTaskInstance == null) {
continue;
}
if (info.getDirect().equals(Direct.IN)) {
String value = globalMap.get(paramName);
if (StringUtils.isNotEmpty(value)) {
info.setValue(value);
String preVarPool = preTaskInstance.getVarPool();
if (StringUtils.isNotEmpty(preVarPool)) {
List<Property> properties = JSONUtils.toList(preVarPool, Property.class);
for (Property info : properties) {
setVarPoolValue(allProperty, allTaskInstance, preTaskInstance, info);
}
}
}
result.put(LOCAL_PARAMS, allParam);
if (allProperty.size() > 0) {
taskInstance.setVarPool(JSONUtils.toJsonString(allProperty.values()));
}
}
}
private void setVarPoolValue(Map<String, Property> allProperty, Map<String, TaskInstance> allTaskInstance, TaskInstance preTaskInstance, Property thisProperty) {
//for this taskInstance all the param in this part is IN.
thisProperty.setDirect(Direct.IN);
//get the pre taskInstance Property's name
String proName = thisProperty.getProp();
//if the Previous nodes have the Property of same name
if (allProperty.containsKey(proName)) {
//comparison the value of two Property
Property otherPro = allProperty.get(proName);
//if this property'value of loop is empty,use the other,whether the other's value is empty or not
if (StringUtils.isEmpty(thisProperty.getValue())) {
allProperty.put(proName, otherPro);
//if property'value of loop is not empty,and the other's value is not empty too, use the earlier value
} else if (StringUtils.isNotEmpty(otherPro.getValue())) {
TaskInstance otherTask = allTaskInstance.get(proName);
if (otherTask.getEndTime().getTime() > preTaskInstance.getEndTime().getTime()) {
allProperty.put(proName, thisProperty);
allTaskInstance.put(proName,preTaskInstance);
} else {
allProperty.put(proName, otherPro);
}
} else {
allProperty.put(proName, thisProperty);
allTaskInstance.put(proName,preTaskInstance);
}
} else {
allProperty.put(proName, thisProperty);
allTaskInstance.put(proName,preTaskInstance);
}
return JSONUtils.toJsonString(result);
}
private void submitPostNode(String parentNodeName) {
Set<String> submitTaskNodeList = DagHelper.parsePostNodes(parentNodeName, skipTaskNodeList, dag, completeTaskList);
List<TaskInstance> taskInstances = new ArrayList<>();
for (String taskNode : submitTaskNodeList) {
try {
VarPoolUtils.convertVarPoolToMap(propToValue, processInstance.getVarPool());
} catch (ParseException e) {
logger.error("parse {} exception", processInstance.getVarPool(), e);
throw new RuntimeException();
}
TaskNode taskNodeObject = dag.getNode(taskNode);
taskInstances.add(createTaskInstance(processInstance, taskNodeObject));
}

8
dolphinscheduler-server/src/main/java/org/apache/dolphinscheduler/server/utils/ParamUtils.java

@ -47,6 +47,7 @@ public class ParamUtils {
public static Map<String,Property> convert(Map<String,Property> globalParams,
Map<String,String> globalParamsMap,
Map<String,Property> localParams,
Map<String,Property> varParams,
CommandType commandType,
Date scheduleTime) {
if (globalParams == null && localParams == null) {
@ -64,10 +65,15 @@ public class ParamUtils {
}
if (globalParams != null && localParams != null) {
globalParams.putAll(localParams);
localParams.putAll(globalParams);
globalParams = localParams;
} else if (globalParams == null && localParams != null) {
globalParams = localParams;
}
if (varParams != null) {
varParams.putAll(globalParams);
globalParams = varParams;
}
Iterator<Map.Entry<String, Property>> iter = globalParams.entrySet().iterator();
while (iter.hasNext()) {
Map.Entry<String, Property> en = iter.next();

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

@ -151,10 +151,10 @@ public class TaskExecuteThread implements Runnable, Delayed {
taskExecutionContext.getTaskInstanceId()));
task = TaskManager.newTask(taskExecutionContext, taskLogger, alertClientService);
// task init
task.init();
//init varPool
task.getParameters().setVarPool(taskExecutionContext.getVarPool());
// task handle
task.handle();
@ -165,8 +165,7 @@ public class TaskExecuteThread implements Runnable, Delayed {
responseCommand.setEndTime(new Date());
responseCommand.setProcessId(task.getProcessId());
responseCommand.setAppIds(task.getAppIds());
responseCommand.setVarPool(task.getVarPool());
responseCommand.setResult(task.getResultString());
responseCommand.setVarPool(JSONUtils.toJsonString(task.getParameters().getVarPool()));
logger.info("task instance id : {},task final status : {}", taskExecutionContext.getTaskInstanceId(), task.getExitStatus());
} catch (Exception e) {
logger.error("task scheduler failure", e);

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

@ -88,11 +88,6 @@ public abstract class AbstractCommandExecutor {
protected boolean logOutputIsScuccess = false;
/**
* SHELL result string
*/
protected String taskResultString;
/**
* taskExecutionContext
*/
@ -365,7 +360,6 @@ public abstract class AbstractCommandExecutor {
varPool.append("$VarPool$");
} else {
logBuffer.add(line);
taskResultString = line;
}
}
} catch (Exception e) {
@ -593,11 +587,4 @@ public abstract class AbstractCommandExecutor {
protected abstract void createCommandFileIfNotExists(String execCommand, String commandFile) throws IOException;
public String getTaskResultString() {
return taskResultString;
}
public void setTaskResultString(String taskResultString) {
this.taskResultString = taskResultString;
}
}

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

@ -35,11 +35,6 @@ import org.slf4j.Logger;
*/
public abstract class AbstractTask {
/**
* varPool string
*/
protected String varPool;
/**
* taskExecutionContext
**/
@ -56,11 +51,6 @@ public abstract class AbstractTask {
*/
protected int processId;
/**
* SHELL result string
*/
protected String resultString;
/**
* other resource manager appId , for example : YARN etc
*/
@ -81,7 +71,7 @@ public abstract class AbstractTask {
* constructor
*
* @param taskExecutionContext taskExecutionContext
* @param logger logger
* @param logger logger
*/
protected AbstractTask(TaskExecutionContext taskExecutionContext, Logger logger) {
this.taskExecutionContext = taskExecutionContext;
@ -139,14 +129,6 @@ public abstract class AbstractTask {
}
}
public void setVarPool(String varPool) {
this.varPool = varPool;
}
public String getVarPool() {
return varPool;
}
/**
* get exit status code
*
@ -176,14 +158,6 @@ public abstract class AbstractTask {
this.processId = processId;
}
public String getResultString() {
return resultString;
}
public void setResultString(String resultString) {
this.resultString = resultString;
}
/**
* get task parameters
*

1
dolphinscheduler-server/src/main/java/org/apache/dolphinscheduler/server/worker/task/datax/DataxTask.java

@ -158,6 +158,7 @@ public class DataxTask extends AbstractTask {
Map<String, Property> paramsMap = ParamUtils.convert(ParamUtils.getUserDefParamsMap(taskExecutionContext.getDefinedParams()),
taskExecutionContext.getDefinedParams(),
dataXParameters.getLocalParametersMap(),
dataXParameters.getVarPoolMap(),
CommandType.of(taskExecutionContext.getCmdTypeIfComplement()),
taskExecutionContext.getScheduleTime());

1
dolphinscheduler-server/src/main/java/org/apache/dolphinscheduler/server/worker/task/flink/FlinkTask.java

@ -84,6 +84,7 @@ public class FlinkTask extends AbstractYarnTask {
Map<String, Property> paramsMap = ParamUtils.convert(ParamUtils.getUserDefParamsMap(taskExecutionContext.getDefinedParams()),
taskExecutionContext.getDefinedParams(),
flinkParameters.getLocalParametersMap(),
flinkParameters.getVarPoolMap(),
CommandType.of(taskExecutionContext.getCmdTypeIfComplement()),
taskExecutionContext.getScheduleTime());

1
dolphinscheduler-server/src/main/java/org/apache/dolphinscheduler/server/worker/task/http/HttpTask.java

@ -135,6 +135,7 @@ public class HttpTask extends AbstractTask {
Map<String, Property> paramsMap = ParamUtils.convert(ParamUtils.getUserDefParamsMap(taskExecutionContext.getDefinedParams()),
taskExecutionContext.getDefinedParams(),
httpParameters.getLocalParametersMap(),
httpParameters.getVarPoolMap(),
CommandType.of(taskExecutionContext.getCmdTypeIfComplement()),
taskExecutionContext.getScheduleTime());
List<HttpProperty> httpPropertyList = new ArrayList<>();

1
dolphinscheduler-server/src/main/java/org/apache/dolphinscheduler/server/worker/task/mr/MapReduceTask.java

@ -88,6 +88,7 @@ public class MapReduceTask extends AbstractYarnTask {
Map<String, Property> paramsMap = ParamUtils.convert(ParamUtils.getUserDefParamsMap(taskExecutionContext.getDefinedParams()),
taskExecutionContext.getDefinedParams(),
mapreduceParameters.getLocalParametersMap(),
mapreduceParameters.getVarPoolMap(),
CommandType.of(taskExecutionContext.getCmdTypeIfComplement()),
taskExecutionContext.getScheduleTime());

1
dolphinscheduler-server/src/main/java/org/apache/dolphinscheduler/server/worker/task/procedure/ProcedureTask.java

@ -122,6 +122,7 @@ public class ProcedureTask extends AbstractTask {
Map<String, Property> paramsMap = ParamUtils.convert(ParamUtils.getUserDefParamsMap(taskExecutionContext.getDefinedParams()),
taskExecutionContext.getDefinedParams(),
procedureParameters.getLocalParametersMap(),
procedureParameters.getVarPoolMap(),
CommandType.of(taskExecutionContext.getCmdTypeIfComplement()),
taskExecutionContext.getScheduleTime());

3
dolphinscheduler-server/src/main/java/org/apache/dolphinscheduler/server/worker/task/python/PythonTask.java

@ -92,7 +92,7 @@ public class PythonTask extends AbstractTask {
setExitStatusCode(commandExecuteResult.getExitStatusCode());
setAppIds(commandExecuteResult.getAppIds());
setProcessId(commandExecuteResult.getProcessId());
setVarPool(pythonCommandExecutor.getVarPool());
pythonParameters.dealOutParam(pythonCommandExecutor.getVarPool());
}
catch (Exception e) {
logger.error("python task failure", e);
@ -119,6 +119,7 @@ public class PythonTask extends AbstractTask {
Map<String, Property> paramsMap = ParamUtils.convert(ParamUtils.getUserDefParamsMap(taskExecutionContext.getDefinedParams()),
taskExecutionContext.getDefinedParams(),
pythonParameters.getLocalParametersMap(),
pythonParameters.getVarPoolMap(),
CommandType.of(taskExecutionContext.getCmdTypeIfComplement()),
taskExecutionContext.getScheduleTime());

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

@ -105,8 +105,7 @@ public class ShellTask extends AbstractTask {
setExitStatusCode(commandExecuteResult.getExitStatusCode());
setAppIds(commandExecuteResult.getAppIds());
setProcessId(commandExecuteResult.getProcessId());
setResult(shellCommandExecutor.getTaskResultString());
setVarPool(shellCommandExecutor.getVarPool());
shellParameters.dealOutParam(shellCommandExecutor.getVarPool());
} catch (Exception e) {
logger.error("shell task error", e);
setExitStatusCode(Constants.EXIT_CODE_FAILURE);
@ -169,6 +168,7 @@ public class ShellTask extends AbstractTask {
Map<String, Property> paramsMap = ParamUtils.convert(ParamUtils.getUserDefParamsMap(taskExecutionContext.getDefinedParams()),
taskExecutionContext.getDefinedParams(),
shellParameters.getLocalParametersMap(),
shellParameters.getVarPoolMap(),
CommandType.of(taskExecutionContext.getCmdTypeIfComplement()),
taskExecutionContext.getScheduleTime());
// replace variable TIME with $[YYYYmmddd...] in shell file when history run job and batch complement job
@ -188,17 +188,4 @@ 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);
}
}

1
dolphinscheduler-server/src/main/java/org/apache/dolphinscheduler/server/worker/task/spark/SparkTask.java

@ -113,6 +113,7 @@ public class SparkTask extends AbstractYarnTask {
Map<String, Property> paramsMap = ParamUtils.convert(ParamUtils.getUserDefParamsMap(taskExecutionContext.getDefinedParams()),
taskExecutionContext.getDefinedParams(),
sparkParameters.getLocalParametersMap(),
sparkParameters.getVarPoolMap(),
CommandType.of(taskExecutionContext.getCmdTypeIfComplement()),
taskExecutionContext.getScheduleTime());

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

@ -111,8 +111,7 @@ public class SqlTask extends AbstractTask {
Thread.currentThread().setName(threadLoggerInfoName);
logger.info("Full sql parameters: {}", sqlParameters);
logger.info("sql type : {}, datasource : {}, sql : {} , localParams : {}, udfs : {},showType : {}, "
+ "connParams : {}, query max result limit : {}",
logger.info("sql type : {}, datasource : {}, sql : {} , localParams : {},udfs : {},showType : {},connParams : {},varPool : {} ,query max result limit {}",
sqlParameters.getType(),
sqlParameters.getDatasource(),
sqlParameters.getSql(),
@ -120,6 +119,7 @@ public class SqlTask extends AbstractTask {
sqlParameters.getUdfs(),
sqlParameters.getShowType(),
sqlParameters.getConnParams(),
sqlParameters.getVarPool(),
sqlParameters.getLimit());
try {
SQLTaskExecutionContext sqlTaskExecutionContext = taskExecutionContext.getSqlTaskExecutionContext();
@ -171,6 +171,7 @@ public class SqlTask extends AbstractTask {
Map<String, Property> paramsMap = ParamUtils.convert(ParamUtils.getUserDefParamsMap(taskExecutionContext.getDefinedParams()),
taskExecutionContext.getDefinedParams(),
sqlParameters.getLocalParametersMap(),
sqlParameters.getVarPoolMap(),
CommandType.of(taskExecutionContext.getCmdTypeIfComplement()),
taskExecutionContext.getScheduleTime());
@ -264,10 +265,9 @@ public class SqlTask extends AbstractTask {
String updateResult = String.valueOf(stmt.executeUpdate());
result = setNonQuerySqlReturn(updateResult, sqlParameters.getLocalParams());
}
//deal out params
sqlParameters.dealOutParam(result);
postSql(connection, postStatementsBinds);
this.setResultString(result);
} catch (Exception e) {
logger.error("execute sql error: {}", e.getMessage());
throw e;
@ -276,6 +276,7 @@ public class SqlTask extends AbstractTask {
}
}
public String setNonQuerySqlReturn(String updateResult, List<Property> properties) {
String result = null;
for (Property info :properties) {

1
dolphinscheduler-server/src/main/java/org/apache/dolphinscheduler/server/worker/task/sqoop/SqoopTask.java

@ -76,6 +76,7 @@ public class SqoopTask extends AbstractYarnTask {
Map<String, Property> paramsMap = ParamUtils.convert(ParamUtils.getUserDefParamsMap(sqoopTaskExecutionContext.getDefinedParams()),
sqoopTaskExecutionContext.getDefinedParams(),
sqoopParameters.getLocalParametersMap(),
sqoopParameters.getVarPoolMap(),
CommandType.of(sqoopTaskExecutionContext.getCmdTypeIfComplement()),
sqoopTaskExecutionContext.getScheduleTime());

48
dolphinscheduler-server/src/test/java/org/apache/dolphinscheduler/server/master/MasterExecThreadTest.java

@ -44,10 +44,14 @@ import java.lang.reflect.Field;
import java.lang.reflect.Method;
import java.text.ParseException;
import java.util.Collections;
import java.util.Date;
import java.util.HashMap;
import java.util.HashSet;
import java.util.LinkedList;
import java.util.List;
import java.util.Map;
import java.util.Set;
import java.util.concurrent.ConcurrentHashMap;
import org.junit.Assert;
import org.junit.Before;
@ -207,6 +211,50 @@ public class MasterExecThreadTest {
}
}
@Test
public void testGetPreVarPool() {
try {
Set<String> preTaskName = new HashSet<>();
preTaskName.add("test1");
preTaskName.add("test2");
Map<String, TaskInstance> completeTaskList = new ConcurrentHashMap<>();
TaskInstance taskInstance = new TaskInstance();
TaskInstance taskInstance1 = new TaskInstance();
taskInstance1.setId(1);
taskInstance1.setName("test1");
taskInstance1.setVarPool("[{\"direct\":\"OUT\",\"prop\":\"test1\",\"type\":\"VARCHAR\",\"value\":\"1\"}]");
taskInstance1.setEndTime(new Date());
TaskInstance taskInstance2 = new TaskInstance();
taskInstance2.setId(2);
taskInstance2.setName("test2");
taskInstance2.setVarPool("[{\"direct\":\"OUT\",\"prop\":\"test2\",\"type\":\"VARCHAR\",\"value\":\"2\"}]");
taskInstance2.setEndTime(new Date());
completeTaskList.put("test1", taskInstance1);
completeTaskList.put("test2", taskInstance2);
Class<MasterExecThread> masterExecThreadClass = MasterExecThread.class;
Field field = masterExecThreadClass.getDeclaredField("completeTaskList");
field.setAccessible(true);
field.set(masterExecThread, completeTaskList);
masterExecThread.getPreVarPool(taskInstance, preTaskName);
Assert.assertNotNull(taskInstance.getVarPool());
taskInstance2.setVarPool("[{\"direct\":\"OUT\",\"prop\":\"test1\",\"type\":\"VARCHAR\",\"value\":\"2\"}]");
completeTaskList.put("test2", taskInstance2);
field.setAccessible(true);
field.set(masterExecThread, completeTaskList);
masterExecThread.getPreVarPool(taskInstance, preTaskName);
Assert.assertNotNull(taskInstance.getVarPool());
} catch (Exception e) {
Assert.fail();
}
}
private List<Schedule> zeroSchedulerList() {
return Collections.EMPTY_LIST;
}

33
dolphinscheduler-server/src/test/java/org/apache/dolphinscheduler/server/master/ParamsTest.java

@ -20,20 +20,20 @@ import org.apache.dolphinscheduler.common.enums.CommandType;
import org.apache.dolphinscheduler.common.enums.DataType;
import org.apache.dolphinscheduler.common.enums.Direct;
import org.apache.dolphinscheduler.common.process.Property;
import org.apache.dolphinscheduler.common.utils.JSONUtils;
import org.apache.dolphinscheduler.common.utils.ParameterUtils;
import org.apache.dolphinscheduler.common.utils.placeholder.BusinessTimeUtils;
import org.apache.dolphinscheduler.server.utils.ParamUtils;
import org.apache.dolphinscheduler.common.utils.*;
import org.junit.Test;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import java.util.Calendar;
import java.util.Date;
import java.util.HashMap;
import java.util.Map;
import org.junit.Test;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
/**
* user define param
@ -73,20 +73,19 @@ public class ParamsTest {
}
@Test
public void convertTest()throws Exception{
Map<String,Property> globalParams = new HashMap<>();
public void convertTest() throws Exception {
Map<String, Property> globalParams = new HashMap<>();
Property property = new Property();
property.setProp("global_param");
property.setDirect(Direct.IN);
property.setType(DataType.VARCHAR);
property.setValue("${system.biz.date}");
globalParams.put("global_param",property);
globalParams.put("global_param", property);
Map<String,String> globalParamsMap = new HashMap<>();
globalParamsMap.put("global_param","${system.biz.date}");
Map<String, String> globalParamsMap = new HashMap<>();
globalParamsMap.put("global_param", "${system.biz.date}");
Map<String,Property> localParams = new HashMap<>();
Map<String, Property> localParams = new HashMap<>();
Property localProperty = new Property();
localProperty.setProp("local_param");
localProperty.setDirect(Direct.IN);
@ -94,8 +93,16 @@ public class ParamsTest {
localProperty.setValue("${global_param}");
localParams.put("local_param", localProperty);
Map<String, Property> varPoolParams = new HashMap<>();
Property varProperty = new Property();
varProperty.setProp("local_param");
varProperty.setDirect(Direct.IN);
varProperty.setType(DataType.VARCHAR);
varProperty.setValue("${global_param}");
varPoolParams.put("varPool", varProperty);
Map<String, Property> paramsMap = ParamUtils.convert(globalParams, globalParamsMap,
localParams, CommandType.START_PROCESS, new Date());
localParams,varPoolParams, CommandType.START_PROCESS, new Date());
logger.info(JSONUtils.toJsonString(paramsMap));

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

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

42
dolphinscheduler-server/src/test/java/org/apache/dolphinscheduler/server/utils/ParamUtilsTest.java

@ -17,23 +17,24 @@
package org.apache.dolphinscheduler.server.utils;
import static org.junit.Assert.assertEquals;
import static org.junit.Assert.assertNull;
import org.apache.dolphinscheduler.common.enums.CommandType;
import org.apache.dolphinscheduler.common.enums.DataType;
import org.apache.dolphinscheduler.common.enums.Direct;
import org.apache.dolphinscheduler.common.process.Property;
import org.apache.dolphinscheduler.common.utils.*;
import org.junit.Before;
import org.junit.Test;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.apache.dolphinscheduler.common.utils.JSONUtils;
import java.util.Calendar;
import java.util.Date;
import java.util.HashMap;
import java.util.Map;
import static org.junit.Assert.assertEquals;
import static org.junit.Assert.assertNull;
import org.junit.Before;
import org.junit.Test;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
/**
* Test ParamUtils
@ -49,8 +50,11 @@ public class ParamUtilsTest {
public Map<String, Property> localParams = new HashMap<>();
public Map<String, Property> varPoolParams = new HashMap<>();
/**
* Init params
*
* @throws Exception
*/
@Before
@ -71,6 +75,14 @@ public class ParamUtilsTest {
localProperty.setType(DataType.VARCHAR);
localProperty.setValue("${global_param}");
localParams.put("local_param", localProperty);
Property varProperty = new Property();
varProperty.setProp("local_param");
varProperty.setDirect(Direct.IN);
varProperty.setType(DataType.VARCHAR);
varProperty.setValue("${global_param}");
varPoolParams.put("varPool", varProperty);
}
/**
@ -80,16 +92,20 @@ public class ParamUtilsTest {
public void testConvert() {
//The expected value
String expected = "{\"global_param\":{\"prop\":\"global_param\",\"direct\":\"IN\",\"type\":\"VARCHAR\",\"value\":\"20191229\"},\"local_param\":{\"prop\":\"local_param\",\"direct\":\"IN\",\"type\":\"VARCHAR\",\"value\":\"20191229\"}}";
String expected = "{\"varPool\":{\"prop\":\"local_param\",\"direct\":\"IN\",\"type\":\"VARCHAR\",\"value\":\"20191229\"},"
+ "\"global_param\":{\"prop\":\"global_param\",\"direct\":\"IN\",\"type\":\"VARCHAR\",\"value\":\"20191229\"},"
+ "\"local_param\":{\"prop\":\"local_param\",\"direct\":\"IN\",\"type\":\"VARCHAR\",\"value\":\"20191229\"}}";
//The expected value when globalParams is null but localParams is not null
String expected1 = "{\"local_param\":{\"prop\":\"local_param\",\"direct\":\"IN\",\"type\":\"VARCHAR\",\"value\":\"20191229\"}}";
String expected1 = "{\"varPool\":{\"prop\":\"local_param\",\"direct\":\"IN\",\"type\":\"VARCHAR\",\"value\":\"20191229\"},"
+ "\"global_param\":{\"prop\":\"global_param\",\"direct\":\"IN\",\"type\":\"VARCHAR\",\"value\":\"20191229\"},"
+ "\"local_param\":{\"prop\":\"local_param\",\"direct\":\"IN\",\"type\":\"VARCHAR\",\"value\":\"20191229\"}}";
//Define expected date , the month is 0-base
Calendar calendar = Calendar.getInstance();
calendar.set(2019,11,30);
calendar.set(2019, 11, 30);
Date date = calendar.getTime();
//Invoke convert
Map<String, Property> paramsMap = ParamUtils.convert(globalParams, globalParamsMap, localParams, CommandType.START_PROCESS, date);
Map<String, Property> paramsMap = ParamUtils.convert(globalParams, globalParamsMap, localParams, varPoolParams,CommandType.START_PROCESS, date);
String result = JSONUtils.toJsonString(paramsMap);
assertEquals(expected, result);
@ -101,12 +117,12 @@ public class ParamUtilsTest {
}
//Invoke convert with null globalParams
Map<String, Property> paramsMap1 = ParamUtils.convert(null, globalParamsMap, localParams, CommandType.START_PROCESS, date);
Map<String, Property> paramsMap1 = ParamUtils.convert(null, globalParamsMap, localParams,varPoolParams, CommandType.START_PROCESS, date);
String result1 = JSONUtils.toJsonString(paramsMap1);
assertEquals(expected1, result1);
//Null check, invoke convert with null globalParams and null localParams
Map<String, Property> paramsMap2 = ParamUtils.convert(null, globalParamsMap, null, CommandType.START_PROCESS, date);
Map<String, Property> paramsMap2 = ParamUtils.convert(null, globalParamsMap, null, varPoolParams,CommandType.START_PROCESS, date);
assertNull(paramsMap2);
}

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

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

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

@ -30,6 +30,7 @@ import org.apache.dolphinscheduler.server.entity.TaskExecutionContext;
import org.apache.dolphinscheduler.server.utils.ProcessUtils;
import org.apache.dolphinscheduler.server.worker.cache.impl.TaskExecutionContextCacheManagerImpl;
import org.apache.dolphinscheduler.server.worker.config.WorkerConfig;
import org.apache.dolphinscheduler.server.worker.runner.WorkerManagerThread;
import org.apache.dolphinscheduler.service.bean.SpringApplicationContext;
import org.apache.dolphinscheduler.service.log.LogClientService;
@ -53,6 +54,8 @@ public class TaskKillProcessorTest {
private TaskKillProcessor taskKillProcessor;
private WorkerManagerThread workerManager;
private TaskExecutionContextCacheManagerImpl taskExecutionContextCacheManager;
private Channel channel;
@ -85,6 +88,8 @@ public class TaskKillProcessorTest {
PowerMockito.mockStatic(LoggerUtils.class);
PowerMockito.when(SpringApplicationContext.getBean(TaskCallbackService.class)).thenReturn(taskCallbackService);
PowerMockito.when(SpringApplicationContext.getBean(WorkerConfig.class)).thenReturn(workerConfig);
WorkerManagerThread workerManager = PowerMockito.mock(WorkerManagerThread.class);
PowerMockito.when(SpringApplicationContext.getBean(WorkerManagerThread.class)).thenReturn(workerManager);
PowerMockito.when(SpringApplicationContext.getBean(TaskExecutionContextCacheManagerImpl.class)).thenReturn(taskExecutionContextCacheManager);
PowerMockito.doNothing().when(taskCallbackService).addRemoteChannel(anyInt(), any());
PowerMockito.whenNew(NettyRemoteChannel.class).withAnyArguments().thenReturn(null);
@ -102,7 +107,6 @@ public class TaskKillProcessorTest {
@Test
public void testProcess() {
PowerMockito.when(taskExecutionContextCacheManager.getByTaskInstanceId(1)).thenReturn(taskExecutionContext);
taskKillProcessor.process(channel, command);

3
dolphinscheduler-server/src/test/java/org/apache/dolphinscheduler/server/worker/runner/TaskExecuteThreadTest.java

@ -21,6 +21,7 @@ import org.apache.dolphinscheduler.common.enums.ExecutionStatus;
import org.apache.dolphinscheduler.common.enums.TaskType;
import org.apache.dolphinscheduler.common.model.TaskNode;
import org.apache.dolphinscheduler.common.task.AbstractParameters;
import org.apache.dolphinscheduler.common.task.sql.SqlParameters;
import org.apache.dolphinscheduler.common.utils.CommonUtils;
import org.apache.dolphinscheduler.common.utils.JSONUtils;
import org.apache.dolphinscheduler.common.utils.LoggerUtils;
@ -165,7 +166,7 @@ public class TaskExecuteThreadTest {
@Override
public AbstractParameters getParameters() {
return null;
return new SqlParameters();
}
@Override

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

@ -1,53 +0,0 @@
/*
* 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());
}
}

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

@ -110,17 +110,6 @@ public class ShellTaskReturnTest {
} 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());
}
}

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

@ -164,11 +164,6 @@ public class TaskManagerTest {
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

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

@ -0,0 +1,77 @@
/*
* 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.junit.Assert.assertNotNull;
import org.apache.dolphinscheduler.common.enums.DataType;
import org.apache.dolphinscheduler.common.enums.Direct;
import org.apache.dolphinscheduler.common.process.Property;
import org.apache.dolphinscheduler.common.task.shell.ShellParameters;
import org.apache.dolphinscheduler.common.task.sql.SqlParameters;
import java.util.ArrayList;
import java.util.List;
import org.junit.Test;
import org.junit.runner.RunWith;
import org.powermock.modules.junit4.PowerMockRunner;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
/**
* shell task return test.
*/
@RunWith(PowerMockRunner.class)
public class TaskParamsTest {
private static final Logger logger = LoggerFactory.getLogger(TaskParamsTest.class);
@Test
public void testDealOutParam() {
List<Property> properties = new ArrayList<>();
Property property = new Property();
property.setProp("test1");
property.setDirect(Direct.OUT);
property.setType(DataType.VARCHAR);
property.setValue("test1");
properties.add(property);
ShellParameters shellParameters = new ShellParameters();
String resultShell = "key1=value1$VarPoolkey2=value2";
shellParameters.varPool = new ArrayList<>();
shellParameters.setLocalParams(properties);
shellParameters.dealOutParam(resultShell);
assertNotNull(shellParameters.getVarPool().get(0));
String sqlResult = "[{\"id\":6,\"test1\":\"6\"},{\"id\":70002,\"test1\":\"+1\"}]";
SqlParameters sqlParameters = new SqlParameters();
String sqlResult1 = "[{\"id\":6,\"test1\":\"6\"}]";
sqlParameters.setLocalParams(properties);
sqlParameters.varPool = new ArrayList<>();
sqlParameters.dealOutParam(sqlResult1);
assertNotNull(sqlParameters.getVarPool().get(0));
property.setType(DataType.LIST);
properties.clear();
properties.add(property);
sqlParameters.setLocalParams(properties);
sqlParameters.dealOutParam(sqlResult);
assertNotNull(sqlParameters.getVarPool().get(0));
}
}

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

@ -62,7 +62,6 @@ 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");
@ -85,6 +84,7 @@ public class ShellTaskTest {
taskExecutionContext.setTenantCode("roo");
taskExecutionContext.setScheduleTime(new Date());
taskExecutionContext.setQueue("default");
taskExecutionContext.setVarPool("[{\"direct\":\"IN\",\"prop\":\"test\",\"type\":\"VARCHAR\",\"value\":\"\"}]");
taskExecutionContext.setTaskParams(
"{\"rawScript\":\"#!/bin/sh\\necho $[yyyy-MM-dd HH:mm:ss +3]\\necho \\\" ?? ${time1} \\\"\\necho \\\" ????? ${time2}\\\"\\n\",\"localParams\":"
+
@ -105,6 +105,7 @@ public class ShellTaskTest {
public void testComplementData() throws Exception {
shellTask = new ShellTask(taskExecutionContext, logger);
shellTask.init();
shellTask.getParameters().setVarPool(taskExecutionContext.getVarPool());
shellCommandExecutor.isSuccessOfYarnState(new ArrayList<>());
shellCommandExecutor.isSuccessOfYarnState(null);
PowerMockito.when(shellCommandExecutor.run(anyString())).thenReturn(commandExecuteResult);
@ -116,16 +117,9 @@ public class ShellTaskTest {
taskExecutionContext.setCmdTypeIfComplement(0);
shellTask = new ShellTask(taskExecutionContext, logger);
shellTask.init();
shellTask.getParameters().setVarPool(taskExecutionContext.getVarPool());
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);
}
}

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

@ -97,6 +97,7 @@ public class SqlTaskTest {
PowerMockito.when(taskExecutionContext.getStartTime()).thenReturn(new Date());
PowerMockito.when(taskExecutionContext.getTaskTimeout()).thenReturn(10000);
PowerMockito.when(taskExecutionContext.getLogPath()).thenReturn("/tmp/dx");
PowerMockito.when(taskExecutionContext.getVarPool()).thenReturn("[{\"direct\":\"IN\",\"prop\":\"test\",\"type\":\"VARCHAR\",\"value\":\"\"}]");
SQLTaskExecutionContext sqlTaskExecutionContext = new SQLTaskExecutionContext();
sqlTaskExecutionContext.setConnectionParams(CONNECTION_PARAMS);
@ -106,6 +107,7 @@ public class SqlTaskTest {
PowerMockito.when(SpringApplicationContext.getBean(Mockito.any())).thenReturn(new AlertDao());
alertClientService = PowerMockito.mock(AlertClientService.class);
sqlTask = new SqlTask(taskExecutionContext, logger, alertClientService);
sqlTask.getParameters().setVarPool(taskExecutionContext.getVarPool());
sqlTask.init();
}

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

@ -133,7 +133,6 @@ 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;
/**
@ -1584,71 +1583,51 @@ public class ProcessService {
int processId,
String appIds,
int taskInstId,
String varPool,
String result) {
String varPool) {
taskInstance.setPid(processId);
taskInstance.setAppLink(appIds);
taskInstance.setState(state);
taskInstance.setEndTime(endTime);
taskInstance.setVarPool(varPool);
changeOutParam(result, taskInstance);
changeOutParam(taskInstance);
saveTaskInstance(taskInstance);
}
public void changeOutParam(String result, TaskInstance taskInstance) {
if (StringUtils.isEmpty(result)) {
/**
* for show in page of taskInstance
* @param taskInstance
*/
public void changeOutParam(TaskInstance taskInstance) {
if (StringUtils.isEmpty(taskInstance.getVarPool())) {
return;
}
List<Map<String, String>> workerResultParam = getListMapByString(result);
if (CollectionUtils.isEmpty(workerResultParam)) {
List<Property> properties = JSONUtils.toList(taskInstance.getVarPool(), Property.class);
if (CollectionUtils.isEmpty(properties)) {
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;
}
Map<String, Object> taskParams = JSONUtils.toMap(taskInstance.getTaskParams(), 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);
Map<String, String> outProperty = new HashMap<>();
for (Property info : properties) {
if (info.getDirect() == Direct.OUT) {
outProperty.put(info.getProp(), info.getValue());
}
}
for (Property info : allParam) {
if (info.getDirect() == Direct.OUT) {
String paramName = info.getProp();
Property property = allParamMap.get(paramName);
if (property == null) {
continue;
}
String value = String.valueOf(row.get(paramName));
if (StringUtils.isNotEmpty(value)) {
property.setValue(value);
info.setValue(value);
}
info.setValue(outProperty.get(paramName));
}
}
taskParams.put(LOCAL_PARAMS, allParam);
taskInstance.setTaskParams(JSONUtils.toJsonString(taskParams));
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

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

@ -509,4 +509,19 @@ public class ProcessServiceTest {
Mockito.verify(commandMapper, Mockito.times(1)).insert(command);
}
@Test
public void testChangeOutParam() {
TaskInstance taskInstance = new TaskInstance();
taskInstance.setProcessInstanceId(62);
ProcessInstance processInstance = new ProcessInstance();
processInstance.setId(62);
taskInstance.setVarPool("[{\"direct\":\"OUT\",\"prop\":\"test1\",\"type\":\"VARCHAR\",\"value\":\"\"}]");
taskInstance.setTaskParams("{\"type\":\"MYSQL\",\"datasource\":1,\"sql\":\"select id from tb_test limit 1\","
+ "\"udfs\":\"\",\"sqlType\":\"0\",\"sendEmail\":false,\"displayRows\":10,\"title\":\"\","
+ "\"groupId\":null,\"localParams\":[{\"prop\":\"test1\",\"direct\":\"OUT\",\"type\":\"VARCHAR\",\"value\":\"12\"}],"
+ "\"connParams\":\"\",\"preStatements\":[],\"postStatements\":[],\"conditionResult\":\"{\\\"successNode\\\":[\\\"\\\"],"
+ "\\\"failedNode\\\":[\\\"\\\"]}\",\"dependence\":\"{}\"}");
processService.changeOutParam(taskInstance);
}
}

2
pom.xml

@ -1009,8 +1009,8 @@
<include>**/server/worker/task/processdure/ProcedureTaskTest.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/PythonCommandExecutorTest.java</include>
<include>**/server/worker/task/TaskParamsTest.java</include>
<include>**/server/worker/task/ShellTaskReturnTest.java</include>
<include>**/server/worker/task/sql/SqlTaskTest.java</include>
<include>**/server/worker/runner/TaskExecuteThreadTest.java</include>

Loading…
Cancel
Save