@VisibleForTesting
    Pair<List<WorkflowTask>, Map<String, Map<String, Object>>> getDynamicForkJoinTasksAndInput(
            WorkflowTask workflowTask, WorkflowModel workflowModel)
            throws TerminateWorkflowException {
        String dynamicForkJoinTaskParam = workflowTask.getDynamicForkJoinTasksParam();
        Map<String, Object> input =
                parametersUtils.getTaskInput(
                        workflowTask.getInputParameters(), workflowModel, null, null);
        Object paramValue = input.get(dynamicForkJoinTaskParam);
        DynamicForkJoinTaskList dynamicForkJoinTaskList =
                objectMapper.convertValue(paramValue, DynamicForkJoinTaskList.class);

        if (dynamicForkJoinTaskList == null) {
            String reason =
                    String.format(
                            "Dynamic tasks could not be created. The value of %s from task's input %s has no dynamic tasks to be scheduled",
                            dynamicForkJoinTaskParam, input);
            LOGGER.error(reason);
            throw new TerminateWorkflowException(reason);
        }

        Map<String, Map<String, Object>> dynamicForkJoinTasksInput = new HashMap<>();

        List<WorkflowTask> dynamicForkJoinWorkflowTasks =
                dynamicForkJoinTaskList.getDynamicTasks().stream()
                        .peek(
                                dynamicForkJoinTask ->
                                        dynamicForkJoinTasksInput.put(
                                                dynamicForkJoinTask.getReferenceName(),
                                                dynamicForkJoinTask
                                                        .getInput())) // TODO create a custom pair
                        // collector
                        .map(
                                dynamicForkJoinTask -> {
                                    WorkflowTask dynamicForkJoinWorkflowTask = new WorkflowTask();
                                    dynamicForkJoinWorkflowTask.setTaskReferenceName(
                                            dynamicForkJoinTask.getReferenceName());
                                    dynamicForkJoinWorkflowTask.setName(
                                            dynamicForkJoinTask.getTaskName());
                                    dynamicForkJoinWorkflowTask.setType(
                                            dynamicForkJoinTask.getType());
                                    if (dynamicForkJoinWorkflowTask.getTaskDefinition() == null
                                            && StringUtils.isNotBlank(
                                                    dynamicForkJoinWorkflowTask.getName())) {
                                        dynamicForkJoinWorkflowTask.setTaskDefinition(
                                                metadataDAO.getTaskDef(
                                                        dynamicForkJoinTask.getTaskName()));
                                    }
                                    return dynamicForkJoinWorkflowTask;
                                })
                        .collect(Collectors.toCollection(LinkedList::new));

        return new ImmutablePair<>(dynamicForkJoinWorkflowTasks, dynamicForkJoinTasksInput);
    }
This method is used to get the List of dynamic workflow tasks and their input based on the
{@link WorkflowTask#getDynamicForkJoinTasksParam()}

<p><b>NOTE:</b> This method is kept for legacy reasons, new workflows should use the {@link
#getDynamicForkTasksAndInput}

@param workflowTask: The Task of type FORK_JOIN_DYNAMIC that needs to scheduled, which has
    the input parameters
@param workflowModel: The instance of the {@link WorkflowModel} which represents the workflow
    being executed.
@return {@link Pair} representing the list of dynamic fork tasks in {@link Pair#getLeft()}
    and the input for the dynamic fork tasks in {@link Pair#getRight()}
@throws TerminateWorkflowException : In case of the {@link WorkflowTask#getInputParameters()}
    does not have a payload that contains the list of the dynamic tasks
core/src/main/java/com/netflix/conductor/core/execution/mapper/ForkJoinDynamicTaskMapper.java