From ec5f8eaca35fdc4ab6f473d7c6cf8f4d51306d69 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=E5=BC=A0=E4=B8=87=E4=B9=89?= Date: Fri, 21 Aug 2026 00:02:41 +0800 Subject: [PATCH 1/3] [Fix-16879] Remove parent VarPool from sub-workflow start params to prevent parameter duplication When multiple sub-processes exist in a parent workflow, the parent's accumulated VarPool was being merged into each sub-workflow's start parameters. This caused all sub-workflows to receive the same VarPool parameters, leading to parameter duplication and incorrect values in the sub-workflow's task instances. The fix removes the parent workflow's VarPool from the sub-workflow trigger parameters. Sub-workflows should only receive global params and command params from the parent. The sub-workflow's own tasks should generate their VarPool from within the sub-workflow's execution context. Closes apache/dolphinscheduler#16879 --- .../executor/plugin/subworkflow/SubWorkflowLogicTask.java | 3 +-- 1 file changed, 1 insertion(+), 2 deletions(-) diff --git a/dolphinscheduler-master/src/main/java/org/apache/dolphinscheduler/server/master/engine/executor/plugin/subworkflow/SubWorkflowLogicTask.java b/dolphinscheduler-master/src/main/java/org/apache/dolphinscheduler/server/master/engine/executor/plugin/subworkflow/SubWorkflowLogicTask.java index 4cdfdac685d5..3826dc330ab3 100644 --- a/dolphinscheduler-master/src/main/java/org/apache/dolphinscheduler/server/master/engine/executor/plugin/subworkflow/SubWorkflowLogicTask.java +++ b/dolphinscheduler-master/src/main/java/org/apache/dolphinscheduler/server/master/engine/executor/plugin/subworkflow/SubWorkflowLogicTask.java @@ -236,8 +236,7 @@ private SubWorkflowLogicTaskRuntimeContext triggerNewSubWorkflow() { final List paramList = mergeParams(asList( new ArrayList<>(deserializeVarPool(workflowInstance.getGlobalParams())), - commandParam.getCommandParams(), - new ArrayList<>(deserializeVarPool(workflowInstance.getVarPool())))); + commandParam.getCommandParams())); final WorkflowManualTriggerRequest workflowManualTriggerRequest = WorkflowManualTriggerRequest.builder() .userId(taskExecutionContext.getExecutorId()) From e3b1a4c2691e2ca99890e6233d9145b557622d6f Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=E5=BC=A0=E4=B8=87=E4=B9=89?= Date: Thu, 3 Sep 2026 12:04:11 +0800 Subject: [PATCH 2/3] Copy predecessor-scoped VarPool from taskInstance to TaskExecutionContext The TaskExecutionContextBuilder.buildTaskInstanceRelatedInfo() was not copying taskInstance.getVarPool() into TaskExecutionContext, causing taskExecutionContext.getVarPool() to return null. This fixes the sub-workflow VarPool propagation by copying the predecessor-scoped VarPool (generated by generateTaskInstanceVarPool()) into the TaskExecutionContext. Related to #18575 --- .../engine/task/execution/TaskExecutionContextBuilder.java | 2 ++ 1 file changed, 2 insertions(+) diff --git a/dolphinscheduler-master/src/main/java/org/apache/dolphinscheduler/server/master/engine/task/execution/TaskExecutionContextBuilder.java b/dolphinscheduler-master/src/main/java/org/apache/dolphinscheduler/server/master/engine/task/execution/TaskExecutionContextBuilder.java index 69765d690e18..6f150c62829f 100644 --- a/dolphinscheduler-master/src/main/java/org/apache/dolphinscheduler/server/master/engine/task/execution/TaskExecutionContextBuilder.java +++ b/dolphinscheduler-master/src/main/java/org/apache/dolphinscheduler/server/master/engine/task/execution/TaskExecutionContextBuilder.java @@ -29,6 +29,7 @@ import org.apache.dolphinscheduler.plugin.task.api.enums.TaskTimeoutStrategy; import org.apache.dolphinscheduler.plugin.task.api.model.Property; import org.apache.dolphinscheduler.plugin.task.api.parameters.resource.ResourceParametersHelper; +import org.apache.dolphinscheduler.plugin.task.api.utils.VarPoolUtils; import java.util.Map; import java.util.concurrent.TimeUnit; @@ -67,6 +68,7 @@ public TaskExecutionContextBuilder buildTaskInstanceRelatedInfo(final TaskInstan taskExecutionContext.setCpuQuota(taskInstance.getCpuQuota()); taskExecutionContext.setMemoryMax(taskInstance.getMemoryMax()); taskExecutionContext.setAppIds(taskInstance.getAppLink()); + taskExecutionContext.setVarPool(VarPoolUtils.deserializeVarPool(taskInstance.getVarPool())); return this; } From a6a6144904790b3a71c0067a0fddaf697f78694a Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=E5=BC=A0=E4=B8=87=E4=B9=89?= Date: Fri, 4 Sep 2026 22:05:42 +0800 Subject: [PATCH 3/3] [Fix-16879] Add regression tests for VarPool propagation and parameter precedence - TaskExecutionContextBuilderTest: verifies predecessor-scoped VarPool is correctly copied from TaskInstance to TaskExecutionContext - SubWorkflowLogicTaskMergeParamsTest: verifies mergeParams precedence (global < command < upstream VarPool) and conflict resolution - Make SubWorkflowLogicTask.mergeParams package-private for testability --- .../subworkflow/SubWorkflowLogicTask.java | 2 +- .../SubWorkflowLogicTaskMergeParamsTest.java | 193 ++++++++++++++++++ .../TaskExecutionContextBuilderTest.java | 127 ++++++++++++ 3 files changed, 321 insertions(+), 1 deletion(-) create mode 100644 dolphinscheduler-master/src/test/java/org/apache/dolphinscheduler/server/master/engine/executor/plugin/subworkflow/SubWorkflowLogicTaskMergeParamsTest.java create mode 100644 dolphinscheduler-master/src/test/java/org/apache/dolphinscheduler/server/master/engine/task/execution/TaskExecutionContextBuilderTest.java diff --git a/dolphinscheduler-master/src/main/java/org/apache/dolphinscheduler/server/master/engine/executor/plugin/subworkflow/SubWorkflowLogicTask.java b/dolphinscheduler-master/src/main/java/org/apache/dolphinscheduler/server/master/engine/executor/plugin/subworkflow/SubWorkflowLogicTask.java index ccbe7947cfd0..42b9e7094f78 100644 --- a/dolphinscheduler-master/src/main/java/org/apache/dolphinscheduler/server/master/engine/executor/plugin/subworkflow/SubWorkflowLogicTask.java +++ b/dolphinscheduler-master/src/main/java/org/apache/dolphinscheduler/server/master/engine/executor/plugin/subworkflow/SubWorkflowLogicTask.java @@ -261,7 +261,7 @@ private SubWorkflowLogicTaskRuntimeContext triggerNewSubWorkflow() { return SubWorkflowLogicTaskRuntimeContext.of(subWorkflowInstanceId); } - private List mergeParams(List> params) { + static List mergeParams(List> params) { if (CollectionUtils.isEmpty(params)) { return Collections.emptyList(); } diff --git a/dolphinscheduler-master/src/test/java/org/apache/dolphinscheduler/server/master/engine/executor/plugin/subworkflow/SubWorkflowLogicTaskMergeParamsTest.java b/dolphinscheduler-master/src/test/java/org/apache/dolphinscheduler/server/master/engine/executor/plugin/subworkflow/SubWorkflowLogicTaskMergeParamsTest.java new file mode 100644 index 000000000000..ccfddb09ab5c --- /dev/null +++ b/dolphinscheduler-master/src/test/java/org/apache/dolphinscheduler/server/master/engine/executor/plugin/subworkflow/SubWorkflowLogicTaskMergeParamsTest.java @@ -0,0 +1,193 @@ +/* + * 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.master.engine.executor.plugin.subworkflow; + +import static com.google.common.truth.Truth.assertThat; + +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 java.util.Arrays; +import java.util.Collections; +import java.util.List; + +import org.junit.jupiter.api.DisplayName; +import org.junit.jupiter.api.Test; + +class SubWorkflowLogicTaskMergeParamsTest { + + @Test + @DisplayName("Test mergeParams: upstream VarPool (OUT) overrides global parameter (IN)") + void testMergeParams_upstreamVarPoolOverridesGlobalParam() { + // Simulates: globalParams=[param1=global_val], varPool=[param1=upstream_val] + // The varPool (last in merge order) should win + final Property globalParam = Property.builder() + .prop("param1") + .direct(Direct.IN) + .type(DataType.VARCHAR) + .value("global_val") + .build(); + final Property upstreamOutParam = Property.builder() + .prop("param1") + .direct(Direct.OUT) + .type(DataType.VARCHAR) + .value("upstream_val") + .build(); + + final List result = SubWorkflowLogicTask.mergeParams(Arrays.asList( + Collections.singletonList(globalParam), + Collections.emptyList(), + Collections.singletonList(upstreamOutParam))); + + assertThat(result).hasSize(1); + assertThat(result.get(0).getProp()).isEqualTo("param1"); + assertThat(result.get(0).getValue()).isEqualTo("upstream_val"); + } + + @Test + @DisplayName("Test mergeParams: command (start) parameter overrides global parameter") + void testMergeParams_commandParamOverridesGlobalParam() { + // Simulates: globalParams=[param1=global_val], commandParams=[param1=start_val] + final Property globalParam = Property.builder() + .prop("param1") + .direct(Direct.IN) + .type(DataType.VARCHAR) + .value("global_val") + .build(); + final Property commandParam = Property.builder() + .prop("param1") + .direct(Direct.IN) + .type(DataType.VARCHAR) + .value("start_val") + .build(); + + final List result = SubWorkflowLogicTask.mergeParams(Arrays.asList( + Collections.singletonList(globalParam), + Collections.singletonList(commandParam), + Collections.emptyList())); + + assertThat(result).hasSize(1); + assertThat(result.get(0).getProp()).isEqualTo("param1"); + assertThat(result.get(0).getValue()).isEqualTo("start_val"); + } + + @Test + @DisplayName("Test mergeParams: upstream VarPool overrides command (start) parameter") + void testMergeParams_varPoolOverridesCommandParam() { + // Simulates: commandParams=[param1=start_val], varPool=[param1=upstream_val] + // The varPool (last in merge order) should win + final Property commandParam = Property.builder() + .prop("param1") + .direct(Direct.IN) + .type(DataType.VARCHAR) + .value("start_val") + .build(); + final Property upstreamOutParam = Property.builder() + .prop("param1") + .direct(Direct.OUT) + .type(DataType.VARCHAR) + .value("upstream_val") + .build(); + + final List result = SubWorkflowLogicTask.mergeParams(Arrays.asList( + Collections.emptyList(), + Collections.singletonList(commandParam), + Collections.singletonList(upstreamOutParam))); + + assertThat(result).hasSize(1); + assertThat(result.get(0).getProp()).isEqualTo("param1"); + assertThat(result.get(0).getValue()).isEqualTo("upstream_val"); + } + + @Test + @DisplayName("Test mergeParams: full precedence — global < command < upstream VarPool") + void testMergeParams_fullPrecedence() { + // Simulates a conflict where all three sources provide the same key + // globalParams=[param1=global_val], commandParams=[param1=start_val], varPool=[param1=upstream_val] + // The varPool (last in merge order) should win + final Property globalParam = Property.builder() + .prop("param1") + .direct(Direct.IN) + .type(DataType.VARCHAR) + .value("global_val") + .build(); + final Property commandParam = Property.builder() + .prop("param1") + .direct(Direct.IN) + .type(DataType.VARCHAR) + .value("start_val") + .build(); + final Property upstreamOutParam = Property.builder() + .prop("param1") + .direct(Direct.OUT) + .type(DataType.VARCHAR) + .value("upstream_val") + .build(); + + final List result = SubWorkflowLogicTask.mergeParams(Arrays.asList( + Collections.singletonList(globalParam), + Collections.singletonList(commandParam), + Collections.singletonList(upstreamOutParam))); + + assertThat(result).hasSize(1); + assertThat(result.get(0).getProp()).isEqualTo("param1"); + assertThat(result.get(0).getValue()).isEqualTo("upstream_val"); + } + + @Test + @DisplayName("Test mergeParams: non-conflicting parameters from all sources are preserved") + void testMergeParams_nonConflictingParamsAllPreserved() { + // globalParams=[global_only=global], commandParams=[start_only=start], varPool=[upstream_only=upstream] + final Property globalParam = Property.builder() + .prop("global_only") + .direct(Direct.IN) + .type(DataType.VARCHAR) + .value("global_val") + .build(); + final Property commandParam = Property.builder() + .prop("start_only") + .direct(Direct.IN) + .type(DataType.VARCHAR) + .value("start_val") + .build(); + final Property upstreamOutParam = Property.builder() + .prop("upstream_only") + .direct(Direct.OUT) + .type(DataType.VARCHAR) + .value("upstream_val") + .build(); + + final List result = SubWorkflowLogicTask.mergeParams(Arrays.asList( + Collections.singletonList(globalParam), + Collections.singletonList(commandParam), + Collections.singletonList(upstreamOutParam))); + + assertThat(result).hasSize(3); + // Verify each parameter is present with its expected value + assertThat( + result.stream().anyMatch(p -> "global_only".equals(p.getProp()) && "global_val".equals(p.getValue()))) + .isTrue(); + assertThat(result.stream().anyMatch(p -> "start_only".equals(p.getProp()) && "start_val".equals(p.getValue()))) + .isTrue(); + assertThat( + result.stream() + .anyMatch(p -> "upstream_only".equals(p.getProp()) && "upstream_val".equals(p.getValue()))) + .isTrue(); + } +} diff --git a/dolphinscheduler-master/src/test/java/org/apache/dolphinscheduler/server/master/engine/task/execution/TaskExecutionContextBuilderTest.java b/dolphinscheduler-master/src/test/java/org/apache/dolphinscheduler/server/master/engine/task/execution/TaskExecutionContextBuilderTest.java new file mode 100644 index 000000000000..224ae1b4e03b --- /dev/null +++ b/dolphinscheduler-master/src/test/java/org/apache/dolphinscheduler/server/master/engine/task/execution/TaskExecutionContextBuilderTest.java @@ -0,0 +1,127 @@ +/* + * 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.master.engine.task.execution; + +import static com.google.common.truth.Truth.assertThat; + +import org.apache.dolphinscheduler.dao.entity.TaskInstance; +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.dolphinscheduler.plugin.task.api.utils.VarPoolUtils; + +import java.util.Arrays; +import java.util.Collections; +import java.util.List; + +import org.junit.jupiter.api.DisplayName; +import org.junit.jupiter.api.Test; + +class TaskExecutionContextBuilderTest { + + private static final String WORKFLOW_INSTANCE_HOST = "127.0.0.1:5678"; + + @Test + @DisplayName("Test that VarPool from TaskInstance is propagated to TaskExecutionContext") + void testBuildTaskInstanceRelatedInfo_copiesVarPool() { + // Given: a TaskInstance with a predecessor-scoped VarPool containing + // an OUT parameter from an upstream task + final Property upstreamOutParam = Property.builder() + .prop("output1") + .direct(Direct.OUT) + .type(DataType.VARCHAR) + .value("upstream_value") + .build(); + final List predecessorScopedVarPool = Collections.singletonList(upstreamOutParam); + + final TaskInstance taskInstance = new TaskInstance(); + taskInstance.setId(1); + taskInstance.setName("sub_workflow_task"); + taskInstance.setVarPool(VarPoolUtils.serializeVarPool(predecessorScopedVarPool)); + + // When: building the TaskExecutionContext + final TaskExecutionContext taskExecutionContext = TaskExecutionContextBuilder.get() + .buildTaskInstanceRelatedInfo(taskInstance) + .buildWorkflowInstanceHost(WORKFLOW_INSTANCE_HOST) + .create(); + + // Then: the VarPool is correctly propagated from TaskInstance to TaskExecutionContext + assertThat(taskExecutionContext.getVarPool()).isNotNull(); + assertThat(taskExecutionContext.getVarPool()).hasSize(1); + final Property varPoolEntry = taskExecutionContext.getVarPool().get(0); + assertThat(varPoolEntry.getProp()).isEqualTo("output1"); + assertThat(varPoolEntry.getDirect()).isEqualTo(Direct.OUT); + assertThat(varPoolEntry.getValue()).isEqualTo("upstream_value"); + } + + @Test + @DisplayName("Test that null VarPool in TaskInstance results in empty list in TaskExecutionContext") + void testBuildTaskInstanceRelatedInfo_nullVarPool() { + // Given: a TaskInstance with null VarPool + final TaskInstance taskInstance = new TaskInstance(); + taskInstance.setId(1); + taskInstance.setName("sub_workflow_task"); + taskInstance.setVarPool(null); + + // When: building the TaskExecutionContext + final TaskExecutionContext taskExecutionContext = TaskExecutionContextBuilder.get() + .buildTaskInstanceRelatedInfo(taskInstance) + .buildWorkflowInstanceHost(WORKFLOW_INSTANCE_HOST) + .create(); + + // Then: the VarPool is an empty list (not null) + assertThat(taskExecutionContext.getVarPool()).isNotNull(); + assertThat(taskExecutionContext.getVarPool()).isEmpty(); + } + + @Test + @DisplayName("Test that multiple OUT parameters from predecessor are all propagated") + void testBuildTaskInstanceRelatedInfo_multipleVarPoolEntries() { + // Given: a TaskInstance with multiple predecessor-scoped VarPool entries + final Property outParam1 = Property.builder() + .prop("output1") + .direct(Direct.OUT) + .type(DataType.VARCHAR) + .value("value1") + .build(); + final Property outParam2 = Property.builder() + .prop("output2") + .direct(Direct.OUT) + .type(DataType.INTEGER) + .value("42") + .build(); + final List predecessorScopedVarPool = Arrays.asList(outParam1, outParam2); + + final TaskInstance taskInstance = new TaskInstance(); + taskInstance.setId(1); + taskInstance.setName("sub_workflow_task"); + taskInstance.setVarPool(VarPoolUtils.serializeVarPool(predecessorScopedVarPool)); + + // When: building the TaskExecutionContext + final TaskExecutionContext taskExecutionContext = TaskExecutionContextBuilder.get() + .buildTaskInstanceRelatedInfo(taskInstance) + .buildWorkflowInstanceHost(WORKFLOW_INSTANCE_HOST) + .create(); + + // Then: all VarPool entries are propagated + assertThat(taskExecutionContext.getVarPool()).hasSize(2); + assertThat(taskExecutionContext.getVarPool().get(0).getProp()).isEqualTo("output1"); + assertThat(taskExecutionContext.getVarPool().get(1).getProp()).isEqualTo("output2"); + } +}