Removed unused params in TaskExecutionContext (#17195)

Author: ruanwenjun

dolphinscheduler-master/src/main/java/org/apache/dolphinscheduler/server/master/engine/task/runnable/TaskExecutionContextBuilder.java

@@ -102,11 +102,9 @@ public TaskExecutionContextBuilder buildProcessInstanceRelatedInfo(final Workflo
         taskExecutionContext.setScheduleTime(DateUtils.dateToTimeStamp(workflowInstance.getScheduleTime()));
         taskExecutionContext.setGlobalParams(workflowInstance.getGlobalParams());
         taskExecutionContext.setExecutorId(workflowInstance.getExecutorId());
-        taskExecutionContext.setCmdTypeIfComplement(workflowInstance.getCmdTypeIfComplement().getCode());
         taskExecutionContext.setTenantCode(workflowInstance.getTenantCode());
         taskExecutionContext.setWorkflowDefinitionCode(workflowInstance.getWorkflowDefinitionCode());
         taskExecutionContext.setWorkflowDefinitionVersion(workflowInstance.getWorkflowDefinitionVersion());
-        taskExecutionContext.setProjectCode(workflowInstance.getProjectCode());
         return this;
     }
 
@@ -128,27 +126,14 @@ public TaskExecutionContextBuilder buildK8sTaskRelatedInfo(final K8sTaskExecutio
     }
 
     /**
-     * build global and local params
+     * The runtime params, include local params from task, global params from workflow, startup params from command, varpool params from pre-task, built-in params from system  
      *
-     * @param propertyMap
-     * @return
      */
     public TaskExecutionContextBuilder buildPrepareParams(final Map<String, Property> propertyMap) {
         taskExecutionContext.setPrepareParamsMap(propertyMap);
         return this;
     }
 
-    /**
-     * build business params
-     *
-     * @param businessParamsMap
-     * @return
-     */
-    public TaskExecutionContextBuilder buildBusinessParams(final Map<String, Property> businessParamsMap) {
-        taskExecutionContext.setParamsMap(businessParamsMap);
-        return this;
-    }
-
     public TaskExecutionContextBuilder buildWorkflowInstanceHost(final String masterHost) {
         taskExecutionContext.setWorkflowInstanceHost(masterHost);
         return this;

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

@@ -85,7 +85,7 @@ public TaskExecutionContext createTaskExecutionContext(TaskExecutionContextCreat
                 .buildTaskDefinitionRelatedInfo(request.getTaskDefinition())
                 .buildProcessInstanceRelatedInfo(request.getWorkflowInstance())
                 .buildResourceParameters(getResourceParameters(taskInstance))
-                .buildBusinessParams(getBusinessParams(workflowInstance))
+                // todo: use TaskRuntimeParameters to replace Map<String, Property> in TaskExecutionContext
                 .buildPrepareParams(getPrepareParams(taskInstance, workflowInstance, workflowDefinition, project))
                 .buildK8sTaskRelatedInfo(getK8sTaskExecutionContext(taskInstance))
                 .create();
@@ -152,10 +152,6 @@ private K8sTaskExecutionContext getK8sTaskExecutionContext(final TaskInstance ta
         return k8sTaskExecutionContext;
     }
 
-    private Map<String, Property> getBusinessParams(final WorkflowInstance workflowInstance) {
-        return curingParamsService.preBuildBusinessParams(workflowInstance);
-    }
-
     private Map<String, Property> getPrepareParams(final TaskInstance taskInstance,
                                                    final WorkflowInstance workflowInstance,
                                                    final WorkflowDefinition workflowDefinition,

dolphinscheduler-task-plugin/dolphinscheduler-task-api/src/main/java/org/apache/dolphinscheduler/plugin/task/api/AbstractCommandExecutor.java

@@ -93,10 +93,6 @@ public AbstractCommandExecutor(Consumer<LinkedBlockingQueue<String>> logHandler,
         this.logBuffer = new LinkedBlockingQueue<>();
         this.logBuffer.add(EMPTY_STRING);
 
-        if (this.taskRequest != null) {
-            // set logBufferEnable=true if the task uses logHandler and logBuffer to buffer log messages
-            this.taskRequest.setLogBufferEnable(true);
-        }
     }
 
     // todo: We need to build the IShellActuator in outer class, since different task may have specific logic to build

dolphinscheduler-task-plugin/dolphinscheduler-task-api/src/main/java/org/apache/dolphinscheduler/plugin/task/api/AbstractYarnTask.java

@@ -25,9 +25,9 @@
 import org.apache.dolphinscheduler.plugin.task.api.shell.IShellInterceptorBuilder;
 import org.apache.dolphinscheduler.plugin.task.api.shell.ShellInterceptorBuilderFactory;
 import org.apache.dolphinscheduler.plugin.task.api.utils.LogUtils;
+import org.apache.dolphinscheduler.plugin.task.api.utils.ParameterUtils;
 
 import java.util.List;
-import java.util.Map;
 
 import lombok.extern.slf4j.Slf4j;
 
@@ -46,7 +46,7 @@ public AbstractYarnTask(TaskExecutionContext taskRequest) {
     public void handle(TaskCallBack taskCallBack) throws TaskException {
         try {
             IShellInterceptorBuilder shellActuatorBuilder = ShellInterceptorBuilderFactory.newBuilder()
-                    .properties(getProperties())
+                    .properties(ParameterUtils.convert(taskRequest.getPrepareParamsMap()))
                     // todo: do we need to move the replace to subclass?
                     .appendScript(getScript().replaceAll("\\r\\n", System.lineSeparator()));
             // SHELL task exit code
@@ -109,9 +109,4 @@ public List<String> getApplicationIds() throws TaskException {
      * Get the script used to bootstrap the task
      */
     protected abstract String getScript();
-
-    /**
-     * Get the properties of the task used to replace the placeholders in the script.
-     */
-    protected abstract Map<String, String> getProperties();
 }

dolphinscheduler-task-plugin/dolphinscheduler-task-api/src/main/java/org/apache/dolphinscheduler/plugin/task/api/TaskExecutionContext.java

@@ -80,26 +80,17 @@ public class TaskExecutionContext implements Serializable {
 
     private int executorId;
 
-    private int cmdTypeIfComplement;
-
     private String tenantCode;
 
     private int workflowDefinitionId;
 
-    private int projectId;
-
-    private long projectCode;
-
     private String taskParams;
 
     private String environmentConfig;
 
     /**
-     * definedParams
-     * // todo: we need to rename definedParams, prepareParamsMap, paramsMap, this is confusing
+     * Include local params, global params and system built-in params
      */
-    private Map<String, String> definedParams;
-
     private Map<String, Property> prepareParamsMap;
 
     // Please use task instanceId
@@ -126,16 +117,12 @@ public class TaskExecutionContext implements Serializable {
 
     private int dryRun;
 
-    private Map<String, Property> paramsMap;
-
     private Integer cpuQuota;
 
     private Integer memoryMax;
 
     private int testFlag;
 
-    private boolean logBufferEnable;
-
     private int dispatchFailTimes;
 
     private boolean failover;

dolphinscheduler-task-plugin/dolphinscheduler-task-flink-stream/src/main/java/org/apache/dolphinscheduler/plugin/task/flink/FlinkStreamTask.java

@@ -28,7 +28,6 @@
 
 import java.io.IOException;
 import java.util.List;
-import java.util.Map;
 import java.util.stream.Collectors;
 
 import lombok.extern.slf4j.Slf4j;
@@ -70,11 +69,6 @@ protected String getScript() {
         return args.stream().collect(Collectors.joining(" "));
     }
 
-    @Override
-    protected Map<String, String> getProperties() {
-        return taskExecutionContext.getDefinedParams();
-    }
-
     @Override
     public AbstractParameters getParameters() {
         return flinkParameters;

dolphinscheduler-task-plugin/dolphinscheduler-task-flink/src/main/java/org/apache/dolphinscheduler/plugin/task/flink/FlinkTask.java

@@ -24,7 +24,6 @@
 import org.apache.dolphinscheduler.plugin.task.api.parameters.AbstractParameters;
 
 import java.util.List;
-import java.util.Map;
 import java.util.regex.Matcher;
 import java.util.regex.Pattern;
 import java.util.stream.Collectors;
@@ -79,11 +78,6 @@ protected String getScript() {
         return args.stream().collect(Collectors.joining(" "));
     }
 
-    @Override
-    protected Map<String, String> getProperties() {
-        return taskExecutionContext.getDefinedParams();
-    }
-
     @Override
     public AbstractParameters getParameters() {
         return flinkParameters;

dolphinscheduler-task-plugin/dolphinscheduler-task-mr/src/main/java/org/apache/dolphinscheduler/plugin/task/mr/MapReduceTask.java

@@ -91,11 +91,6 @@ protected String getScript() {
         return args.stream().collect(Collectors.joining(" "));
     }
 
-    @Override
-    protected Map<String, String> getProperties() {
-        return taskExecutionContext.getDefinedParams();
-    }
-
     @Override
     public AbstractParameters getParameters() {
         return mapreduceParameters;

dolphinscheduler-task-plugin/dolphinscheduler-task-python/src/main/java/org/apache/dolphinscheduler/plugin/task/python/PythonTask.java

@@ -98,6 +98,7 @@ public void handle(TaskCallBack taskCallBack) throws TaskException {
             setProcessId(taskResponse.getProcessId());
             setTaskOutputParams(shellCommandExecutor.getTaskOutputParams());
             pythonParameters.dealOutParam(shellCommandExecutor.getTaskOutputParams());
+            taskRequest.setVarPool(JSONUtils.toJsonString(pythonParameters.getVarPool()));
         } catch (Exception e) {
             log.error("python task failure", e);
             setExitStatusCode(TaskConstants.EXIT_CODE_FAILURE);

dolphinscheduler-task-plugin/dolphinscheduler-task-spark/src/main/java/org/apache/dolphinscheduler/plugin/task/spark/SparkTask.java

@@ -112,11 +112,6 @@ protected String getScript() {
         return args.stream().collect(Collectors.joining(" "));
     }
 
-    @Override
-    protected Map<String, String> getProperties() {
-        return ParameterUtils.convert(taskExecutionContext.getPrepareParamsMap());
-    }
-
     /**
      * build spark options
      *