Description
The DynamicForkTask class incorrectly sets both dynamic_fork_join_tasks_param and dynamic_fork_tasks_input_param_name, which violates Conductor's validation rules for FORK_JOIN_DYNAMIC tasks.
Current Behavior
When registering a workflow that uses DynamicForkTask with the standard approach (providing dynamicTasks and dynamicTasksInputs as separate parameters), the workflow registration fails with:
{
"status": 400,
"message": "Validation failed, check below errors for detail.",
"validationErrors": [{
"path": "updateWorkflowDef.arg0[0].tasks[5]",
"message": "dynamicForkJoinTasksParam or combination of dynamicForkTasksInputParamName and dynamicForkTasksParam cam be used for taskType: FORK_JOIN_DYNAMIC taskName: trigger_cf2_dynamic"
}]
}
Expected Behavior
The workflow should register successfully when using DynamicForkTask with separate task definitions and inputs.
Code to Reproduce
from conductor.client.workflow.task.dynamic_fork_task import DynamicForkTask
from conductor.client.workflow.task.join_task import JoinTask
join_task = JoinTask(task_ref_name="join")
dynamic_fork = DynamicForkTask(
task_ref_name="dynamic_fork",
tasks_param="dynamicTasks",
tasks_input_param_name="dynamicTasksInputs",
join_task=join_task,
)
dynamic_fork.input_parameters["dynamicTasks"] = [...]
dynamic_fork.input_parameters["dynamicTasksInputs"] = [...]
Root Cause
In conductor/client/workflow/task/dynamic_fork_task.py, line 23 of the to_workflow_task() method incorrectly sets:
wf_task.dynamic_fork_join_tasks_param = self.tasks_param
This should be:
wf_task.dynamic_fork_tasks_param = self.tasks_param
According to Conductor's validation logic, a FORK_JOIN_DYNAMIC task must use either:
dynamicForkJoinTasksParam alone (combined approach), OR
dynamicForkTasksParam + dynamicForkTasksInputParamName together (separate approach)
The current code mixes both approaches by setting dynamic_fork_join_tasks_param alongside dynamic_fork_tasks_input_param_name, which is invalid.
Proposed Solution
Change line 23 in dynamic_fork_task.py from:
wf_task.dynamic_fork_join_tasks_param = self.tasks_param
to:
wf_task.dynamic_fork_tasks_param = self.tasks_param
This aligns with the separate parameter approach used throughout the class.
Description
The
DynamicForkTaskclass incorrectly sets both dynamic_fork_join_tasks_param and dynamic_fork_tasks_input_param_name, which violates Conductor's validation rules for FORK_JOIN_DYNAMIC tasks.Current Behavior
When registering a workflow that uses DynamicForkTask with the standard approach (providing dynamicTasks and dynamicTasksInputs as separate parameters), the workflow registration fails with:
Expected Behavior
The workflow should register successfully when using DynamicForkTask with separate task definitions and inputs.
Code to Reproduce
Root Cause
In conductor/client/workflow/task/dynamic_fork_task.py, line 23 of the to_workflow_task() method incorrectly sets:
wf_task.dynamic_fork_join_tasks_param = self.tasks_paramThis should be:
wf_task.dynamic_fork_tasks_param = self.tasks_paramAccording to Conductor's validation logic, a FORK_JOIN_DYNAMIC task must use either:
dynamicForkJoinTasksParam alone (combined approach), OR
dynamicForkTasksParam + dynamicForkTasksInputParamName together (separate approach)
The current code mixes both approaches by setting dynamic_fork_join_tasks_param alongside dynamic_fork_tasks_input_param_name, which is invalid.
Proposed Solution
Change line 23 in dynamic_fork_task.py from:
wf_task.dynamic_fork_join_tasks_param = self.tasks_paramto:
wf_task.dynamic_fork_tasks_param = self.tasks_paramThis aligns with the separate parameter approach used throughout the class.