@Override
    public List<TaskModel> getMappedTasks(TaskMapperContext taskMapperContext)
            throws TerminateWorkflowException {
        LOGGER.debug("TaskMapperContext {} in ForkJoinDynamicTaskMapper", taskMapperContext);

        WorkflowTask workflowTask = taskMapperContext.getWorkflowTask();
        WorkflowModel workflowModel = taskMapperContext.getWorkflowModel();
        int retryCount = taskMapperContext.getRetryCount();

        List<TaskModel> mappedTasks = new LinkedList<>();
        // Get the list of dynamic tasks and the input for the tasks
        Pair<List<WorkflowTask>, Map<String, Map<String, Object>>> workflowTasksAndInputPair =
                Optional.ofNullable(workflowTask.getDynamicForkTasksParam())
                        .map(
                                dynamicForkTaskParam ->
                                        getDynamicForkTasksAndInput(
                                                workflowTask, workflowModel, dynamicForkTaskParam))
                        .orElseGet(
                                () -> getDynamicForkJoinTasksAndInput(workflowTask, workflowModel));

        List<WorkflowTask> dynForkTasks = workflowTasksAndInputPair.getLeft();
        Map<String, Map<String, Object>> tasksInput = workflowTasksAndInputPair.getRight();

        // Create Fork Task which needs to be followed by the dynamic tasks
        TaskModel forkDynamicTask = createDynamicForkTask(taskMapperContext, dynForkTasks);

        mappedTasks.add(forkDynamicTask);

        List<String> joinOnTaskRefs = new LinkedList<>();
        // Add each dynamic task to the mapped tasks and also get the last dynamic task in the list,
        // which indicates that the following task after that needs to be a join task
        for (WorkflowTask dynForkTask :
                dynForkTasks) { // TODO this is a cyclic dependency, break it out using function
            // composition
            List<TaskModel> forkedTasks =
                    taskMapperContext
                            .getDeciderService()
                            .getTasksToBeScheduled(workflowModel, dynForkTask, retryCount);

            // It's an error state if no forkedTasks can be decided upon. In the cases where we've
            // seen
            // this happen is when a dynamic task is attempting to be created here, but a task with
            // the
            // same reference name has already been created in the Workflow.
            if (forkedTasks == null || forkedTasks.isEmpty()) {
                Optional<String> existingTaskRefName =
                        workflowModel.getTasks().stream()
                                .filter(
                                        runningTask ->
                                                runningTask
                                                                .getStatus()
                                                                .equals(
                                                                        TaskModel.Status
                                                                                .IN_PROGRESS)
                                                        || runningTask.getStatus().isTerminal())
                                .map(TaskModel::getReferenceTaskName)
                                .filter(
                                        refTaskName ->
                                                refTaskName.equals(
                                                        dynForkTask.getTaskReferenceName()))
                                .findAny();

                // Construct an informative error message
                String terminateMessage =
                        "No dynamic tasks could be created for the Workflow: "
                                + workflowModel.toShortString()
                                + ", Dynamic Fork Task: "
                                + dynForkTask;
                if (existingTaskRefName.isPresent()) {
                    terminateMessage +=
                            "Attempted to create a duplicate task reference name: "
                                    + existingTaskRefName.get();
                }
                throw new TerminateWorkflowException(terminateMessage);
            }

            for (TaskModel forkedTask : forkedTasks) {
                try {
                    Map<String, Object> forkedTaskInput =
                            tasksInput.get(forkedTask.getReferenceTaskName());
                    forkedTask.addInput(forkedTaskInput);
                } catch (Exception e) {
                    String reason =
                            String.format(
                                    "Tasks could not be dynamically forked due to invalid input: %s",
                                    e.getMessage());
                    throw new TerminateWorkflowException(reason);
                }
            }
            mappedTasks.addAll(forkedTasks);
            // Get the last of the dynamic tasks so that the join can be performed once this task is
            // done
            TaskModel last = forkedTasks.get(forkedTasks.size() - 1);
            joinOnTaskRefs.add(last.getReferenceTaskName());
        }

        // From the workflow definition get the next task and make sure that it is a JOIN task.
        // The dynamic fork tasks need to be followed by a join task
        WorkflowTask joinWorkflowTask =
                workflowModel
                        .getWorkflowDefinition()
                        .getNextTask(workflowTask.getTaskReferenceName());

        if (joinWorkflowTask == null || !joinWorkflowTask.getType().equals(TaskType.JOIN.name())) {
            throw new TerminateWorkflowException(
                    "Dynamic join definition is not followed by a join task.  Check the workflow definition.");
        }

        // Create Join task
        HashMap<String, Object> joinInput = new HashMap<>();
        joinInput.put("joinOn", joinOnTaskRefs);
        TaskModel joinTask = createJoinTask(workflowModel, joinWorkflowTask, joinInput);
        mappedTasks.add(joinTask);

        return mappedTasks;
    }
This method gets the list of tasks that need to scheduled when the task to scheduled is of
type {@link TaskType#FORK_JOIN_DYNAMIC}. Creates a Fork Task, followed by the Dynamic tasks
and a final JOIN task.

<p>The definitions of the dynamic forks that need to be scheduled are available in the {@link
WorkflowTask#getInputParameters()} which are accessed using the {@link
TaskMapperContext#getWorkflowTask()}. The dynamic fork task definitions are referred by a key
value either by {@link WorkflowTask#getDynamicForkTasksParam()} or by {@link
WorkflowTask#getDynamicForkJoinTasksParam()} When creating the list of tasks to be scheduled
a set of preconditions are validated:

<ul>
  <li>If the input parameter representing the Dynamic fork tasks is available as part of
      {@link WorkflowTask#getDynamicForkTasksParam()} then the input for the dynamic task is
      validated to be a map by using {@link WorkflowTask#getDynamicForkTasksInputParamName()}
  <li>If the input parameter representing the Dynamic fork tasks is available as part of
      {@link WorkflowTask#getDynamicForkJoinTasksParam()} then the input for the dynamic
      tasks is available in the payload of the tasks definition.
  <li>A check is performed that the next following task in the {@link WorkflowDef} is a
      {@link TaskType#JOIN}
</ul>

@param taskMapperContext: A wrapper class containing the {@link WorkflowTask}, {@link
    WorkflowDef}, {@link WorkflowModel} and a string representation of the TaskId
@return List of tasks in the following order:
    <ul>
      <li>{@link TaskType#TASK_TYPE_FORK} with {@link TaskModel.Status#COMPLETED}
      <li>Might be any kind of task, but this is most cases is a UserDefinedTask with {@link
          TaskModel.Status#SCHEDULED}
      <li>{@link TaskType#JOIN} with {@link TaskModel.Status#IN_PROGRESS}
    </ul>

@throws TerminateWorkflowException In case of:
    <ul>
      <li>When the task after {@link TaskType#FORK_JOIN_DYNAMIC} is not a {@link
          TaskType#JOIN}
      <li>When the input parameters for the dynamic tasks are not of type {@link Map}
    </ul>
core/src/main/java/com/netflix/conductor/core/execution/mapper/ForkJoinDynamicTaskMapper.java