@VisibleForTesting
    boolean scheduleTask(WorkflowModel workflow, List<TaskModel> tasks) {
        List<TaskModel> tasksToBeQueued;
        boolean startedSystemTasks = false;

        try {
            if (tasks == null || tasks.isEmpty()) {
                return false;
            }

            // Get the highest seq number
            int count = workflow.getTasks().stream().mapToInt(TaskModel::getSeq).max().orElse(0);

            for (TaskModel task : tasks) {
                if (task.getSeq() == 0) { // Set only if the seq was not set
                    task.setSeq(++count);
                }
            }

            // metric to track the distribution of number of tasks within a workflow
            Monitors.recordNumTasksInWorkflow(
                    workflow.getTasks().size() + tasks.size(),
                    workflow.getWorkflowName(),
                    String.valueOf(workflow.getWorkflowVersion()));

            // Save the tasks in the DAO
            executionDAOFacade.createTasks(tasks);

            List<TaskModel> systemTasks =
                    tasks.stream()
                            .filter(task -> systemTaskRegistry.isSystemTask(task.getTaskType()))
                            .collect(Collectors.toList());

            tasksToBeQueued =
                    tasks.stream()
                            .filter(task -> !systemTaskRegistry.isSystemTask(task.getTaskType()))
                            .collect(Collectors.toList());

            // Traverse through all the system tasks, start the sync tasks, in case of async queue
            // the tasks
            for (TaskModel task : systemTasks) {
                WorkflowSystemTask workflowSystemTask = systemTaskRegistry.get(task.getTaskType());
                if (workflowSystemTask == null) {
                    throw new NotFoundException(
                            "No system task found by name %s", task.getTaskType());
                }
                if (task.getStatus() != null
                        && !task.getStatus().isTerminal()
                        && task.getStartTime() == 0) {
                    task.setStartTime(System.currentTimeMillis());
                }
                if (!workflowSystemTask.isAsync()) {
                    try {
                        // start execution of synchronous system tasks
                        workflowSystemTask.start(workflow, task, this);
                    } catch (Exception e) {
                        String errorMsg =
                                String.format(
                                        "Unable to start system task: %s, {id: %s, name: %s}",
                                        task.getTaskType(),
                                        task.getTaskId(),
                                        task.getTaskDefName());
                        throw new NonTransientException(errorMsg, e);
                    }
                    startedSystemTasks = true;
                    executionDAOFacade.updateTask(task);
                } else {
                    tasksToBeQueued.add(task);
                }
            }

        } catch (Exception e) {
            List<String> taskIds =
                    tasks.stream().map(TaskModel::getTaskId).collect(Collectors.toList());
            String errorMsg =
                    String.format(
                            "Error scheduling tasks: %s, for workflow: %s",
                            taskIds, workflow.getWorkflowId());
            LOGGER.error(errorMsg, e);
            Monitors.error(CLASS_NAME, "scheduleTask");
            throw new TerminateWorkflowException(errorMsg);
        }

        // On addTaskToQueue failures, ignore the exceptions and let WorkflowRepairService take care
        // of republishing the messages to the queue.
        try {
            addTaskToQueue(tasksToBeQueued);
        } catch (Exception e) {
            List<String> taskIds =
                    tasksToBeQueued.stream().map(TaskModel::getTaskId).collect(Collectors.toList());
            String errorMsg =
                    String.format(
                            "Error pushing tasks to the queue: %s, for workflow: %s",
                            taskIds, workflow.getWorkflowId());
            LOGGER.warn(errorMsg, e);
            Monitors.error(CLASS_NAME, "scheduleTask");
        }
        return startedSystemTasks;
    }
Gets the active domain from the list of domains where the task is to be queued. The domain
list must be ordered. In sequence, check if any worker has polled for last
`activeWorkerLastPollMs`, if so that is the Active domain. When no active domains are found:
<li>If NO_DOMAIN token is provided, return null.
<li>Else, return last domain from list.

@param taskType the taskType of the task for which active domain is to be found
@param domains the array of domains for the task. (Must contain atleast one element).
@return the active domain where the task will be queued
core/src/main/java/com/netflix/conductor/core/execution/WorkflowExecutor.java