This is an automated email from the ASF dual-hosted git repository.
wenjun pushed a commit to branch dev
in repository https://gitbox.apache.org/repos/asf/dolphinscheduler.git
The following commit(s) were added to refs/heads/dev by this push:
new 6f12d1ded9 [Fix-17184] Fix varpool cannot use (#17199)
6f12d1ded9 is described below
commit 6f12d1ded97a14eaf253a5654486e0ee3c541dfe
Author: Wenjun Ruan <[email protected]>
AuthorDate: Tue Jun 3 22:16:50 2025 +0800
[Fix-17184] Fix varpool cannot use (#17199)
---
.../plugin/dependent/DependentTaskTracker.java | 3 +-
.../engine/executor/plugin/fake/LogicFakeTask.java | 62 ++++-
.../WorkerGroupDispatcherCoordinator.java | 9 +-
.../lifecycle/event/TaskKillLifecycleEvent.java | 3 +-
.../lifecycle/event/TaskPauseLifecycleEvent.java | 3 +-
.../lifecycle/event/TaskSuccessLifecycleEvent.java | 4 +-
.../task/runnable/ITaskExecutionRunnable.java | 6 +
.../task/runnable/TaskExecutionContextBuilder.java | 1 -
.../TaskExecutionContextCreateRequest.java | 2 +
.../task/runnable/TaskExecutionRunnable.java | 18 +-
.../task/statemachine/AbstractTaskStateAction.java | 12 +-
.../statemachine/TaskSubmittedStateAction.java | 1 +
.../task/statemachine/TaskSuccessStateAction.java | 3 +-
.../statemachine/IWorkflowStateAction.java | 2 +-
.../master/runner/TaskExecutionContextFactory.java | 29 +++
.../cases/WorkflowInstanceRecoverStopTestCase.java | 7 +-
.../integration/cases/WorkflowStartTestCase.java | 54 ++++
.../src/test/resources/application.yaml | 2 +-
...flow_with_local_param_overwrite_by_varpool.yaml | 137 ++++++++++
.../service/expand/CuringParamsService.java | 16 --
.../service/expand/CuringParamsServiceImpl.java | 80 +++---
.../TimePlaceholderResolverExpandService.java | 35 ---
.../TimePlaceholderResolverExpandServiceImpl.java | 34 ---
.../service/expand/CuringParamsServiceTest.java | 18 --
.../TimePlaceholderResolverExpandServiceTest.java | 53 ----
.../events/TaskExecutorSuccessLifecycleEvent.java | 5 +-
.../plugin/task/api/TaskExecutionContext.java | 5 +-
.../plugin/task/api/model/Property.java | 29 ---
.../task/api/parameters/AbstractParameters.java | 24 +-
.../plugin/task/api/utils/VarPoolUtils.java | 26 +-
.../plugin/task/api/utils/VarPoolUtilsTest.java | 5 +-
.../plugin/task/python/PythonTask.java | 2 +-
.../plugin/task/shell/ShellTask.java | 1 +
.../worker/executor/PhysicalTaskExecutor.java | 11 +-
.../worker/utils/TaskFilesTransferUtils.java | 285 ---------------------
.../worker/utils/TaskFilesTransferUtilsTest.java | 255 ------------------
36 files changed, 393 insertions(+), 849 deletions(-)
diff --git
a/dolphinscheduler-master/src/main/java/org/apache/dolphinscheduler/server/master/engine/executor/plugin/dependent/DependentTaskTracker.java
b/dolphinscheduler-master/src/main/java/org/apache/dolphinscheduler/server/master/engine/executor/plugin/dependent/DependentTaskTracker.java
index c9572a99ad..5fbfb86569 100644
---
a/dolphinscheduler-master/src/main/java/org/apache/dolphinscheduler/server/master/engine/executor/plugin/dependent/DependentTaskTracker.java
+++
b/dolphinscheduler-master/src/main/java/org/apache/dolphinscheduler/server/master/engine/executor/plugin/dependent/DependentTaskTracker.java
@@ -19,7 +19,6 @@ package
org.apache.dolphinscheduler.server.master.engine.executor.plugin.depende
import org.apache.dolphinscheduler.common.constants.Constants;
import org.apache.dolphinscheduler.common.enums.ContextType;
-import org.apache.dolphinscheduler.common.utils.JSONUtils;
import
org.apache.dolphinscheduler.dao.entity.DependentResultTaskInstanceContext;
import org.apache.dolphinscheduler.dao.entity.Project;
import org.apache.dolphinscheduler.dao.entity.TaskDefinition;
@@ -111,7 +110,7 @@ public class DependentTaskTracker {
DependResult dependResult = calculateDependResult();
log.info("The final Dependent result is: {}", dependResult);
if (dependResult == DependResult.SUCCESS) {
-
dependentParameters.setVarPool(JSONUtils.toJsonString(dependVarPoolPropertyMap.values()));
+ dependentParameters.setVarPool(new
ArrayList<>(dependVarPoolPropertyMap.values()));
log.info("Set dependentParameters varPool: {}",
dependentParameters.getVarPool());
return TaskExecutionStatus.SUCCESS;
} else {
diff --git
a/dolphinscheduler-master/src/main/java/org/apache/dolphinscheduler/server/master/engine/executor/plugin/fake/LogicFakeTask.java
b/dolphinscheduler-master/src/main/java/org/apache/dolphinscheduler/server/master/engine/executor/plugin/fake/LogicFakeTask.java
index 409d529761..f0a62a9858 100644
---
a/dolphinscheduler-master/src/main/java/org/apache/dolphinscheduler/server/master/engine/executor/plugin/fake/LogicFakeTask.java
+++
b/dolphinscheduler-master/src/main/java/org/apache/dolphinscheduler/server/master/engine/executor/plugin/fake/LogicFakeTask.java
@@ -17,10 +17,12 @@
package org.apache.dolphinscheduler.server.master.engine.executor.plugin.fake;
+import org.apache.dolphinscheduler.common.thread.ThreadUtils;
import org.apache.dolphinscheduler.common.utils.JSONUtils;
import org.apache.dolphinscheduler.plugin.task.api.TaskExecutionContext;
import org.apache.dolphinscheduler.plugin.task.api.enums.TaskExecutionStatus;
import
org.apache.dolphinscheduler.plugin.task.api.parameters.LogicFakeTaskParameters;
+import
org.apache.dolphinscheduler.plugin.task.api.parser.TaskOutputParameterParser;
import org.apache.dolphinscheduler.plugin.task.api.utils.ParameterUtils;
import
org.apache.dolphinscheduler.server.master.engine.executor.plugin.AbstractLogicTask;
import
org.apache.dolphinscheduler.server.master.engine.executor.plugin.ITaskParameterDeserializer;
@@ -28,6 +30,14 @@ import
org.apache.dolphinscheduler.server.master.exception.MasterTaskExecuteExce
import org.apache.commons.lang3.StringUtils;
+import java.io.BufferedReader;
+import java.io.InputStreamReader;
+import java.util.Map;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Future;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.TimeoutException;
+
import lombok.extern.slf4j.Slf4j;
import com.fasterxml.jackson.core.type.TypeReference;
@@ -41,7 +51,7 @@ import com.google.common.annotations.VisibleForTesting;
@VisibleForTesting
public class LogicFakeTask extends AbstractLogicTask<LogicFakeTaskParameters> {
- private Process process;
+ private volatile Process process;
public LogicFakeTask(final TaskExecutionContext taskExecutionContext) {
super(taskExecutionContext);
@@ -60,22 +70,33 @@ public class LogicFakeTask extends
AbstractLogicTask<LogicFakeTaskParameters> {
if
(StringUtils.isNotEmpty(taskExecutionContext.getEnvironmentConfig())) {
shellScript = taskExecutionContext.getEnvironmentConfig() +
"\n" + shellScript;
}
- final String[] cmd = {"/bin/sh", "-c", shellScript};
- process = Runtime.getRuntime().exec(cmd);
+ ProcessBuilder processBuilder = new ProcessBuilder("/bin/sh",
"-c", shellScript);
+ processBuilder.redirectErrorStream(true);
+ process = processBuilder.start();
+ final Future<Map<String, String>> parseVarPoolFuture =
parseVarPool();
int exitCode = process.waitFor();
- if (taskExecutionStatus != TaskExecutionStatus.RUNNING_EXECUTION) {
+ log.info("LogicFakeTask: {} execute finished with exit code: {}",
+ taskExecutionContext.getTaskName(),
+ exitCode);
+ if (taskExecutionStatus == TaskExecutionStatus.KILL) {
+ try {
+ parseVarPoolFuture.get(1, TimeUnit.SECONDS);
+ } catch (TimeoutException interruptedException) {
+ // ignore
+ }
// The task has been killed
+ log.info("LogicFakeTask: {} has been killed",
taskExecutionContext.getTaskName());
return;
}
+
+ final Map<String, String> taskOutputParams =
parseVarPoolFuture.get();
if (exitCode == 0) {
- log.info("LogicFakeTask: {} execute success with exit code:
{}",
- taskExecutionContext.getTaskName(),
- exitCode);
+ log.info("LogicFakeTask: {} execute success",
taskExecutionContext.getTaskName());
+ taskParameters.dealOutParam(taskOutputParams);
+ taskExecutionContext.setVarPool(taskParameters.getVarPool());
onTaskSuccess();
} else {
- log.info("LogicFakeTask: {} execute failed with exit code: {}",
- taskExecutionContext.getTaskName(),
- exitCode);
+ log.info("LogicFakeTask: {} execute failed",
taskExecutionContext.getTaskName());
onTaskFailed();
}
} catch (Exception ex) {
@@ -104,4 +125,25 @@ public class LogicFakeTask extends
AbstractLogicTask<LogicFakeTaskParameters> {
});
}
+ private Future<Map<String, String>> parseVarPool() {
+ ExecutorService varPoolParseThreadPool =
ThreadUtils.newSingleDaemonScheduledExecutorService(
+ "ResolveOutputLog-thread-" +
taskExecutionContext.getTaskName());
+ Future<Map<String, String>> future = varPoolParseThreadPool.submit(()
-> {
+ TaskOutputParameterParser taskOutputParameterParser = new
TaskOutputParameterParser();
+ try (BufferedReader inReader = new BufferedReader(new
InputStreamReader(process.getInputStream()))) {
+ String line;
+ while ((line = inReader.readLine()) != null) {
+ log.info(line);
+ taskOutputParameterParser.appendParseLog(line);
+ }
+ } catch (Exception e) {
+ log.error("Parse var pool error", e);
+ }
+ return taskOutputParameterParser.getTaskOutputParams();
+ });
+
+ varPoolParseThreadPool.shutdown();
+ return future;
+ }
+
}
diff --git
a/dolphinscheduler-master/src/main/java/org/apache/dolphinscheduler/server/master/engine/task/dispatcher/WorkerGroupDispatcherCoordinator.java
b/dolphinscheduler-master/src/main/java/org/apache/dolphinscheduler/server/master/engine/task/dispatcher/WorkerGroupDispatcherCoordinator.java
index 1683b86d5c..a85674c6f4 100644
---
a/dolphinscheduler-master/src/main/java/org/apache/dolphinscheduler/server/master/engine/task/dispatcher/WorkerGroupDispatcherCoordinator.java
+++
b/dolphinscheduler-master/src/main/java/org/apache/dolphinscheduler/server/master/engine/task/dispatcher/WorkerGroupDispatcherCoordinator.java
@@ -56,7 +56,8 @@ public class WorkerGroupDispatcherCoordinator implements
AutoCloseable {
final long delayTimeMills) {
final String workerGroup =
taskExecutionRunnable.getTaskInstance().getWorkerGroup();
getOrCreateWorkerGroupDispatcher(workerGroup).dispatchTask(taskExecutionRunnable,
delayTimeMills);
- log.info("Success add Task: {} to WorkerGroupDispatcher: {}",
taskExecutionRunnable.getId(), workerGroup);
+ log.info("Success add Task[id={}] to WorkerGroupDispatcher[name={}]",
taskExecutionRunnable.getId(),
+ workerGroup);
}
/**
@@ -67,10 +68,10 @@ public class WorkerGroupDispatcherCoordinator implements
AutoCloseable {
final String workerGroup =
taskExecutionRunnable.getTaskInstance().getWorkerGroup();
boolean removed =
getOrCreateWorkerGroupDispatcher(workerGroup).removeTask(taskExecutionRunnable);
if (removed) {
- log.info("Success removed Task: {} from WorkerGroupDispatcher: {}",
+ log.info("Success removed Task[id={}] from
WorkerGroupDispatcher[name={}]",
taskExecutionRunnable.getId(), workerGroup);
} else {
- log.info("Failed to remove Task: {} from WorkerGroupDispatcher:
{}, this task has been dispatched",
+ log.info("Failed to remove Task[id={}] from
WorkerGroupDispatcher[name={}], this task has been dispatched",
taskExecutionRunnable.getId(), workerGroup);
}
return removed;
@@ -90,7 +91,7 @@ public class WorkerGroupDispatcherCoordinator implements
AutoCloseable {
try {
workerGroupDispatcher.close();
} catch (Exception e) {
- log.error("close WorkerGroupDispatcher: {} error",
workerGroupDispatcher.getName(), e);
+ log.error("close WorkerGroupDispatcher[name={}] error",
workerGroupDispatcher.getName(), e);
}
}
log.info("WorkerGroupDispatcherCoordinator closed...");
diff --git
a/dolphinscheduler-master/src/main/java/org/apache/dolphinscheduler/server/master/engine/task/lifecycle/event/TaskKillLifecycleEvent.java
b/dolphinscheduler-master/src/main/java/org/apache/dolphinscheduler/server/master/engine/task/lifecycle/event/TaskKillLifecycleEvent.java
index 568c6f5a18..e7275ce874 100644
---
a/dolphinscheduler-master/src/main/java/org/apache/dolphinscheduler/server/master/engine/task/lifecycle/event/TaskKillLifecycleEvent.java
+++
b/dolphinscheduler-master/src/main/java/org/apache/dolphinscheduler/server/master/engine/task/lifecycle/event/TaskKillLifecycleEvent.java
@@ -50,7 +50,8 @@ public class TaskKillLifecycleEvent extends
AbstractTaskLifecycleEvent {
@Override
public String toString() {
return "TaskKillLifecycleEvent{" +
- "task=" + taskExecutionRunnable.getName() +
+ "task=" + taskExecutionRunnable.getName() + ", " +
+ "delayTime=" + delayTime +
'}';
}
}
diff --git
a/dolphinscheduler-master/src/main/java/org/apache/dolphinscheduler/server/master/engine/task/lifecycle/event/TaskPauseLifecycleEvent.java
b/dolphinscheduler-master/src/main/java/org/apache/dolphinscheduler/server/master/engine/task/lifecycle/event/TaskPauseLifecycleEvent.java
index beb8d724c7..afa7b58c08 100644
---
a/dolphinscheduler-master/src/main/java/org/apache/dolphinscheduler/server/master/engine/task/lifecycle/event/TaskPauseLifecycleEvent.java
+++
b/dolphinscheduler-master/src/main/java/org/apache/dolphinscheduler/server/master/engine/task/lifecycle/event/TaskPauseLifecycleEvent.java
@@ -50,7 +50,8 @@ public class TaskPauseLifecycleEvent extends
AbstractTaskLifecycleEvent {
@Override
public String toString() {
return "TaskPauseLifecycleEvent{" +
- "task=" + taskExecutionRunnable.getName() +
+ "task=" + taskExecutionRunnable.getName() + ", " +
+ "delayTime=" + delayTime +
'}';
}
}
diff --git
a/dolphinscheduler-master/src/main/java/org/apache/dolphinscheduler/server/master/engine/task/lifecycle/event/TaskSuccessLifecycleEvent.java
b/dolphinscheduler-master/src/main/java/org/apache/dolphinscheduler/server/master/engine/task/lifecycle/event/TaskSuccessLifecycleEvent.java
index ccb29180f3..542943428e 100644
---
a/dolphinscheduler-master/src/main/java/org/apache/dolphinscheduler/server/master/engine/task/lifecycle/event/TaskSuccessLifecycleEvent.java
+++
b/dolphinscheduler-master/src/main/java/org/apache/dolphinscheduler/server/master/engine/task/lifecycle/event/TaskSuccessLifecycleEvent.java
@@ -17,12 +17,14 @@
package org.apache.dolphinscheduler.server.master.engine.task.lifecycle.event;
+import org.apache.dolphinscheduler.plugin.task.api.model.Property;
import org.apache.dolphinscheduler.server.master.engine.ILifecycleEventType;
import
org.apache.dolphinscheduler.server.master.engine.task.lifecycle.AbstractTaskLifecycleEvent;
import
org.apache.dolphinscheduler.server.master.engine.task.lifecycle.TaskLifecycleEventType;
import
org.apache.dolphinscheduler.server.master.engine.task.runnable.ITaskExecutionRunnable;
import java.util.Date;
+import java.util.List;
import lombok.AllArgsConstructor;
import lombok.Builder;
@@ -37,7 +39,7 @@ public class TaskSuccessLifecycleEvent extends
AbstractTaskLifecycleEvent {
private final Date endTime;
- private final String varPool;
+ private final List<Property> varPool;
@Override
public ILifecycleEventType getEventType() {
diff --git
a/dolphinscheduler-master/src/main/java/org/apache/dolphinscheduler/server/master/engine/task/runnable/ITaskExecutionRunnable.java
b/dolphinscheduler-master/src/main/java/org/apache/dolphinscheduler/server/master/engine/task/runnable/ITaskExecutionRunnable.java
index d0e0ad8d9d..2e75abffa4 100644
---
a/dolphinscheduler-master/src/main/java/org/apache/dolphinscheduler/server/master/engine/task/runnable/ITaskExecutionRunnable.java
+++
b/dolphinscheduler-master/src/main/java/org/apache/dolphinscheduler/server/master/engine/task/runnable/ITaskExecutionRunnable.java
@@ -56,6 +56,12 @@ public interface ITaskExecutionRunnable
*/
void initializeFirstRunTaskInstance();
+ /**
+ * Initialize {@link TaskExecutionContext}.
+ * <p> The TaskExecutionContext should be initialized before dispatch
stage.
+ */
+ void initializeTaskExecutionContext();
+
/**
* Whether the task instance is running.
*/
diff --git
a/dolphinscheduler-master/src/main/java/org/apache/dolphinscheduler/server/master/engine/task/runnable/TaskExecutionContextBuilder.java
b/dolphinscheduler-master/src/main/java/org/apache/dolphinscheduler/server/master/engine/task/runnable/TaskExecutionContextBuilder.java
index 883dd77702..5ec43786c0 100644
---
a/dolphinscheduler-master/src/main/java/org/apache/dolphinscheduler/server/master/engine/task/runnable/TaskExecutionContextBuilder.java
+++
b/dolphinscheduler-master/src/main/java/org/apache/dolphinscheduler/server/master/engine/task/runnable/TaskExecutionContextBuilder.java
@@ -67,7 +67,6 @@ public class TaskExecutionContextBuilder {
taskExecutionContext.setLogPath(taskInstance.getLogPath());
taskExecutionContext.setWorkerGroup(taskInstance.getWorkerGroup());
taskExecutionContext.setHost(taskInstance.getHost());
- taskExecutionContext.setVarPool(taskInstance.getVarPool());
taskExecutionContext.setDryRun(taskInstance.getDryRun());
taskExecutionContext.setTestFlag(taskInstance.getTestFlag());
taskExecutionContext.setCpuQuota(taskInstance.getCpuQuota());
diff --git
a/dolphinscheduler-master/src/main/java/org/apache/dolphinscheduler/server/master/engine/task/runnable/TaskExecutionContextCreateRequest.java
b/dolphinscheduler-master/src/main/java/org/apache/dolphinscheduler/server/master/engine/task/runnable/TaskExecutionContextCreateRequest.java
index 223e3ecb47..07e81e6856 100644
---
a/dolphinscheduler-master/src/main/java/org/apache/dolphinscheduler/server/master/engine/task/runnable/TaskExecutionContextCreateRequest.java
+++
b/dolphinscheduler-master/src/main/java/org/apache/dolphinscheduler/server/master/engine/task/runnable/TaskExecutionContextCreateRequest.java
@@ -22,6 +22,7 @@ import org.apache.dolphinscheduler.dao.entity.TaskDefinition;
import org.apache.dolphinscheduler.dao.entity.TaskInstance;
import org.apache.dolphinscheduler.dao.entity.WorkflowDefinition;
import org.apache.dolphinscheduler.dao.entity.WorkflowInstance;
+import
org.apache.dolphinscheduler.server.master.engine.graph.IWorkflowExecutionGraph;
import lombok.AllArgsConstructor;
import lombok.Builder;
@@ -32,6 +33,7 @@ import lombok.Getter;
@AllArgsConstructor
public class TaskExecutionContextCreateRequest {
+ private IWorkflowExecutionGraph workflowExecutionGraph;
private WorkflowDefinition workflowDefinition;
private WorkflowInstance workflowInstance;
private TaskDefinition taskDefinition;
diff --git
a/dolphinscheduler-master/src/main/java/org/apache/dolphinscheduler/server/master/engine/task/runnable/TaskExecutionRunnable.java
b/dolphinscheduler-master/src/main/java/org/apache/dolphinscheduler/server/master/engine/task/runnable/TaskExecutionRunnable.java
index 9f45490a72..b2e4c33cbc 100644
---
a/dolphinscheduler-master/src/main/java/org/apache/dolphinscheduler/server/master/engine/task/runnable/TaskExecutionRunnable.java
+++
b/dolphinscheduler-master/src/main/java/org/apache/dolphinscheduler/server/master/engine/task/runnable/TaskExecutionRunnable.java
@@ -72,9 +72,6 @@ public class TaskExecutionRunnable implements
ITaskExecutionRunnable {
this.workflowInstance =
checkNotNull(taskExecutionRunnableBuilder.getWorkflowInstance());
this.taskDefinition =
checkNotNull(taskExecutionRunnableBuilder.getTaskDefinition());
this.taskInstance = taskExecutionRunnableBuilder.getTaskInstance();
- if (isTaskInstanceInitialized()) {
- initializeTaskExecutionContext();
- }
}
@Override
@@ -92,7 +89,6 @@ public class TaskExecutionRunnable implements
ITaskExecutionRunnable {
.withTaskDefinition(taskDefinition)
.withWorkflowInstance(workflowInstance)
.build();
- initializeTaskExecutionContext();
}
@Override
@@ -108,7 +104,6 @@ public class TaskExecutionRunnable implements
ITaskExecutionRunnable {
.builder()
.withTaskInstance(taskInstance)
.build();
- initializeTaskExecutionContext();
getWorkflowEventBus().publish(TaskStartLifecycleEvent.of(this));
}
@@ -124,8 +119,6 @@ public class TaskExecutionRunnable implements
ITaskExecutionRunnable {
.builder()
.withTaskInstance(taskInstance)
.build();
- initializeTaskExecutionContext();
-
getWorkflowEventBus().publish(TaskStartLifecycleEvent.of(this));
}
@@ -139,17 +132,20 @@ public class TaskExecutionRunnable implements
ITaskExecutionRunnable {
getWorkflowEventBus().publish(TaskKillLifecycleEvent.of(this));
}
- private void initializeTaskExecutionContext() {
- checkState(isTaskInstanceInitialized(), "The task instance is null,
can't initialize TaskExecutionContext.");
+ @Override
+ public void initializeTaskExecutionContext() {
+ checkState(isTaskInstanceInitialized(),
+ "The task instance is not initialized, can't initialize
TaskExecutionContext.");
final TaskExecutionContextCreateRequest request =
TaskExecutionContextCreateRequest.builder()
+ .workflowExecutionGraph(workflowExecutionGraph)
.workflowDefinition(workflowDefinition)
.project(project)
.workflowInstance(workflowInstance)
.taskDefinition(taskDefinition)
.taskInstance(taskInstance)
.build();
- this.taskExecutionContext =
applicationContext.getBean(TaskExecutionContextFactory.class)
- .createTaskExecutionContext(request);
+ this.taskExecutionContext =
+
applicationContext.getBean(TaskExecutionContextFactory.class).createTaskExecutionContext(request);
}
private boolean takeOverTaskFromExecutor() {
diff --git
a/dolphinscheduler-master/src/main/java/org/apache/dolphinscheduler/server/master/engine/task/statemachine/AbstractTaskStateAction.java
b/dolphinscheduler-master/src/main/java/org/apache/dolphinscheduler/server/master/engine/task/statemachine/AbstractTaskStateAction.java
index 52bc662c43..4aa04aa213 100644
---
a/dolphinscheduler-master/src/main/java/org/apache/dolphinscheduler/server/master/engine/task/statemachine/AbstractTaskStateAction.java
+++
b/dolphinscheduler-master/src/main/java/org/apache/dolphinscheduler/server/master/engine/task/statemachine/AbstractTaskStateAction.java
@@ -20,10 +20,12 @@ package
org.apache.dolphinscheduler.server.master.engine.task.statemachine;
import static com.google.common.base.Preconditions.checkNotNull;
import static
org.apache.dolphinscheduler.plugin.task.api.enums.TaskExecutionStatus.DISPATCH;
+import org.apache.dolphinscheduler.common.utils.JSONUtils;
import org.apache.dolphinscheduler.dao.entity.TaskInstance;
import org.apache.dolphinscheduler.dao.entity.WorkflowInstance;
import org.apache.dolphinscheduler.dao.repository.TaskInstanceDao;
import org.apache.dolphinscheduler.plugin.task.api.enums.TaskExecutionStatus;
+import org.apache.dolphinscheduler.plugin.task.api.model.Property;
import org.apache.dolphinscheduler.plugin.task.api.utils.VarPoolUtils;
import org.apache.dolphinscheduler.server.master.engine.AbstractLifecycleEvent;
import org.apache.dolphinscheduler.server.master.engine.ITaskGroupCoordinator;
@@ -45,6 +47,8 @@ import
org.apache.dolphinscheduler.server.master.engine.workflow.runnable.IWorkf
import org.apache.commons.lang3.StringUtils;
+import java.util.List;
+
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
@@ -203,9 +207,9 @@ public abstract class AbstractTaskStateAction implements
ITaskStateAction {
final ITaskExecutionRunnable
taskExecutionRunnable) {
final TaskInstance taskInstance =
taskExecutionRunnable.getTaskInstance();
final WorkflowInstance workflowInstance =
workflowExecutionRunnable.getWorkflowInstance();
- final String finalVarPool = VarPoolUtils.mergeVarPoolJsonString(
+ final List<Property> finalVarPool =
VarPoolUtils.mergeVarPoolJsonString(
Lists.newArrayList(workflowInstance.getVarPool(),
taskInstance.getVarPool()));
- workflowInstance.setVarPool(finalVarPool);
+
workflowInstance.setVarPool(VarPoolUtils.serializeVarPool(finalVarPool));
}
protected void persistentTaskInstanceSuccessEventToDB(final
ITaskExecutionRunnable taskExecutionRunnable,
@@ -213,7 +217,9 @@ public abstract class AbstractTaskStateAction implements
ITaskStateAction {
final TaskInstance taskInstance =
taskExecutionRunnable.getTaskInstance();
taskInstance.setState(TaskExecutionStatus.SUCCESS);
taskInstance.setEndTime(taskSuccessEvent.getEndTime());
- taskInstance.setVarPool(taskSuccessEvent.getVarPool());
+ final List<Property> finalVarPool =
VarPoolUtils.mergeVarPoolJsonString(taskInstance.getVarPool(),
+ JSONUtils.toJsonString(taskSuccessEvent.getVarPool()));
+ taskInstance.setVarPool(VarPoolUtils.serializeVarPool(finalVarPool));
taskInstanceDao.updateById(taskInstance);
}
diff --git
a/dolphinscheduler-master/src/main/java/org/apache/dolphinscheduler/server/master/engine/task/statemachine/TaskSubmittedStateAction.java
b/dolphinscheduler-master/src/main/java/org/apache/dolphinscheduler/server/master/engine/task/statemachine/TaskSubmittedStateAction.java
index 66df54d9d7..13e2a38c85 100644
---
a/dolphinscheduler-master/src/main/java/org/apache/dolphinscheduler/server/master/engine/task/statemachine/TaskSubmittedStateAction.java
+++
b/dolphinscheduler-master/src/main/java/org/apache/dolphinscheduler/server/master/engine/task/statemachine/TaskSubmittedStateAction.java
@@ -109,6 +109,7 @@ public class TaskSubmittedStateAction extends
AbstractTaskStateAction {
taskInstance.getDelayTime(),
remainTimeMills);
}
+ taskExecutionRunnable.initializeTaskExecutionContext();
workerGroupDispatcherCoordinator.dispatchTask(taskExecutionRunnable,
remainTimeMills);
}
diff --git
a/dolphinscheduler-master/src/main/java/org/apache/dolphinscheduler/server/master/engine/task/statemachine/TaskSuccessStateAction.java
b/dolphinscheduler-master/src/main/java/org/apache/dolphinscheduler/server/master/engine/task/statemachine/TaskSuccessStateAction.java
index fa69c9a21d..a1c0f5d770 100644
---
a/dolphinscheduler-master/src/main/java/org/apache/dolphinscheduler/server/master/engine/task/statemachine/TaskSuccessStateAction.java
+++
b/dolphinscheduler-master/src/main/java/org/apache/dolphinscheduler/server/master/engine/task/statemachine/TaskSuccessStateAction.java
@@ -18,6 +18,7 @@
package org.apache.dolphinscheduler.server.master.engine.task.statemachine;
import org.apache.dolphinscheduler.plugin.task.api.enums.TaskExecutionStatus;
+import org.apache.dolphinscheduler.plugin.task.api.utils.VarPoolUtils;
import
org.apache.dolphinscheduler.server.master.engine.task.lifecycle.event.TaskDispatchLifecycleEvent;
import
org.apache.dolphinscheduler.server.master.engine.task.lifecycle.event.TaskDispatchedLifecycleEvent;
import
org.apache.dolphinscheduler.server.master.engine.task.lifecycle.event.TaskFailedLifecycleEvent;
@@ -51,7 +52,7 @@ public class TaskSuccessStateAction extends
AbstractTaskStateAction {
throwExceptionIfStateIsNotMatch(taskExecutionRunnable);
final TaskSuccessLifecycleEvent taskSuccessLifecycleEvent =
TaskSuccessLifecycleEvent.builder()
.taskExecutionRunnable(taskExecutionRunnable)
- .varPool(taskExecutionRunnable.getTaskInstance().getVarPool())
+
.varPool(VarPoolUtils.deserializeVarPool(taskExecutionRunnable.getTaskInstance().getVarPool()))
.endTime(taskExecutionRunnable.getTaskInstance().getEndTime())
.build();
super.succeedEventAction(workflowExecutionRunnable,
taskExecutionRunnable, taskSuccessLifecycleEvent);
diff --git
a/dolphinscheduler-master/src/main/java/org/apache/dolphinscheduler/server/master/engine/workflow/statemachine/IWorkflowStateAction.java
b/dolphinscheduler-master/src/main/java/org/apache/dolphinscheduler/server/master/engine/workflow/statemachine/IWorkflowStateAction.java
index 5bac4ada66..3aafd79f92 100644
---
a/dolphinscheduler-master/src/main/java/org/apache/dolphinscheduler/server/master/engine/workflow/statemachine/IWorkflowStateAction.java
+++
b/dolphinscheduler-master/src/main/java/org/apache/dolphinscheduler/server/master/engine/workflow/statemachine/IWorkflowStateAction.java
@@ -65,7 +65,7 @@ public interface IWorkflowStateAction {
final WorkflowPauseLifecycleEvent
workflowPauseEvent);
/**
- * Perform the necessary actions when the workflow in a certain state
receive a {@link WorkflowStopLifecycleEvent}.
+ * Perform the necessary actions when the workflow in a certain state
receive a {@link WorkflowPausedLifecycleEvent}.
*/
void pausedEventAction(final IWorkflowExecutionRunnable
workflowExecutionRunnable,
final WorkflowPausedLifecycleEvent
workflowPausedEvent);
diff --git
a/dolphinscheduler-master/src/main/java/org/apache/dolphinscheduler/server/master/runner/TaskExecutionContextFactory.java
b/dolphinscheduler-master/src/main/java/org/apache/dolphinscheduler/server/master/runner/TaskExecutionContextFactory.java
index 7788b96b3f..cf88c86234 100644
---
a/dolphinscheduler-master/src/main/java/org/apache/dolphinscheduler/server/master/runner/TaskExecutionContextFactory.java
+++
b/dolphinscheduler-master/src/main/java/org/apache/dolphinscheduler/server/master/runner/TaskExecutionContextFactory.java
@@ -24,6 +24,7 @@ import org.apache.dolphinscheduler.common.utils.JSONUtils;
import org.apache.dolphinscheduler.dao.entity.DataSource;
import org.apache.dolphinscheduler.dao.entity.Environment;
import org.apache.dolphinscheduler.dao.entity.Project;
+import org.apache.dolphinscheduler.dao.entity.TaskDefinition;
import org.apache.dolphinscheduler.dao.entity.TaskInstance;
import org.apache.dolphinscheduler.dao.entity.WorkflowDefinition;
import org.apache.dolphinscheduler.dao.entity.WorkflowInstance;
@@ -39,17 +40,24 @@ import
org.apache.dolphinscheduler.plugin.task.api.parameters.resource.AbstractR
import
org.apache.dolphinscheduler.plugin.task.api.parameters.resource.DataSourceParameters;
import
org.apache.dolphinscheduler.plugin.task.api.parameters.resource.ResourceParametersHelper;
import org.apache.dolphinscheduler.plugin.task.api.utils.MapUtils;
+import org.apache.dolphinscheduler.plugin.task.api.utils.VarPoolUtils;
import org.apache.dolphinscheduler.server.master.config.MasterConfig;
+import
org.apache.dolphinscheduler.server.master.engine.graph.IWorkflowExecutionGraph;
+import
org.apache.dolphinscheduler.server.master.engine.task.runnable.ITaskExecutionRunnable;
import
org.apache.dolphinscheduler.server.master.engine.task.runnable.TaskExecutionContextBuilder;
import
org.apache.dolphinscheduler.server.master.engine.task.runnable.TaskExecutionContextCreateRequest;
import org.apache.dolphinscheduler.service.expand.CuringParamsService;
import org.apache.dolphinscheduler.service.process.ProcessService;
+import org.apache.commons.collections4.CollectionUtils;
import org.apache.commons.lang3.StringUtils;
+import java.util.Collections;
+import java.util.List;
import java.util.Map;
import java.util.Objects;
import java.util.Optional;
+import java.util.stream.Collectors;
import lombok.extern.slf4j.Slf4j;
@@ -78,6 +86,10 @@ public class TaskExecutionContextFactory {
final WorkflowDefinition workflowDefinition =
request.getWorkflowDefinition();
final Project project = request.getProject();
+ final List<Property> varPools =
+ generateTaskInstanceVarPool(request.getTaskDefinition(),
request.getWorkflowExecutionGraph());
+ taskInstance.setVarPool(VarPoolUtils.serializeVarPool(varPools));
+
return TaskExecutionContextBuilder.get()
.buildWorkflowInstanceHost(masterConfig.getMasterAddress())
.buildTaskInstanceRelatedInfo(taskInstance)
@@ -180,4 +192,21 @@ public class TaskExecutionContextFactory {
return Optional.ofNullable(environmentOptional.get().getConfig());
}
+ // The successors of the task instance will be used to generate the var
pool
+ // All out varPool from the successors will be merged into the var pool of
the task instance
+ private List<Property> generateTaskInstanceVarPool(TaskDefinition
taskDefinition,
+ IWorkflowExecutionGraph
workflowExecutionGraph) {
+ List<ITaskExecutionRunnable> predecessors =
workflowExecutionGraph.getPredecessors(taskDefinition.getName());
+ if (CollectionUtils.isEmpty(predecessors)) {
+ return Collections.emptyList();
+ }
+ List<String> varPoolsFromPredecessors = predecessors
+ .stream()
+ .filter(ITaskExecutionRunnable::isTaskInstanceInitialized)
+ .map(ITaskExecutionRunnable::getTaskInstance)
+ .map(TaskInstance::getVarPool)
+ .collect(Collectors.toList());
+ return VarPoolUtils.mergeVarPoolJsonString(varPoolsFromPredecessors);
+ }
+
}
diff --git
a/dolphinscheduler-master/src/test/java/org/apache/dolphinscheduler/server/master/integration/cases/WorkflowInstanceRecoverStopTestCase.java
b/dolphinscheduler-master/src/test/java/org/apache/dolphinscheduler/server/master/integration/cases/WorkflowInstanceRecoverStopTestCase.java
index ce6cd2594d..8c0c11b61d 100644
---
a/dolphinscheduler-master/src/test/java/org/apache/dolphinscheduler/server/master/integration/cases/WorkflowInstanceRecoverStopTestCase.java
+++
b/dolphinscheduler-master/src/test/java/org/apache/dolphinscheduler/server/master/integration/cases/WorkflowInstanceRecoverStopTestCase.java
@@ -27,6 +27,7 @@ import
org.apache.dolphinscheduler.dao.entity.WorkflowInstance;
import org.apache.dolphinscheduler.extract.base.client.Clients;
import org.apache.dolphinscheduler.extract.master.IWorkflowControlClient;
import
org.apache.dolphinscheduler.extract.master.command.RunWorkflowCommandParam;
+import
org.apache.dolphinscheduler.extract.master.transportor.workflow.WorkflowInstanceRecoverSuspendTasksResponse;
import
org.apache.dolphinscheduler.extract.master.transportor.workflow.WorkflowInstanceStopRequest;
import
org.apache.dolphinscheduler.extract.master.transportor.workflow.WorkflowInstanceStopResponse;
import org.apache.dolphinscheduler.plugin.task.api.enums.TaskExecutionStatus;
@@ -81,10 +82,11 @@ public class WorkflowInstanceRecoverStopTestCase extends
AbstractMasterIntegrati
});
});
-
assertThat(workflowOperator.recoverSuspendWorkflowInstance(workflowInstanceId).isSuccess());
+
assertThat(workflowOperator.recoverSuspendWorkflowInstance(workflowInstanceId))
+
.matches(WorkflowInstanceRecoverSuspendTasksResponse::isSuccess);
await()
- .pollInterval(Duration.ofMillis(100))
+ .pollInterval(Duration.ofSeconds(1))
.atMost(Duration.ofMinutes(1))
.untilAsserted(() -> {
@@ -158,6 +160,7 @@ public class WorkflowInstanceRecoverStopTestCase extends
AbstractMasterIntegrati
await()
.atMost(Duration.ofMinutes(1))
+ .pollInterval(Duration.ofSeconds(1))
.untilAsserted(() -> {
assertThat(repository.queryWorkflowInstance(workflow))
.hasSize(1)
diff --git
a/dolphinscheduler-master/src/test/java/org/apache/dolphinscheduler/server/master/integration/cases/WorkflowStartTestCase.java
b/dolphinscheduler-master/src/test/java/org/apache/dolphinscheduler/server/master/integration/cases/WorkflowStartTestCase.java
index 0d180e2f7f..4c56976dfd 100644
---
a/dolphinscheduler-master/src/test/java/org/apache/dolphinscheduler/server/master/integration/cases/WorkflowStartTestCase.java
+++
b/dolphinscheduler-master/src/test/java/org/apache/dolphinscheduler/server/master/integration/cases/WorkflowStartTestCase.java
@@ -30,6 +30,7 @@ import
org.apache.dolphinscheduler.plugin.task.api.enums.DataType;
import org.apache.dolphinscheduler.plugin.task.api.enums.Direct;
import org.apache.dolphinscheduler.plugin.task.api.enums.TaskExecutionStatus;
import org.apache.dolphinscheduler.plugin.task.api.model.Property;
+import org.apache.dolphinscheduler.plugin.task.api.utils.VarPoolUtils;
import
org.apache.dolphinscheduler.server.master.AbstractMasterIntegrationTestCase;
import org.apache.dolphinscheduler.server.master.integration.WorkflowOperator;
import
org.apache.dolphinscheduler.server.master.integration.WorkflowTestCaseContext;
@@ -584,6 +585,59 @@ public class WorkflowStartTestCase extends
AbstractMasterIntegrationTestCase {
masterContainer.assertAllResourceReleased();
}
+ @Test
+ @DisplayName("Test start a workflow contains fake task using local param
will be overwrite by varpool")
+ public void testStartWorkflow_fakeTask_usingLocalParamOverWriteByVarPool()
{
+ final String yaml =
"/it/start/workflow_with_local_param_overwrite_by_varpool.yaml";
+ final WorkflowTestCaseContext context =
workflowTestCaseContextFactory.initializeContextFromYaml(yaml);
+ final WorkflowDefinition workflow = context.getOneWorkflow();
+
+ final RunWorkflowCommandParam runWorkflowCommandParam =
RunWorkflowCommandParam.builder()
+ .build();
+
+ final WorkflowOperator.WorkflowTriggerDTO workflowTriggerDTO =
WorkflowOperator.WorkflowTriggerDTO.builder()
+ .workflowDefinition(workflow)
+ .runWorkflowCommandParam(runWorkflowCommandParam)
+ .build();
+ workflowOperator.manualTriggerWorkflow(workflowTriggerDTO);
+
+ List<Property> assertVarPools = Lists.newArrayList(
+
Property.builder().prop("output").direct(Direct.OUT).type(DataType.VARCHAR).value("1").build());
+ await()
+ .atMost(Duration.ofMinutes(1))
+ .untilAsserted(() -> {
+ Assertions
+
.assertThat(repository.queryWorkflowInstance(workflow))
+ .satisfiesExactly(workflowInstance -> {
+
assertThat(workflowInstance.getState()).isEqualTo(WorkflowExecutionStatus.SUCCESS);
+
assertThat(VarPoolUtils.deserializeVarPool(workflowInstance.getVarPool()))
+ .isEqualTo(assertVarPools);
+ });
+ Assertions
+ .assertThat(repository.queryTaskInstance(workflow))
+ .hasSize(3)
+ .anySatisfy(taskInstance -> {
+
assertThat(taskInstance.getName()).isEqualTo("A");
+
assertThat(VarPoolUtils.deserializeVarPool(taskInstance.getVarPool()))
+ .isEqualTo(assertVarPools);
+
assertThat(taskInstance.getState()).isEqualTo(TaskExecutionStatus.SUCCESS);
+ })
+ .anySatisfy(taskInstance -> {
+
assertThat(taskInstance.getName()).isEqualTo("B");
+
assertThat(VarPoolUtils.deserializeVarPool(taskInstance.getVarPool()))
+ .isEqualTo(assertVarPools);
+
assertThat(taskInstance.getState()).isEqualTo(TaskExecutionStatus.SUCCESS);
+ })
+ .anySatisfy(taskInstance -> {
+
assertThat(taskInstance.getName()).isEqualTo("C");
+
assertThat(VarPoolUtils.deserializeVarPool(taskInstance.getVarPool()))
+ .isEqualTo(assertVarPools);
+
assertThat(taskInstance.getState()).isEqualTo(TaskExecutionStatus.SUCCESS);
+ });
+ });
+ masterContainer.assertAllResourceReleased();
+ }
+
@Test
@DisplayName("Test start a workflow with one fake task(A) failed")
public void testStartWorkflow_with_oneFailedTask() {
diff --git a/dolphinscheduler-master/src/test/resources/application.yaml
b/dolphinscheduler-master/src/test/resources/application.yaml
index f9cb331abf..3aca1e3142 100644
--- a/dolphinscheduler-master/src/test/resources/application.yaml
+++ b/dolphinscheduler-master/src/test/resources/application.yaml
@@ -72,7 +72,7 @@ master:
memory-usage-weight: 40
cpu-usage-weight: 30
task-thread-pool-usage-weight: 30
- worker-group-refresh-interval: 10s
+ worker-group-refresh-interval: 5m
command-fetch-strategy:
type: ID_SLOT_BASED
config:
diff --git
a/dolphinscheduler-master/src/test/resources/it/start/workflow_with_local_param_overwrite_by_varpool.yaml
b/dolphinscheduler-master/src/test/resources/it/start/workflow_with_local_param_overwrite_by_varpool.yaml
new file mode 100644
index 0000000000..82f5bee85f
--- /dev/null
+++
b/dolphinscheduler-master/src/test/resources/it/start/workflow_with_local_param_overwrite_by_varpool.yaml
@@ -0,0 +1,137 @@
+#
+# 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.
+#
+
+project:
+ name: MasterIntegrationTest
+ code: 1
+ description: This is a fake project
+ userId: 1
+ userName: admin
+ createTime: 2024-08-12 00:00:00
+ updateTime: 2021-08-12 00:00:00
+
+workflows:
+ - name: workflow_with_fake_task_success
+ code: 1
+ version: 1
+ projectCode: 1
+ description: This is a fake workflow with three task using varpool
+ releaseState: ONLINE
+ createTime: 2024-08-12 00:00:00
+ updateTime: 2021-08-12 00:00:00
+ userId: 1
+ executionType: PARALLEL
+
+tasks:
+ - name: A
+ code: 1
+ version: 1
+ projectCode: 1
+ userId: 1
+ taskType: LogicFakeTask
+ taskParams: >
+ {
+ "localParams": [
+ {
+ "prop": "output",
+ "direct": "OUT",
+ "type": "VARCHAR",
+ "value": ""
+ }
+ ],
+ "shellScript": "echo '${setValue(output=1)}'",
+ "resourceList": []
+ }
+ workerGroup: default
+ createTime: 2024-08-12 00:00:00
+ updateTime: 2021-08-12 00:00:00
+ taskExecuteType: BATCH
+ - name: B
+ code: 2
+ version: 1
+ projectCode: 1
+ userId: 1
+ taskType: LogicFakeTask
+ taskParams: >
+ {
+ "localParams": [
+ {
+ "prop": "output",
+ "direct": "IN",
+ "type": "VARCHAR",
+ "value": ""
+ }
+ ],
+ "shellScript": "echo '${output}'",
+ "resourceList": []
+ }
+ workerGroup: default
+ createTime: 2024-08-12 00:00:00
+ updateTime: 2021-08-12 00:00:00
+ taskExecuteType: BATCH
+ - name: C
+ code: 3
+ version: 1
+ projectCode: 1
+ userId: 1
+ taskType: LogicFakeTask
+ taskParams: >
+ {
+ "localParams": [
+ {
+ "prop": "output",
+ "direct": "IN",
+ "type": "VARCHAR",
+ "value": ""
+ }
+ ],
+ "shellScript": "echo '${output}'",
+ "resourceList": []
+ }
+ workerGroup: default
+ createTime: 2024-08-12 00:00:00
+ updateTime: 2021-08-12 00:00:00
+ taskExecuteType: BATCH
+
+taskRelations:
+ - projectCode: 1
+ workflowDefinitionCode: 1
+ workflowDefinitionVersion: 1
+ preTaskCode: 0
+ preTaskVersion: 0
+ postTaskCode: 1
+ postTaskVersion: 1
+ createTime: 2024-08-12 00:00:00
+ updateTime: 2024-08-12 00:00:00
+ - projectCode: 1
+ workflowDefinitionCode: 1
+ workflowDefinitionVersion: 1
+ preTaskCode: 1
+ preTaskVersion: 1
+ postTaskCode: 2
+ postTaskVersion: 1
+ createTime: 2024-08-12 00:00:00
+ updateTime: 2024-08-12 00:00:00
+ - projectCode: 1
+ workflowDefinitionCode: 1
+ workflowDefinitionVersion: 1
+ preTaskCode: 2
+ preTaskVersion: 1
+ postTaskCode: 3
+ postTaskVersion: 1
+ createTime: 2024-08-12 00:00:00
+ updateTime: 2024-08-12 00:00:00
diff --git
a/dolphinscheduler-service/src/main/java/org/apache/dolphinscheduler/service/expand/CuringParamsService.java
b/dolphinscheduler-service/src/main/java/org/apache/dolphinscheduler/service/expand/CuringParamsService.java
index 40fb9b6b26..0bdf5468b8 100644
---
a/dolphinscheduler-service/src/main/java/org/apache/dolphinscheduler/service/expand/CuringParamsService.java
+++
b/dolphinscheduler-service/src/main/java/org/apache/dolphinscheduler/service/expand/CuringParamsService.java
@@ -33,22 +33,6 @@ import lombok.NonNull;
public interface CuringParamsService {
- /**
- * time function need expand
- * @param placeholderName
- * @return
- */
- boolean timeFunctionNeedExpand(String placeholderName);
-
- /**
- * time function extension
- * @param workflowInstanceId
- * @param timezone
- * @param placeholderName
- * @return
- */
- String timeFunctionExtension(Integer workflowInstanceId, String timezone,
String placeholderName);
-
/**
* convert parameter placeholders
* @param val
diff --git
a/dolphinscheduler-service/src/main/java/org/apache/dolphinscheduler/service/expand/CuringParamsServiceImpl.java
b/dolphinscheduler-service/src/main/java/org/apache/dolphinscheduler/service/expand/CuringParamsServiceImpl.java
index 59e04666dd..d98b2cfc4c 100644
---
a/dolphinscheduler-service/src/main/java/org/apache/dolphinscheduler/service/expand/CuringParamsServiceImpl.java
+++
b/dolphinscheduler-service/src/main/java/org/apache/dolphinscheduler/service/expand/CuringParamsServiceImpl.java
@@ -47,13 +47,16 @@ import
org.apache.dolphinscheduler.plugin.task.api.parameters.AbstractParameters
import org.apache.dolphinscheduler.plugin.task.api.utils.MapUtils;
import org.apache.dolphinscheduler.plugin.task.api.utils.ParameterUtils;
import org.apache.dolphinscheduler.plugin.task.api.utils.PropertyUtils;
+import org.apache.dolphinscheduler.plugin.task.api.utils.VarPoolUtils;
import org.apache.commons.collections4.CollectionUtils;
import org.apache.commons.lang3.StringUtils;
+import java.util.Collections;
import java.util.Date;
import java.util.HashMap;
import java.util.Iterator;
+import java.util.LinkedHashMap;
import java.util.List;
import java.util.Map;
import java.util.Set;
@@ -70,9 +73,6 @@ import org.springframework.stereotype.Component;
@Component
public class CuringParamsServiceImpl implements CuringParamsService {
- @Autowired
- private TimePlaceholderResolverExpandService
timePlaceholderResolverExpandService;
-
@Autowired
private ProjectParameterMapper projectParameterMapper;
@@ -86,17 +86,6 @@ public class CuringParamsServiceImpl implements
CuringParamsService {
return ParameterUtils.convertParameterPlaceholders(val, paramMap);
}
- @Override
- public boolean timeFunctionNeedExpand(String placeholderName) {
- return
timePlaceholderResolverExpandService.timeFunctionNeedExpand(placeholderName);
- }
-
- @Override
- public String timeFunctionExtension(Integer workflowInstanceId, String
timezone, String placeholderName) {
- return
timePlaceholderResolverExpandService.timeFunctionExtension(workflowInstanceId,
timezone,
- placeholderName);
- }
-
/**
* here it is judged whether external expansion calculation is required
and the calculation result is obtained
*
@@ -133,10 +122,6 @@ public class CuringParamsServiceImpl implements
CuringParamsService {
String val = entry.getValue();
if (val.contains(Constants.FUNCTION_START_WITH)) {
String str = val;
- // whether external scaling calculation is required
- if (timeFunctionNeedExpand(val)) {
- str = timeFunctionExtension(workflowInstanceId, timezone,
val);
- }
resolveMap.put(entry.getKey(), str);
}
}
@@ -175,7 +160,10 @@ public class CuringParamsServiceImpl implements
CuringParamsService {
}
/**
- * the global parameters and local parameters used in the worker will be
prepared here, and built-in parameters.
+ * Generate prepare params include project params, global parameters,
local parameters, built-in parameters, varpool, start-up params.
+ * <p> The priority of the parameters is as follows:
+ * <p> varpool > command parameters > local parameters > global parameters
> project parameters > built-in parameters
+ * todo: Use TaskRuntimeParams to represent this.
*
* @param taskInstance
* @param parameters
@@ -193,14 +181,13 @@ public class CuringParamsServiceImpl implements
CuringParamsService {
Map<String, Property> prepareParamsMap = new HashMap<>();
// assign value to definedParams here
- Map<String, Property> globalParams =
setGlobalParamsMap(workflowInstance);
+ Map<String, Property> globalParams =
parseGlobalParamsMap(workflowInstance);
// combining local and global parameters
Map<String, Property> localParams =
parameters.getInputLocalParametersMap();
// stream pass params
- parameters.setVarPool(taskInstance.getVarPool());
- Map<String, Property> varParams = parameters.getVarPoolMap();
+ List<Property> varPools = parseVarPool(taskInstance);
// if it is a complement,
// you need to pass in the task instance id to locate the time
@@ -227,10 +214,6 @@ public class CuringParamsServiceImpl implements
CuringParamsService {
prepareParamsMap.putAll(globalParams);
}
- if (MapUtils.isNotEmpty(varParams)) {
- prepareParamsMap.putAll(varParams);
- }
-
if (MapUtils.isNotEmpty(localParams)) {
prepareParamsMap.putAll(localParams);
}
@@ -240,6 +223,17 @@ public class CuringParamsServiceImpl implements
CuringParamsService {
.collect(Collectors.toMap(Property::getProp,
Function.identity())));
}
+ if (CollectionUtils.isNotEmpty(varPools)) {
+ // overwrite the in parameter by varPool
+ for (Property varPool : varPools) {
+ Property property = prepareParamsMap.get(varPool.getProp());
+ if (property == null || property.getDirect() != Direct.IN) {
+ continue;
+ }
+ property.setValue(varPool.getValue());
+ }
+ }
+
Iterator<Map.Entry<String, Property>> iter =
prepareParamsMap.entrySet().iterator();
while (iter.hasNext()) {
Map.Entry<String, Property> en = iter.next();
@@ -253,14 +247,10 @@ public class CuringParamsServiceImpl implements
CuringParamsService {
* and there are no variables in them.
*/
String val = property.getValue();
- // whether external scaling calculation is required
- if (timeFunctionNeedExpand(val)) {
- val =
timeFunctionExtension(taskInstance.getWorkflowInstanceId(), timeZone, val);
- } else {
- // handle some chain parameter assign, such as `{"var1":
"${var2}", "var2": 1}` should be convert to
- // `{"var1": 1, "var2": 1}`
- val = convertParameterPlaceholders(val, prepareParamsMap);
- }
+
+ // handle some chain parameter assign, such as `{"var1":
"${var2}", "var2": 1}` should be convert to
+ // `{"var1": 1, "var2": 1}`
+ val = convertParameterPlaceholders(val, prepareParamsMap);
property.setValue(val);
}
}
@@ -305,18 +295,20 @@ public class CuringParamsServiceImpl implements
CuringParamsService {
return params;
}
- private Map<String, Property> setGlobalParamsMap(WorkflowInstance
workflowInstance) {
- Map<String, Property> globalParamsMap = new HashMap<>(16);
+ private Map<String, Property> parseGlobalParamsMap(WorkflowInstance
workflowInstance) {
+ final Map<String, Property> globalParametersMaps = new
LinkedHashMap<>();
+ if (StringUtils.isNotEmpty(workflowInstance.getGlobalParams())) {
+ JSONUtils.toList(workflowInstance.getGlobalParams(),
Property.class)
+ .forEach(property ->
globalParametersMaps.put(property.getProp(), property));
+ }
+ return globalParametersMaps;
+ }
- // global params string
- String globalParamsStr = workflowInstance.getGlobalParams();
- if (globalParamsStr != null) {
- List<Property> globalParamsList =
JSONUtils.toList(globalParamsStr, Property.class);
- globalParamsMap
- .putAll(globalParamsList.stream()
- .collect(Collectors.toMap(Property::getProp,
Function.identity())));
+ private List<Property> parseVarPool(TaskInstance taskInstance) {
+ if (StringUtils.isNotEmpty(taskInstance.getVarPool())) {
+ return VarPoolUtils.deserializeVarPool(taskInstance.getVarPool());
}
- return globalParamsMap;
+ return Collections.emptyList();
}
@Override
diff --git
a/dolphinscheduler-service/src/main/java/org/apache/dolphinscheduler/service/expand/TimePlaceholderResolverExpandService.java
b/dolphinscheduler-service/src/main/java/org/apache/dolphinscheduler/service/expand/TimePlaceholderResolverExpandService.java
deleted file mode 100644
index a1d5d737e3..0000000000
---
a/dolphinscheduler-service/src/main/java/org/apache/dolphinscheduler/service/expand/TimePlaceholderResolverExpandService.java
+++ /dev/null
@@ -1,35 +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.service.expand;
-
-public interface TimePlaceholderResolverExpandService {
-
- /**
- * check is need expand function
- * @param placeholderName
- * @return
- */
- boolean timeFunctionNeedExpand(String placeholderName);
-
- /**
- * time function extension
- * @param placeholderName
- * @return
- */
- String timeFunctionExtension(Integer workflowInstanceId, String timeZone,
String placeholderName);
-}
diff --git
a/dolphinscheduler-service/src/main/java/org/apache/dolphinscheduler/service/expand/TimePlaceholderResolverExpandServiceImpl.java
b/dolphinscheduler-service/src/main/java/org/apache/dolphinscheduler/service/expand/TimePlaceholderResolverExpandServiceImpl.java
deleted file mode 100644
index a43c20d44a..0000000000
---
a/dolphinscheduler-service/src/main/java/org/apache/dolphinscheduler/service/expand/TimePlaceholderResolverExpandServiceImpl.java
+++ /dev/null
@@ -1,34 +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.service.expand;
-
-import org.springframework.stereotype.Component;
-
-@Component
-public class TimePlaceholderResolverExpandServiceImpl implements
TimePlaceholderResolverExpandService {
-
- @Override
- public boolean timeFunctionNeedExpand(String placeholderName) {
- return false;
- }
-
- @Override
- public String timeFunctionExtension(Integer workflowInstanceId, String
timeZone, String placeholderName) {
- return null;
- }
-}
diff --git
a/dolphinscheduler-service/src/test/java/org/apache/dolphinscheduler/service/expand/CuringParamsServiceTest.java
b/dolphinscheduler-service/src/test/java/org/apache/dolphinscheduler/service/expand/CuringParamsServiceTest.java
index 212c65add0..f0f2874a9d 100644
---
a/dolphinscheduler-service/src/test/java/org/apache/dolphinscheduler/service/expand/CuringParamsServiceTest.java
+++
b/dolphinscheduler-service/src/test/java/org/apache/dolphinscheduler/service/expand/CuringParamsServiceTest.java
@@ -65,15 +65,9 @@ public class CuringParamsServiceTest {
@InjectMocks
private CuringParamsServiceImpl dolphinSchedulerCuringGlobalParams;
- @Mock
- private TimePlaceholderResolverExpandService
timePlaceholderResolverExpandService;
-
@Mock
private ProjectParameterMapper projectParameterMapper;
- @InjectMocks
- private TimePlaceholderResolverExpandServiceImpl
timePlaceholderResolverExpandServiceImpl;
-
private final Map<String, String> globalParamMap = new HashMap<>();
private final Map<String, Property> paramMap = new HashMap<>();
@@ -91,18 +85,6 @@ public class CuringParamsServiceTest {
Assertions.assertNotNull(result);
}
- @Test
- public void testTimeFunctionNeedExpand() {
- boolean result =
curingGlobalParamsService.timeFunctionNeedExpand(placeHolderName);
- Assertions.assertFalse(result);
- }
-
- @Test
- public void testTimeFunctionExtension() {
- String result = curingGlobalParamsService.timeFunctionExtension(1, "",
placeHolderName);
- Assertions.assertNull(result);
- }
-
@Test
public void testCuringGlobalParams() {
// define globalMap
diff --git
a/dolphinscheduler-service/src/test/java/org/apache/dolphinscheduler/service/expand/TimePlaceholderResolverExpandServiceTest.java
b/dolphinscheduler-service/src/test/java/org/apache/dolphinscheduler/service/expand/TimePlaceholderResolverExpandServiceTest.java
deleted file mode 100644
index 07e44205ee..0000000000
---
a/dolphinscheduler-service/src/test/java/org/apache/dolphinscheduler/service/expand/TimePlaceholderResolverExpandServiceTest.java
+++ /dev/null
@@ -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.service.expand;
-
-import org.apache.commons.lang3.StringUtils;
-
-import org.junit.jupiter.api.Assertions;
-import org.junit.jupiter.api.Test;
-import org.junit.jupiter.api.extension.ExtendWith;
-import org.mockito.InjectMocks;
-import org.mockito.Mock;
-import org.mockito.junit.jupiter.MockitoExtension;
-
-@ExtendWith(MockitoExtension.class)
-public class TimePlaceholderResolverExpandServiceTest {
-
- @Mock
- private TimePlaceholderResolverExpandService
timePlaceholderResolverExpandService;
-
- @InjectMocks
- private TimePlaceholderResolverExpandServiceImpl
timePlaceholderResolverExpandServiceImpl;
-
- private static final String placeHolderName = "$[yyyy-MM-dd-1]";
-
- @Test
- public void testTimePlaceholderResolverExpandService() {
- boolean checkResult =
timePlaceholderResolverExpandService.timeFunctionNeedExpand(placeHolderName);
- Assertions.assertFalse(checkResult);
- String resultString =
timePlaceholderResolverExpandService.timeFunctionExtension(1, "",
placeHolderName);
- Assertions.assertTrue(StringUtils.isEmpty(resultString));
-
- boolean implCheckResult =
timePlaceholderResolverExpandServiceImpl.timeFunctionNeedExpand(placeHolderName);
- Assertions.assertFalse(implCheckResult);
- String implResultString =
-
timePlaceholderResolverExpandServiceImpl.timeFunctionExtension(1, "",
placeHolderName);
- Assertions.assertTrue(StringUtils.isEmpty(implResultString));
- }
-}
diff --git
a/dolphinscheduler-task-executor/src/main/java/org/apache/dolphinscheduler/task/executor/events/TaskExecutorSuccessLifecycleEvent.java
b/dolphinscheduler-task-executor/src/main/java/org/apache/dolphinscheduler/task/executor/events/TaskExecutorSuccessLifecycleEvent.java
index 123f394132..92a9e66e77 100644
---
a/dolphinscheduler-task-executor/src/main/java/org/apache/dolphinscheduler/task/executor/events/TaskExecutorSuccessLifecycleEvent.java
+++
b/dolphinscheduler-task-executor/src/main/java/org/apache/dolphinscheduler/task/executor/events/TaskExecutorSuccessLifecycleEvent.java
@@ -18,8 +18,11 @@
package org.apache.dolphinscheduler.task.executor.events;
import org.apache.dolphinscheduler.plugin.task.api.TaskExecutionContext;
+import org.apache.dolphinscheduler.plugin.task.api.model.Property;
import org.apache.dolphinscheduler.task.executor.ITaskExecutor;
+import java.util.List;
+
import lombok.Data;
import lombok.EqualsAndHashCode;
import lombok.NoArgsConstructor;
@@ -43,7 +46,7 @@ public class TaskExecutorSuccessLifecycleEvent extends
AbstractTaskExecutorLifec
private long endTime;
- private String varPool;
+ private List<Property> varPool;
private Long latestReportTime;
diff --git
a/dolphinscheduler-task-plugin/dolphinscheduler-task-api/src/main/java/org/apache/dolphinscheduler/plugin/task/api/TaskExecutionContext.java
b/dolphinscheduler-task-plugin/dolphinscheduler-task-api/src/main/java/org/apache/dolphinscheduler/plugin/task/api/TaskExecutionContext.java
index c13a6e4d6f..12f9449924 100644
---
a/dolphinscheduler-task-plugin/dolphinscheduler-task-api/src/main/java/org/apache/dolphinscheduler/plugin/task/api/TaskExecutionContext.java
+++
b/dolphinscheduler-task-plugin/dolphinscheduler-task-api/src/main/java/org/apache/dolphinscheduler/plugin/task/api/TaskExecutionContext.java
@@ -23,6 +23,7 @@ import
org.apache.dolphinscheduler.plugin.task.api.parameters.resource.ResourceP
import org.apache.dolphinscheduler.plugin.task.api.resource.ResourceContext;
import java.io.Serializable;
+import java.util.List;
import java.util.Map;
import lombok.AllArgsConstructor;
@@ -89,7 +90,7 @@ public class TaskExecutionContext implements Serializable {
private String environmentConfig;
/**
- * Include local params, global params and system built-in params
+ * Include local params, global params, varpool transport from successors,
start-up params and system built-in params
*/
private Map<String, Property> prepareParamsMap;
@@ -113,7 +114,7 @@ public class TaskExecutionContext implements Serializable {
private ResourceContext resourceContext;
- private String varPool;
+ private List<Property> varPool;
private int dryRun;
diff --git
a/dolphinscheduler-task-plugin/dolphinscheduler-task-api/src/main/java/org/apache/dolphinscheduler/plugin/task/api/model/Property.java
b/dolphinscheduler-task-plugin/dolphinscheduler-task-api/src/main/java/org/apache/dolphinscheduler/plugin/task/api/model/Property.java
index 3ef896596b..c7ec6c8599 100644
---
a/dolphinscheduler-task-plugin/dolphinscheduler-task-api/src/main/java/org/apache/dolphinscheduler/plugin/task/api/model/Property.java
+++
b/dolphinscheduler-task-plugin/dolphinscheduler-task-api/src/main/java/org/apache/dolphinscheduler/plugin/task/api/model/Property.java
@@ -21,7 +21,6 @@ import
org.apache.dolphinscheduler.plugin.task.api.enums.DataType;
import org.apache.dolphinscheduler.plugin.task.api.enums.Direct;
import java.io.Serializable;
-import java.util.Objects;
import lombok.AllArgsConstructor;
import lombok.Builder;
@@ -55,32 +54,4 @@ public class Property implements Serializable {
*/
private String value;
- @Override
- public boolean equals(Object o) {
- if (this == o) {
- return true;
- }
- if (o == null || getClass() != o.getClass()) {
- return false;
- }
- Property property = (Property) o;
- return Objects.equals(prop, property.prop)
- && Objects.equals(value, property.value);
- }
-
- @Override
- public int hashCode() {
- return Objects.hash(prop, value);
- }
-
- @Override
- public String toString() {
- return "Property{"
- + "prop='" + prop + '\''
- + ", direct=" + direct
- + ", type=" + type
- + ", value='" + value + '\''
- + '}';
- }
-
}
diff --git
a/dolphinscheduler-task-plugin/dolphinscheduler-task-api/src/main/java/org/apache/dolphinscheduler/plugin/task/api/parameters/AbstractParameters.java
b/dolphinscheduler-task-plugin/dolphinscheduler-task-api/src/main/java/org/apache/dolphinscheduler/plugin/task/api/parameters/AbstractParameters.java
index f99578d743..497d851bd8 100644
---
a/dolphinscheduler-task-plugin/dolphinscheduler-task-api/src/main/java/org/apache/dolphinscheduler/plugin/task/api/parameters/AbstractParameters.java
+++
b/dolphinscheduler-task-plugin/dolphinscheduler-task-api/src/main/java/org/apache/dolphinscheduler/plugin/task/api/parameters/AbstractParameters.java
@@ -29,7 +29,6 @@ import
org.apache.dolphinscheduler.plugin.task.api.utils.VarPoolUtils;
import org.apache.commons.collections4.CollectionUtils;
import org.apache.commons.collections4.MapUtils;
-import org.apache.commons.lang3.StringUtils;
import java.util.ArrayList;
import java.util.LinkedHashMap;
@@ -101,27 +100,8 @@ public abstract class AbstractParameters implements
IParameters {
return localParametersMaps;
}
- /**
- * get varPool map
- *
- * @return parameters map
- */
- public Map<String, Property> getVarPoolMap() {
- Map<String, Property> varPoolMap = new LinkedHashMap<>();
- if (varPool != null) {
- for (Property property : varPool) {
- varPoolMap.put(property.getProp(), property);
- }
- }
- return varPoolMap;
- }
-
- public void setVarPool(String varPool) {
- if (StringUtils.isEmpty(varPool)) {
- this.varPool = new ArrayList<>();
- } else {
- this.varPool = JSONUtils.toList(varPool, Property.class);
- }
+ public void setVarPool(List<Property> varPool) {
+ this.varPool = varPool;
}
public void dealOutParam(Map<String, String> taskOutputParams) {
diff --git
a/dolphinscheduler-task-plugin/dolphinscheduler-task-api/src/main/java/org/apache/dolphinscheduler/plugin/task/api/utils/VarPoolUtils.java
b/dolphinscheduler-task-plugin/dolphinscheduler-task-api/src/main/java/org/apache/dolphinscheduler/plugin/task/api/utils/VarPoolUtils.java
index 7c24eb9a21..d830a5e42f 100644
---
a/dolphinscheduler-task-plugin/dolphinscheduler-task-api/src/main/java/org/apache/dolphinscheduler/plugin/task/api/utils/VarPoolUtils.java
+++
b/dolphinscheduler-task-plugin/dolphinscheduler-task-api/src/main/java/org/apache/dolphinscheduler/plugin/task/api/utils/VarPoolUtils.java
@@ -24,6 +24,7 @@ import
org.apache.dolphinscheduler.plugin.task.api.model.Property;
import org.apache.commons.collections4.CollectionUtils;
import java.util.ArrayList;
+import java.util.Collections;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
@@ -32,26 +33,41 @@ import java.util.stream.Collectors;
import lombok.experimental.UtilityClass;
import lombok.extern.slf4j.Slf4j;
+import com.google.common.collect.Lists;
+
@Slf4j
@UtilityClass
public class VarPoolUtils {
+ public String serializeVarPool(List<Property> varPool) {
+ if (CollectionUtils.isEmpty(varPool)) {
+ return null;
+ }
+ return JSONUtils.toJsonString(varPool);
+ }
+
public List<Property> deserializeVarPool(String varPoolJson) {
return JSONUtils.toList(varPoolJson, Property.class);
}
+ public List<Property> mergeVarPoolJsonString(String... varPoolJsons) {
+ if (varPoolJsons == null) {
+ return Collections.emptyList();
+ }
+ return mergeVarPoolJsonString(Lists.newArrayList(varPoolJsons));
+ }
+
/**
* @see #mergeVarPool(List)
*/
- public String mergeVarPoolJsonString(List<String> varPoolJsons) {
+ public List<Property> mergeVarPoolJsonString(List<String> varPoolJsons) {
if (CollectionUtils.isEmpty(varPoolJsons)) {
return null;
}
List<List<Property>> varPools = varPoolJsons.stream()
.map(VarPoolUtils::deserializeVarPool)
.collect(Collectors.toList());
- List<Property> finalVarPool = mergeVarPool(varPools);
- return JSONUtils.toJsonString(finalVarPool);
+ return mergeVarPool(varPools);
}
/**
@@ -61,7 +77,7 @@ public class VarPoolUtils {
*/
public List<Property> mergeVarPool(List<List<Property>> varPools) {
if (CollectionUtils.isEmpty(varPools)) {
- return null;
+ return Collections.emptyList();
}
if (varPools.size() == 1) {
return varPools.get(0);
@@ -73,7 +89,7 @@ public class VarPoolUtils {
}
for (Property property : varPool) {
if (!Direct.OUT.equals(property.getDirect())) {
- log.info("The direct should be OUT in varPool, but got
{}", property.getDirect());
+ log.warn("The direct should be OUT in varPool, but got
{}", property.getDirect());
continue;
}
result.put(property.getProp(), property);
diff --git
a/dolphinscheduler-task-plugin/dolphinscheduler-task-api/src/test/java/org/apache/dolphinscheduler/plugin/task/api/utils/VarPoolUtilsTest.java
b/dolphinscheduler-task-plugin/dolphinscheduler-task-api/src/test/java/org/apache/dolphinscheduler/plugin/task/api/utils/VarPoolUtilsTest.java
index 231d97029f..229ba5ce5f 100644
---
a/dolphinscheduler-task-plugin/dolphinscheduler-task-api/src/test/java/org/apache/dolphinscheduler/plugin/task/api/utils/VarPoolUtilsTest.java
+++
b/dolphinscheduler-task-plugin/dolphinscheduler-task-api/src/test/java/org/apache/dolphinscheduler/plugin/task/api/utils/VarPoolUtilsTest.java
@@ -32,7 +32,7 @@ class VarPoolUtilsTest {
@Test
void mergeVarPool() {
- Truth.assertThat(VarPoolUtils.mergeVarPool(null)).isNull();
+ Truth.assertThat(VarPoolUtils.mergeVarPool(null)).isEmpty();
// Override the value of the same property
// Merge the property with different key.
@@ -51,7 +51,8 @@ class VarPoolUtilsTest {
@Test
void subtractVarPool() {
Truth.assertThat(VarPoolUtils.subtractVarPool(null, null)).isNull();
- List<Property> varpool1 = Lists.newArrayList(new Property("name",
Direct.OUT, DataType.VARCHAR, "tom"),
+ List<Property> varpool1 = Lists.newArrayList(
+ new Property("name", Direct.OUT, DataType.VARCHAR, "tom"),
new Property("age", Direct.OUT, DataType.INTEGER, "10"));
List<Property> varpool2 = Lists.newArrayList(new Property("name",
Direct.OUT, DataType.VARCHAR, "tom"));
List<Property> varpool3 = Lists.newArrayList(new Property("location",
Direct.OUT, DataType.VARCHAR, "china"));
diff --git
a/dolphinscheduler-task-plugin/dolphinscheduler-task-python/src/main/java/org/apache/dolphinscheduler/plugin/task/python/PythonTask.java
b/dolphinscheduler-task-plugin/dolphinscheduler-task-python/src/main/java/org/apache/dolphinscheduler/plugin/task/python/PythonTask.java
index 1b5636a306..715438f6e6 100644
---
a/dolphinscheduler-task-plugin/dolphinscheduler-task-python/src/main/java/org/apache/dolphinscheduler/plugin/task/python/PythonTask.java
+++
b/dolphinscheduler-task-plugin/dolphinscheduler-task-python/src/main/java/org/apache/dolphinscheduler/plugin/task/python/PythonTask.java
@@ -98,7 +98,7 @@ public class PythonTask extends AbstractTask {
setProcessId(taskResponse.getProcessId());
setTaskOutputParams(shellCommandExecutor.getTaskOutputParams());
pythonParameters.dealOutParam(shellCommandExecutor.getTaskOutputParams());
-
taskRequest.setVarPool(JSONUtils.toJsonString(pythonParameters.getVarPool()));
+ taskRequest.setVarPool(pythonParameters.getVarPool());
} catch (Exception e) {
log.error("python task failure", e);
setExitStatusCode(TaskConstants.EXIT_CODE_FAILURE);
diff --git
a/dolphinscheduler-task-plugin/dolphinscheduler-task-shell/src/main/java/org/apache/dolphinscheduler/plugin/task/shell/ShellTask.java
b/dolphinscheduler-task-plugin/dolphinscheduler-task-shell/src/main/java/org/apache/dolphinscheduler/plugin/task/shell/ShellTask.java
index 71cfb9522c..0d90ab998b 100644
---
a/dolphinscheduler-task-plugin/dolphinscheduler-task-shell/src/main/java/org/apache/dolphinscheduler/plugin/task/shell/ShellTask.java
+++
b/dolphinscheduler-task-plugin/dolphinscheduler-task-shell/src/main/java/org/apache/dolphinscheduler/plugin/task/shell/ShellTask.java
@@ -86,6 +86,7 @@ public class ShellTask extends AbstractTask {
setExitStatusCode(commandExecuteResult.getExitStatusCode());
setProcessId(commandExecuteResult.getProcessId());
shellParameters.dealOutParam(shellCommandExecutor.getTaskOutputParams());
+ taskExecutionContext.setVarPool(shellParameters.getVarPool());
} catch (InterruptedException e) {
Thread.currentThread().interrupt();
log.error("The current Shell task has been interrupted", e);
diff --git
a/dolphinscheduler-worker/src/main/java/org/apache/dolphinscheduler/server/worker/executor/PhysicalTaskExecutor.java
b/dolphinscheduler-worker/src/main/java/org/apache/dolphinscheduler/server/worker/executor/PhysicalTaskExecutor.java
index 3886dad3fe..6a018a1ad8 100644
---
a/dolphinscheduler-worker/src/main/java/org/apache/dolphinscheduler/server/worker/executor/PhysicalTaskExecutor.java
+++
b/dolphinscheduler-worker/src/main/java/org/apache/dolphinscheduler/server/worker/executor/PhysicalTaskExecutor.java
@@ -21,12 +21,10 @@ import org.apache.dolphinscheduler.common.utils.JSONUtils;
import org.apache.dolphinscheduler.plugin.storage.api.StorageOperator;
import org.apache.dolphinscheduler.plugin.task.api.AbstractTask;
import org.apache.dolphinscheduler.plugin.task.api.TaskCallBack;
-import org.apache.dolphinscheduler.plugin.task.api.enums.Direct;
import org.apache.dolphinscheduler.plugin.task.api.model.ApplicationInfo;
import org.apache.dolphinscheduler.plugin.task.api.resource.ResourceContext;
import org.apache.dolphinscheduler.server.worker.config.WorkerConfig;
import
org.apache.dolphinscheduler.server.worker.utils.TaskExecutionContextUtils;
-import org.apache.dolphinscheduler.server.worker.utils.TaskFilesTransferUtils;
import org.apache.dolphinscheduler.server.worker.utils.TenantUtils;
import org.apache.dolphinscheduler.task.executor.AbstractTaskExecutor;
import org.apache.dolphinscheduler.task.executor.ITaskExecutor;
@@ -34,6 +32,8 @@ import
org.apache.dolphinscheduler.task.executor.TaskExecutorState;
import org.apache.dolphinscheduler.task.executor.TaskExecutorStateMappings;
import
org.apache.dolphinscheduler.task.executor.events.TaskExecutorRuntimeContextChangedLifecycleEvent;
+import java.util.ArrayList;
+
import lombok.Getter;
import lombok.extern.slf4j.Slf4j;
@@ -64,7 +64,7 @@ public class PhysicalTaskExecutor extends
AbstractTaskExecutor {
this.physicalTask.init();
-
this.physicalTask.getParameters().setVarPool(taskExecutionContext.getVarPool());
+ this.physicalTask.getParameters().setVarPool(new ArrayList<>());
log.info("Set taskVarPool: {} successfully",
taskExecutionContext.getVarPool());
}
@@ -122,11 +122,6 @@ public class PhysicalTaskExecutor extends
AbstractTaskExecutor {
taskExecutionContext.setResourceContext(resourceContext);
log.info("Download resources successfully: \n{}",
taskExecutionContext.getResourceContext());
- // todo: remove this. The cache should be deprecated
- TaskFilesTransferUtils.downloadUpstreamFiles(taskExecutionContext,
storageOperator);
- log.info("Download upstream files: {} successfully",
-
TaskFilesTransferUtils.getFileLocalParams(taskExecutionContext, Direct.IN));
-
log.info("End initialize task {}",
JSONUtils.toPrettyJsonString(taskExecutionContext));
}
diff --git
a/dolphinscheduler-worker/src/main/java/org/apache/dolphinscheduler/server/worker/utils/TaskFilesTransferUtils.java
b/dolphinscheduler-worker/src/main/java/org/apache/dolphinscheduler/server/worker/utils/TaskFilesTransferUtils.java
deleted file mode 100644
index ef956f78f4..0000000000
---
a/dolphinscheduler-worker/src/main/java/org/apache/dolphinscheduler/server/worker/utils/TaskFilesTransferUtils.java
+++ /dev/null
@@ -1,285 +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.utils;
-
-import static
org.apache.dolphinscheduler.common.constants.Constants.CRC_SUFFIX;
-
-import org.apache.dolphinscheduler.common.utils.DateUtils;
-import org.apache.dolphinscheduler.common.utils.FileUtils;
-import org.apache.dolphinscheduler.common.utils.JSONUtils;
-import org.apache.dolphinscheduler.plugin.storage.api.StorageOperator;
-import org.apache.dolphinscheduler.plugin.task.api.TaskException;
-import org.apache.dolphinscheduler.plugin.task.api.TaskExecutionContext;
-import org.apache.dolphinscheduler.plugin.task.api.enums.DataType;
-import org.apache.dolphinscheduler.plugin.task.api.enums.Direct;
-import org.apache.dolphinscheduler.plugin.task.api.model.Property;
-
-import org.apache.commons.lang3.StringUtils;
-
-import java.io.File;
-import java.io.IOException;
-import java.time.format.DateTimeFormatter;
-import java.util.ArrayList;
-import java.util.List;
-import java.util.Map;
-import java.util.stream.Collectors;
-
-import lombok.extern.slf4j.Slf4j;
-
-import org.zeroturnaround.zip.ZipUtil;
-
-import com.fasterxml.jackson.databind.JsonNode;
-
-@Slf4j
-public class TaskFilesTransferUtils {
-
- // tmp path in local path for transfer
- final static String DOWNLOAD_TMP = ".DT_TMP";
-
- // suffix of the package file
- final static String PACK_SUFFIX = "_ds_pack.zip";
-
- // root path in resource storage
- final static String RESOURCE_TAG = "DATA_TRANSFER";
-
- private TaskFilesTransferUtils() {
- throw new IllegalStateException("Utility class");
- }
-
- /**
- * upload output files to resource storage
- *
- * @param taskExecutionContext is the context of task
- * @param storageOperator is the storage operate
- * @throws TaskException TaskException
- */
- public static void uploadOutputFiles(TaskExecutionContext
taskExecutionContext,
- StorageOperator storageOperator)
throws TaskException {
- // get OUTPUT FILE parameters
- List<Property> localParamsProperty =
getFileLocalParams(taskExecutionContext, Direct.OUT);
- if (localParamsProperty.isEmpty()) {
- return;
- }
-
- List<Property> varPools = getVarPools(taskExecutionContext);
- // get map of varPools for quick search
- Map<String, Property> varPoolsMap = varPools.stream()
- .filter(property -> Direct.OUT.equals(property.getDirect()))
- .collect(Collectors.toMap(Property::getProp, x -> x));
-
- log.info("Upload output files ...");
- for (Property property : localParamsProperty) {
- // get local file path
- String path = String.format("%s/%s",
taskExecutionContext.getExecutePath(), property.getValue());
- String srcPath = packIfDir(path);
-
- // get crc file path
- String srcCRCPath = srcPath + CRC_SUFFIX;
- try {
- FileUtils.writeContent2File(FileUtils.getFileChecksum(path),
srcCRCPath);
- } catch (IOException ex) {
- throw new TaskException(ex.getMessage(), ex);
- }
-
- // get remote file path
- String resourcePath = getResourcePath(taskExecutionContext, new
File(srcPath).getName());
- String resourceCRCPath = resourcePath + CRC_SUFFIX;
- try {
- // upload file to storage
- String resourceWholePath =
-
storageOperator.getStorageFileAbsolutePath(taskExecutionContext.getTenantCode(),
resourcePath);
- String resourceCRCWholePath =
-
storageOperator.getStorageFileAbsolutePath(taskExecutionContext.getTenantCode(),
- resourceCRCPath);
- log.info("{} --- Local:{} to Remote:{}", property, srcPath,
resourceWholePath);
- storageOperator.upload(srcPath, resourceWholePath, false,
true);
- log.info("{} --- Local:{} to Remote:{}", "CRC file",
srcCRCPath, resourceCRCWholePath);
- storageOperator.upload(srcCRCPath, resourceCRCWholePath,
false, true);
- } catch (Exception ex) {
- throw new TaskException("Upload file to storage error", ex);
- }
-
- // update varPool
- Property oriProperty;
- // if the property is not in varPool, add it
- if (varPoolsMap.containsKey(property.getProp())) {
- oriProperty = varPoolsMap.get(property.getProp());
- } else {
- oriProperty = new Property(property.getProp(), Direct.OUT,
DataType.FILE, property.getValue());
- varPools.add(oriProperty);
- }
- oriProperty.setProp(String.format("%s.%s",
taskExecutionContext.getTaskName(), oriProperty.getProp()));
- oriProperty.setValue(resourcePath);
- }
- taskExecutionContext.setVarPool(JSONUtils.toJsonString(varPools));
- }
-
- /**
- * download upstream files from storage
- * only download files which are defined in the task parameters
- *
- * @param taskExecutionContext is the context of task
- * @param storageOperator is the storage operate
- * @throws TaskException task exception
- */
- public static void downloadUpstreamFiles(TaskExecutionContext
taskExecutionContext,
- StorageOperator storageOperator) {
- // get "IN FILE" parameters
- List<Property> localParamsProperty =
getFileLocalParams(taskExecutionContext, Direct.IN);
-
- if (localParamsProperty.isEmpty()) {
- return;
- }
-
- List<Property> varPools = getVarPools(taskExecutionContext);
- // get map of varPools for quick search
- Map<String, Property> varPoolsMap = varPools
- .stream()
- .filter(property -> Direct.IN.equals(property.getDirect()))
- .collect(Collectors.toMap(Property::getProp, x -> x));
-
- String executePath = taskExecutionContext.getExecutePath();
- // data path to download packaged data
- String downloadTmpPath = String.format("%s/%s", executePath,
DOWNLOAD_TMP);
-
- log.info("Download upstream files...");
- for (Property property : localParamsProperty) {
- Property inVarPool = varPoolsMap.get(property.getValue());
- if (inVarPool == null) {
- log.error("{} not in {}", property.getValue(),
varPoolsMap.keySet());
- throw new TaskException(String.format("Can not find upstream
file using %s, please check the key",
- property.getValue()));
- }
-
- String resourcePath = inVarPool.getValue();
- String targetPath = String.format("%s/%s", executePath,
property.getProp());
-
- String downloadPath;
- // If the data is packaged, download it to a special directory
(DOWNLOAD_TMP) and unpack it to the
- // targetPath
- boolean isPack = resourcePath.endsWith(PACK_SUFFIX);
- if (isPack) {
- downloadPath = String.format("%s/%s", downloadTmpPath, new
File(resourcePath).getName());
- } else {
- downloadPath = targetPath;
- }
-
- String resourceWholePath =
-
storageOperator.getStorageFileAbsolutePath(taskExecutionContext.getTenantCode(),
resourcePath);
- log.info("{} --- Remote:{} to Local:{}", property,
resourceWholePath, downloadPath);
- storageOperator.download(resourceWholePath, downloadPath, true);
-
- // unpack if the data is packaged
- if (isPack) {
- File downloadFile = new File(downloadPath);
- log.info("Unpack {} to {}", downloadPath, targetPath);
- ZipUtil.unpack(downloadFile, new File(targetPath));
- }
- }
-
- // delete DownloadTmp Folder if DownloadTmpPath exists
- try {
- org.apache.commons.io.FileUtils.deleteDirectory(new
File(downloadTmpPath));
- } catch (IOException e) {
- log.error("Delete DownloadTmpPath {} failed, this will not affect
the task status", downloadTmpPath, e);
- }
- }
-
- /**
- * get local parameters property which type is FILE and direction is equal
to direct
- *
- * @param taskExecutionContext is the context of task
- * @param direct may be Direct.IN or Direct.OUT.
- * @return List<Property>
- */
- public static List<Property> getFileLocalParams(TaskExecutionContext
taskExecutionContext, Direct direct) {
- List<Property> localParamsProperty = new ArrayList<>();
- JsonNode taskParams =
JSONUtils.parseObject(taskExecutionContext.getTaskParams());
- for (JsonNode localParam : taskParams.get("localParams")) {
- Property property = JSONUtils.parseObject(localParam.toString(),
Property.class);
-
- if (property.getDirect().equals(direct) &&
property.getType().equals(DataType.FILE)) {
- localParamsProperty.add(property);
- }
- }
- return localParamsProperty;
- }
-
- /**
- * get Resource path for manage files in storage
- *
- * @param taskExecutionContext is the context of task
- * @param fileName is the file name
- * @return resource path,
RESOURCE_TAG/DATE/ProcessDefineCode/ProcessDefineVersion_ProcessInstanceID/TaskName_TaskInstanceID_FileName
- */
- public static String getResourcePath(TaskExecutionContext
taskExecutionContext, String fileName) {
- String date =
- DateUtils.formatTimeStamp(taskExecutionContext.getEndTime(),
DateTimeFormatter.ofPattern("yyyyMMdd"));
- // get resource Folder:
RESOURCE_TAG/DATE/ProcessDefineCode/ProcessDefineVersion_ProcessInstanceID
- String resourceFolder = String.format("%s/%s/%d/%d_%d", RESOURCE_TAG,
date,
- taskExecutionContext.getWorkflowDefinitionCode(),
taskExecutionContext.getWorkflowDefinitionVersion(),
- taskExecutionContext.getWorkflowInstanceId());
- // get resource fileL: resourceFolder/TaskName_TaskInstanceID_FileName
- return String.format("%s/%s_%s_%s", resourceFolder,
taskExecutionContext.getTaskName().replace(" ", "_"),
- taskExecutionContext.getTaskInstanceId(), fileName);
- }
-
- /**
- * get varPool from taskExecutionContext
- *
- * @param taskExecutionContext is the context of task
- * @return List<Property>
- */
- public static List<Property> getVarPools(TaskExecutionContext
taskExecutionContext) {
- List<Property> varPools = new ArrayList<>();
-
- // get varPool
- String varPoolString = taskExecutionContext.getVarPool();
- if (StringUtils.isEmpty(varPoolString)) {
- return varPools;
- }
- // parse varPool
- for (JsonNode varPoolData : JSONUtils.parseArray(varPoolString)) {
- Property property = JSONUtils.parseObject(varPoolData.toString(),
Property.class);
- varPools.add(property);
- }
- return varPools;
- }
-
- /**
- * If the path is a directory, pack it and return the path of the package
- *
- * @param path is the input path, may be a file or a directory
- * @return new path
- */
- public static String packIfDir(String path) throws TaskException {
- File file = new File(path);
- if (!file.exists()) {
- throw new TaskException(String.format("%s dose not exists", path));
- }
- String newPath;
- if (file.isDirectory()) {
- newPath = file.getPath() + PACK_SUFFIX;
- log.info("Pack {} to {}", path, newPath);
- ZipUtil.pack(file, new File(newPath));
- } else {
- newPath = path;
- }
- return newPath;
- }
-}
diff --git
a/dolphinscheduler-worker/src/test/java/org/apache/dolphinscheduler/server/worker/utils/TaskFilesTransferUtilsTest.java
b/dolphinscheduler-worker/src/test/java/org/apache/dolphinscheduler/server/worker/utils/TaskFilesTransferUtilsTest.java
deleted file mode 100644
index 0e7d242424..0000000000
---
a/dolphinscheduler-worker/src/test/java/org/apache/dolphinscheduler/server/worker/utils/TaskFilesTransferUtilsTest.java
+++ /dev/null
@@ -1,255 +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.utils;
-
-import org.apache.dolphinscheduler.common.utils.DateUtils;
-import org.apache.dolphinscheduler.plugin.storage.api.StorageOperator;
-import org.apache.dolphinscheduler.plugin.task.api.TaskExecutionContext;
-import org.apache.dolphinscheduler.plugin.task.api.enums.DataType;
-import org.apache.dolphinscheduler.plugin.task.api.enums.Direct;
-import org.apache.dolphinscheduler.plugin.task.api.model.Property;
-
-import org.apache.curator.shaded.com.google.common.io.Files;
-
-import java.io.File;
-import java.io.IOException;
-import java.time.format.DateTimeFormatter;
-import java.util.List;
-
-import org.junit.jupiter.api.Assertions;
-import org.junit.jupiter.api.BeforeEach;
-import org.junit.jupiter.api.Test;
-import org.mockito.Mockito;
-import org.zeroturnaround.zip.ZipUtil;
-
-public class TaskFilesTransferUtilsTest {
-
- private final long processDefineCode = 123;
- private final int processDefineVersion = 456;
- private final int processInstanceId = 678;
- private final int taskInstanceId = 789;
- private final String taskName = "test";
-
- private final String tenantCode = "ubuntu";
-
- private long endTime;
-
- private String exceptTemplate;
-
- @BeforeEach
- void init() {
- endTime = System.currentTimeMillis();
- String date = DateUtils.formatTimeStamp(endTime,
DateTimeFormatter.ofPattern("yyyyMMdd"));
- exceptTemplate = String.format("%s/%s/%d/%d_%d/%s_%d",
- TaskFilesTransferUtils.RESOURCE_TAG,
- date,
- processDefineCode,
- processDefineVersion,
- processInstanceId,
- taskName,
- taskInstanceId);
- }
-
- @Test
- void testUploadOutputFiles() throws IOException {
- File executePath = Files.createTempDir();
- File folderPath = new File(executePath, "data");
- File file = new File(folderPath.getPath() + "/test.txt");
- if (!(folderPath.mkdirs() && file.createNewFile())) {
- return;
- }
- String varPool = "[" +
-
String.format("{\"prop\":\"folder\",\"direct\":\"OUT\",\"type\":\"FILE\",\"value\":\"%s\"},",
- folderPath.getName())
- +
- String.format("
{\"prop\":\"file\",\"direct\":\"OUT\",\"type\":\"FILE\",\"value\":\"%s/%s\"},",
- folderPath.getName(), file.getName())
- +
-
"{\"prop\":\"a\",\"direct\":\"OUT\",\"type\":\"VARCHAR\",\"value\":\"a\"}," +
-
"{\"prop\":\"b\",\"direct\":\"OUT\",\"type\":\"VARCHAR\",\"value\":\"b\"}" +
- "]";
- String taskParams = String.format("{\"localParams\": %s}", varPool);
- TaskExecutionContext taskExecutionContext =
TaskExecutionContext.builder()
- .varPool(varPool)
- .taskParams(taskParams)
- .workflowInstanceId(processInstanceId)
- .workflowDefinitionVersion(processDefineVersion)
- .workflowDefinitionCode(processDefineCode)
- .taskInstanceId(taskInstanceId)
- .taskName(taskName)
- .tenantCode(tenantCode)
- .executePath(executePath.toString())
- .endTime(endTime)
- .build();
-
- List<Property> oriProperties =
TaskFilesTransferUtils.getVarPools(taskExecutionContext);
-
- StorageOperator storageOperator = Mockito.mock(StorageOperator.class);
- TaskFilesTransferUtils.uploadOutputFiles(taskExecutionContext,
storageOperator);
- System.out.println(taskExecutionContext.getVarPool());
-
- String exceptFolder =
- String.format("%s_%s", exceptTemplate, folderPath.getName() +
TaskFilesTransferUtils.PACK_SUFFIX);
- String exceptFile = String.format("%s_%s", exceptTemplate,
file.getName());
-
- List<Property> properties =
TaskFilesTransferUtils.getVarPools(taskExecutionContext);
- Assertions.assertEquals(4, properties.size());
-
- Assertions.assertEquals(String.format("%s.%s", taskName, "folder"),
properties.get(0).getProp());
- Assertions.assertEquals(exceptFolder, properties.get(0).getValue());
-
- Assertions.assertEquals(String.format("%s.%s", taskName, "file"),
properties.get(1).getProp());
- Assertions.assertEquals(exceptFile, properties.get(1).getValue());
-
- Assertions.assertEquals(oriProperties.get(2).getProp(),
properties.get(2).getProp());
- Assertions.assertEquals(oriProperties.get(3).getValue(),
properties.get(3).getValue());
-
- }
-
- @Test
- void testDownloadUpstreamFiles() {
- File executePath = Files.createTempDir();
- String folderPath = exceptTemplate + "_folder" +
TaskFilesTransferUtils.PACK_SUFFIX;
- String filePath = exceptTemplate + "_file";
- String varPool = "[" +
- String.format(
-
"{\"prop\":\"task1.folder\",\"direct\":\"IN\",\"type\":\"FILE\",\"value\":\"%s\"},",
folderPath)
- +
- String.format("
{\"prop\":\"task2.file\",\"direct\":\"IN\",\"type\":\"FILE\",\"value\":\"%s\"},",
- filePath)
- +
-
"{\"prop\":\"a\",\"direct\":\"OUT\",\"type\":\"VARCHAR\",\"value\":\"a\"}," +
-
"{\"prop\":\"b\",\"direct\":\"OUT\",\"type\":\"VARCHAR\",\"value\":\"b\"}" +
- "]";
- String varPoolParams = "[" +
-
"{\"prop\":\"folder\",\"direct\":\"IN\",\"type\":\"FILE\",\"value\":\"task1.folder\"},"
+
- "
{\"prop\":\"file\",\"direct\":\"IN\",\"type\":\"FILE\",\"value\":\"task2.file\"}"
+
- "]";
- String taskParams = String.format("{\"localParams\": %s}",
varPoolParams);
- TaskExecutionContext taskExecutionContext =
TaskExecutionContext.builder()
- .varPool(varPool)
- .taskParams(taskParams)
- .workflowInstanceId(processInstanceId)
- .workflowDefinitionVersion(processDefineVersion)
- .workflowDefinitionCode(processDefineCode)
- .taskInstanceId(taskInstanceId)
- .taskName(taskName)
- .tenantCode(tenantCode)
- .executePath(executePath.toString())
- .endTime(endTime)
- .build();
-
- StorageOperator storageOperator = Mockito.mock(StorageOperator.class);
- Mockito.mockStatic(ZipUtil.class);
- Assertions.assertDoesNotThrow(
- () ->
TaskFilesTransferUtils.downloadUpstreamFiles(taskExecutionContext,
storageOperator));
- }
-
- @Test
- void testGetFileLocalParams() {
- String taskParams = "{\"localParams\":[" +
-
"{\"prop\":\"inputFile\",\"direct\":\"IN\",\"type\":\"FILE\",\"value\":\"task1.data\"},"
+
-
"{\"prop\":\"outputFile\",\"direct\":\"OUT\",\"type\":\"FILE\",\"value\":\"data\"},"
+
-
"{\"prop\":\"a\",\"direct\":\"IN\",\"type\":\"VARCHAR\",\"value\":\"a\"}," +
-
"{\"prop\":\"b\",\"direct\":\"OUT\",\"type\":\"VARCHAR\",\"value\":\"b\"}" +
- "]}";
- TaskExecutionContext taskExecutionContext =
Mockito.mock(TaskExecutionContext.class);
-
Mockito.when(taskExecutionContext.getTaskParams()).thenReturn(taskParams);
-
- List<Property> fileLocalParamsIn =
TaskFilesTransferUtils.getFileLocalParams(taskExecutionContext, Direct.IN);
- Assertions.assertEquals(1, fileLocalParamsIn.size());
- Assertions.assertEquals("inputFile",
fileLocalParamsIn.get(0).getProp());
- Assertions.assertEquals("task1.data",
fileLocalParamsIn.get(0).getValue());
-
- List<Property> fileLocalParamsOut =
TaskFilesTransferUtils.getFileLocalParams(taskExecutionContext, Direct.OUT);
- Assertions.assertEquals(1, fileLocalParamsOut.size());
- Assertions.assertEquals("outputFile",
fileLocalParamsOut.get(0).getProp());
- Assertions.assertEquals("data", fileLocalParamsOut.get(0).getValue());
-
- }
-
- @Test
- void testGetResourcePath() {
- String fileName = "test.txt";
- TaskExecutionContext taskExecutionContext =
Mockito.mock(TaskExecutionContext.class);
-
- Mockito.when(taskExecutionContext.getEndTime()).thenReturn(endTime);
-
-
Mockito.when(taskExecutionContext.getWorkflowDefinitionCode()).thenReturn(processDefineCode);
-
Mockito.when(taskExecutionContext.getWorkflowDefinitionVersion()).thenReturn(processDefineVersion);
-
Mockito.when(taskExecutionContext.getWorkflowInstanceId()).thenReturn(processInstanceId);
-
Mockito.when(taskExecutionContext.getTaskInstanceId()).thenReturn(taskInstanceId);
- Mockito.when(taskExecutionContext.getTaskName()).thenReturn(taskName);
-
- String except = String.format("%s_%s", exceptTemplate, fileName);
- Assertions.assertEquals(except,
TaskFilesTransferUtils.getResourcePath(taskExecutionContext, fileName));
-
- }
-
- @Test
- void testGetVarPools() {
- String varPoolsString = "[" +
-
"{\"prop\":\"input\",\"direct\":\"IN\",\"type\":\"FILE\",\"value\":\"task1.output\"}"
+
-
",{\"prop\":\"a\",\"direct\":\"IN\",\"type\":\"VARCHAR\",\"value\":\"${a}\"}" +
- "]";
- TaskExecutionContext taskExecutionContext =
Mockito.mock(TaskExecutionContext.class);
-
Mockito.when(taskExecutionContext.getVarPool()).thenReturn(varPoolsString);
-
- List<Property> varPools =
TaskFilesTransferUtils.getVarPools(taskExecutionContext);
- Assertions.assertEquals(2, varPools.size());
-
- Property varPool0 = varPools.get(0);
- Assertions.assertEquals("input", varPool0.getProp());
- Assertions.assertEquals(Direct.IN, varPool0.getDirect());
- Assertions.assertEquals(DataType.FILE, varPool0.getType());
- Assertions.assertEquals("task1.output", varPool0.getValue());
-
- Property varPool1 = varPools.get(1);
- Assertions.assertEquals("a", varPool1.getProp());
- Assertions.assertEquals(Direct.IN, varPool1.getDirect());
- Assertions.assertEquals(DataType.VARCHAR, varPool1.getType());
- Assertions.assertEquals("${a}", varPool1.getValue());
-
- Mockito.when(taskExecutionContext.getVarPool()).thenReturn("[]");
- List<Property> varPoolsEmpty =
TaskFilesTransferUtils.getVarPools(taskExecutionContext);
- Assertions.assertEquals(0, varPoolsEmpty.size());
-
- Mockito.when(taskExecutionContext.getVarPool()).thenReturn(null);
- List<Property> varPoolsNull =
TaskFilesTransferUtils.getVarPools(taskExecutionContext);
- Assertions.assertEquals(0, varPoolsNull.size());
-
- }
-
- @Test
- void testPackIfDir() throws Exception {
- File folderPath = Files.createTempDir();
- File file1 = new File(folderPath.getPath() + "/test.txt");
- File file2 = new File(folderPath.getPath() + "/test.zip");
- boolean isSuccess1 = file1.createNewFile();
- boolean isSuccess2 = file2.createNewFile();
-
- Assertions.assertTrue(isSuccess1);
- Assertions.assertTrue(isSuccess2);
-
- Assertions.assertEquals(file1.getPath(),
TaskFilesTransferUtils.packIfDir(file1.getPath()));
- Assertions.assertEquals(file2.getPath(),
TaskFilesTransferUtils.packIfDir(file2.getPath()));
-
- String expectFolderPackPath = folderPath.getPath() +
TaskFilesTransferUtils.PACK_SUFFIX;
- Assertions.assertEquals(expectFolderPackPath,
TaskFilesTransferUtils.packIfDir(folderPath.getPath()));
- }
-}
