WorkflowExecutor { @VisibleForTesting boolean scheduleTask(Workflow workflow, List<Task> tasks) { List<Task> createdTasks; List<Task> tasksToBeQueued; boolean startedSystemTasks = false; try { if (tasks == null || tasks.isEmpty()) { return false; } int count = workflow.getTasks().stream() .mapToInt(Task::getSeq) .max() .orElse(0); for (Task task : tasks) { if (task.getSeq() == 0) { task.setSeq(++count); } } createdTasks = executionDAOFacade.createTasks(tasks); List<Task> systemTasks = createdTasks.stream() .filter(isSystemTask) .collect(Collectors.toList()); tasksToBeQueued = createdTasks.stream() .filter(isSystemTask.negate()) .collect(Collectors.toList()); for (Task task : systemTasks) { WorkflowSystemTask workflowSystemTask = WorkflowSystemTask.get(task.getTaskType()); if (workflowSystemTask == null) { throw new ApplicationException(NOT_FOUND, "No system task found by name " + task.getTaskType()); } if (task.getStatus() != null && !task.getStatus().isTerminal() && task.getStartTime() == 0) { task.setStartTime(System.currentTimeMillis()); } if (!workflowSystemTask.isAsync()) { try { deciderService.populateTaskData(task); 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 ApplicationException(Code.INTERNAL_ERROR, errorMsg, e); } startedSystemTasks = true; deciderService.externalizeTaskData(task); executionDAOFacade.updateTask(task); } else { tasksToBeQueued.add(task); } } } catch (Exception e) { List<String> taskIds = tasks.stream() .map(Task::getTaskId) .collect(Collectors.toList()); String errorMsg = String.format("Error scheduling tasks: %s, for workflow: %s", taskIds, workflow.getWorkflowId()); LOGGER.error(errorMsg, e); Monitors.error(className, "scheduleTask"); throw new TerminateWorkflowException(errorMsg); } try { addTaskToQueue(tasksToBeQueued); } catch (Exception e) { List<String> taskIds = tasksToBeQueued.stream() .map(Task::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(className, "scheduleTask"); } return startedSystemTasks; } @Inject  WorkflowExecutor(
        DeciderService deciderService,
        MetadataDAO metadataDAO,
        QueueDAO queueDAO,
        MetadataMapperService metadataMapperService,
        WorkflowStatusListener workflowStatusListener,
        ExecutionDAOFacade executionDAOFacade,
        Configuration config,
        ExecutionLockService executionLockService,
        ParametersUtils parametersUtils
    ); String startWorkflow(String name, Integer version, String correlationId, Map<String, Object> input, String externalInputPayloadStoragePath); String startWorkflow(String name, Integer version, String correlationId, Integer priority, Map<String, Object> input, String externalInputPayloadStoragePath); String startWorkflow(String name, Integer version, String correlationId, Map<String, Object> input, String externalInputPayloadStoragePath, String event); String startWorkflow(String name, Integer version, String correlationId, Integer priority, Map<String, Object> input, String externalInputPayloadStoragePath, String event); String startWorkflow(
            String name,
            Integer version,
            String correlationId,
            Map<String, Object> input,
            String externalInputPayloadStoragePath,
            String event,
            Map<String, String> taskToDomain
    ); String startWorkflow(
            String name,
            Integer version,
            String correlationId,
            Integer priority,
            Map<String, Object> input,
            String externalInputPayloadStoragePath,
            String event,
            Map<String, String> taskToDomain
    ); String startWorkflow(
            String name,
            Integer version,
            Map<String, Object> input,
            String externalInputPayloadStoragePath,
            String correlationId,
            String parentWorkflowId,
            String parentWorkflowTaskId,
            String event
    ); String startWorkflow(
            WorkflowDef workflowDefinition,
            Map<String, Object> workflowInput,
            String externalInputPayloadStoragePath,
            String correlationId,
            String event,
            Map<String, String> taskToDomain
    ); String startWorkflow(
            WorkflowDef workflowDefinition,
            Map<String, Object> workflowInput,
            String externalInputPayloadStoragePath,
            String correlationId,
            Integer priority,
            String event,
            Map<String, String> taskToDomain
    ); String startWorkflow(
            String name,
            Integer version,
            Map<String, Object> workflowInput,
            String externalInputPayloadStoragePath,
            String correlationId,
            String parentWorkflowId,
            String parentWorkflowTaskId,
            String event,
            Map<String, String> taskToDomain
    ); String startWorkflow(
            String name,
            Integer version,
            Map<String, Object> workflowInput,
            String externalInputPayloadStoragePath,
            String correlationId,
            Integer priority,
            String parentWorkflowId,
            String parentWorkflowTaskId,
            String event,
            Map<String, String> taskToDomain
    ); String startWorkflow(
            WorkflowDef workflowDefinition,
            Map<String, Object> workflowInput,
            String externalInputPayloadStoragePath,
            String correlationId,
            Integer priority,
            String parentWorkflowId,
            String parentWorkflowTaskId,
            String event,
            Map<String, String> taskToDomain
    ); void resetCallbacksForWorkflow(String workflowId); String rerun(RerunWorkflowRequest request); void rewind(String workflowId, boolean useLatestDefinitions); void retry(String workflowId); Task getPendingTaskByWorkflow(String taskReferenceName, String workflowId); void terminateWorkflow(String workflowId, String reason); void terminateWorkflow(Workflow workflow, String reason, String failureWorkflow); void updateTask(TaskResult taskResult); Task getTask(String taskId); List<Task> getTasks(String taskType, String startKey, int count); List<Workflow> getRunningWorkflows(String workflowName, int version); List<String> getWorkflows(String name, Integer version, Long startTime, Long endTime); List<String> getRunningWorkflowIds(String workflowName, int version); boolean decide(String workflowId); void pauseWorkflow(String workflowId); void resumeWorkflow(String workflowId); void skipTaskFromWorkflow(String workflowId, String taskReferenceName, SkipTaskRequest skipTaskRequest); Workflow getWorkflow(String workflowId, boolean includeTasks); void addTaskToQueue(Task task); void executeSystemTask(WorkflowSystemTask systemTask, String taskId, int callbackTime); void scheduleNextIteration(Task loopTask, Workflow workflow); TaskDef getTaskDefinition(Task task); static final String DECIDER_QUEUE; }
@Test public void testScheduleTask() { AtomicBoolean httpTaskExecuted = new AtomicBoolean(false); AtomicBoolean http2TaskExecuted = new AtomicBoolean(false); new Wait(); new WorkflowSystemTask("HTTP") { @Override public boolean isAsync() { return true; } @Override public void start(Workflow workflow, Task task, WorkflowExecutor executor) { httpTaskExecuted.set(true); task.setStatus(Status.COMPLETED); super.start(workflow, task, executor); } }; new WorkflowSystemTask("HTTP2") { @Override public void start(Workflow workflow, Task task, WorkflowExecutor executor) { http2TaskExecuted.set(true); task.setStatus(Status.COMPLETED); super.start(workflow, task, executor); } }; Workflow workflow = new Workflow(); workflow.setWorkflowId("1"); List<Task> tasks = new LinkedList<>(); WorkflowTask taskToSchedule = new WorkflowTask(); taskToSchedule.setWorkflowTaskType(TaskType.USER_DEFINED); taskToSchedule.setType("HTTP"); WorkflowTask taskToSchedule2 = new WorkflowTask(); taskToSchedule2.setWorkflowTaskType(TaskType.USER_DEFINED); taskToSchedule2.setType("HTTP2"); WorkflowTask wait = new WorkflowTask(); wait.setWorkflowTaskType(TaskType.WAIT); wait.setType("WAIT"); wait.setTaskReferenceName("wait"); Task task1 = new Task(); task1.setTaskType(taskToSchedule.getType()); task1.setTaskDefName(taskToSchedule.getName()); task1.setReferenceTaskName(taskToSchedule.getTaskReferenceName()); task1.setWorkflowInstanceId(workflow.getWorkflowId()); task1.setCorrelationId(workflow.getCorrelationId()); task1.setScheduledTime(System.currentTimeMillis()); task1.setTaskId(IDGenerator.generate()); task1.setInputData(new HashMap<>()); task1.setStatus(Status.SCHEDULED); task1.setRetryCount(0); task1.setCallbackAfterSeconds(taskToSchedule.getStartDelay()); task1.setWorkflowTask(taskToSchedule); Task task2 = new Task(); task2.setTaskType(Wait.NAME); task2.setTaskDefName(taskToSchedule.getName()); task2.setReferenceTaskName(taskToSchedule.getTaskReferenceName()); task2.setWorkflowInstanceId(workflow.getWorkflowId()); task2.setCorrelationId(workflow.getCorrelationId()); task2.setScheduledTime(System.currentTimeMillis()); task2.setInputData(new HashMap<>()); task2.setTaskId(IDGenerator.generate()); task2.setStatus(Status.IN_PROGRESS); task2.setWorkflowTask(taskToSchedule); Task task3 = new Task(); task3.setTaskType(taskToSchedule2.getType()); task3.setTaskDefName(taskToSchedule.getName()); task3.setReferenceTaskName(taskToSchedule.getTaskReferenceName()); task3.setWorkflowInstanceId(workflow.getWorkflowId()); task3.setCorrelationId(workflow.getCorrelationId()); task3.setScheduledTime(System.currentTimeMillis()); task3.setTaskId(IDGenerator.generate()); task3.setInputData(new HashMap<>()); task3.setStatus(Status.SCHEDULED); task3.setRetryCount(0); task3.setCallbackAfterSeconds(taskToSchedule.getStartDelay()); task3.setWorkflowTask(taskToSchedule); tasks.add(task1); tasks.add(task2); tasks.add(task3); when(executionDAOFacade.createTasks(tasks)).thenReturn(tasks); AtomicInteger startedTaskCount = new AtomicInteger(0); doAnswer(invocation -> { startedTaskCount.incrementAndGet(); return null; }).when(executionDAOFacade) .updateTask(any()); AtomicInteger queuedTaskCount = new AtomicInteger(0); final Answer answer = invocation -> { String queueName = invocation.getArgument(0, String.class); System.out.println(queueName); queuedTaskCount.incrementAndGet(); return null; }; doAnswer(answer).when(queueDAO).push(any(), any(), anyLong()); doAnswer(answer).when(queueDAO).push(any(), any(), anyInt(), anyLong()); boolean stateChanged = workflowExecutor.scheduleTask(workflow, tasks); assertEquals(2, startedTaskCount.get()); assertEquals(1, queuedTaskCount.get()); assertTrue(stateChanged); assertFalse(httpTaskExecuted.get()); assertTrue(http2TaskExecuted.get()); }
@Test(expected = TerminateWorkflowException.class) public void testScheduleTaskFailure() { Workflow workflow = new Workflow(); workflow.setWorkflowId("wid_01"); List<Task> tasks = new LinkedList<>(); Task task1 = new Task(); task1.setTaskType(TaskType.TASK_TYPE_SIMPLE); task1.setTaskDefName("task_1"); task1.setReferenceTaskName("task_1"); task1.setWorkflowInstanceId(workflow.getWorkflowId()); task1.setTaskId("tid_01"); task1.setStatus(Status.SCHEDULED); task1.setRetryCount(0); tasks.add(task1); when(executionDAOFacade.createTasks(tasks)).thenThrow(new RuntimeException()); workflowExecutor.scheduleTask(workflow, tasks); }
@Test public void testQueueFailuresDuringScheduleTask() { Workflow workflow = new Workflow(); workflow.setWorkflowId("wid_01"); List<Task> tasks = new LinkedList<>(); Task task1 = new Task(); task1.setTaskType(TaskType.TASK_TYPE_SIMPLE); task1.setTaskDefName("task_1"); task1.setReferenceTaskName("task_1"); task1.setWorkflowInstanceId(workflow.getWorkflowId()); task1.setTaskId("tid_01"); task1.setStatus(Status.SCHEDULED); task1.setRetryCount(0); tasks.add(task1); when(executionDAOFacade.createTasks(tasks)).thenReturn(tasks); doThrow(new RuntimeException()).when(queueDAO).push(anyString(), anyString(), anyInt(), anyLong()); assertFalse(workflowExecutor.scheduleTask(workflow, tasks)); }