WorkflowExecutor { public void executeSystemTask(WorkflowSystemTask systemTask, String taskId, int callbackTime) { try { Task task = executionDAOFacade.getTaskById(taskId); if (task == null) { LOGGER.error("TaskId: {} could not be found while executing SystemTask", taskId); return; } LOGGER.debug("Task: {} fetched from execution DAO for taskId: {}", task, taskId); String queueName = QueueUtils.getQueueName(task); if (task.getStatus().isTerminal()) { LOGGER.info("Task {}/{} was already completed.", task.getTaskType(), task.getTaskId()); queueDAO.remove(queueName, task.getTaskId()); return; } String workflowId = task.getWorkflowInstanceId(); Workflow workflow = executionDAOFacade.getWorkflowById(workflowId, true); if (task.getStartTime() == 0) { task.setStartTime(System.currentTimeMillis()); Monitors.recordQueueWaitTime(task.getTaskDefName(), task.getQueueWaitTime()); } if (workflow.getStatus().isTerminal()) { LOGGER.info("Workflow {} has been completed for {}/{}", workflow.getWorkflowId(), systemTask.getName(), task.getTaskId()); if (!task.getStatus().isTerminal()) { task.setStatus(CANCELED); } executionDAOFacade.updateTask(task); queueDAO.remove(queueName, task.getTaskId()); return; } if (task.getStatus().equals(SCHEDULED)) { if (executionDAOFacade.exceedsInProgressLimit(task)) { LOGGER.warn("Concurrent Execution limited for {}:{}", taskId, task.getTaskDefName()); queueDAO.postpone(queueName, taskId, task.getWorkflowPriority(), queueTaskMessagePostponeSeconds); return; } if (task.getRateLimitPerFrequency() > 0 && executionDAOFacade.exceedsRateLimitPerFrequency(task, metadataDAO.getTaskDef(task.getTaskDefName()))) { LOGGER.warn("RateLimit Execution limited for {}:{}, limit:{}", taskId, task.getTaskDefName(), task.getRateLimitPerFrequency()); queueDAO.postpone(queueName, taskId, task.getWorkflowPriority(), queueTaskMessagePostponeSeconds); return; } } LOGGER.debug("Executing {}/{}-{}", task.getTaskType(), task.getTaskId(), task.getStatus()); if (task.getStatus() == SCHEDULED || !systemTask.isAsyncComplete(task)) { task.setPollCount(task.getPollCount() + 1); executionDAOFacade.updateTask(task); } deciderService.populateTaskData(task); if (systemTask.isAsyncComplete(task) && task.getStatus() != SCHEDULED) { queueDAO.remove(QueueUtils.getQueueName(task), task.getTaskId()); return; } switch (task.getStatus()) { case SCHEDULED: systemTask.start(workflow, task, this); break; case IN_PROGRESS: systemTask.execute(workflow, task, this); break; default: break; } if (!task.getStatus().isTerminal()) { task.setCallbackAfterSeconds(callbackTime); } updateTask(new TaskResult(task)); LOGGER.debug("Done Executing {}/{}-{} output={}", task.getTaskType(), task.getTaskId(), task.getStatus(), task.getOutputData().toString()); } catch (Exception e) { Monitors.error(className, "executeSystemTask"); LOGGER.error("Error executing system task - {}, with id: {}", systemTask, taskId, e); } } @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 testExecuteSystemTask() { String workflowId = "workflow-id"; Wait wait = new Wait(); String task1Id = IDGenerator.generate(); Task task1 = new Task(); task1.setTaskType(TaskType.WAIT.name()); task1.setReferenceTaskName("waitTask"); task1.setWorkflowInstanceId(workflowId); task1.setScheduledTime(System.currentTimeMillis()); task1.setTaskId(task1Id); task1.setStatus(Status.SCHEDULED); Workflow workflow = new Workflow(); workflow.setWorkflowId(workflowId); workflow.setStatus(Workflow.WorkflowStatus.RUNNING); when(executionDAOFacade.getTaskById(anyString())).thenReturn(task1); when(executionDAOFacade.getWorkflowById(anyString(), anyBoolean())).thenReturn(workflow); workflowExecutor.executeSystemTask(wait, task1Id,30); assertEquals(Status.IN_PROGRESS, task1.getStatus()); }
@Test public void testExecuteSystemTaskWithAsyncComplete() { String workflowId = "workflow-id"; Terminate terminate = new Terminate(); String task1Id = IDGenerator.generate(); Task task1 = new Task(); task1.setTaskType(TaskType.WAIT.name()); task1.setReferenceTaskName("waitTask"); task1.setWorkflowInstanceId(workflowId); task1.setScheduledTime(System.currentTimeMillis()); task1.setTaskId(task1Id); task1.getInputData().put("asyncComplete", true); task1.setStatus(Status.IN_PROGRESS); Workflow workflow = new Workflow(); workflow.setWorkflowId(workflowId); workflow.setStatus(Workflow.WorkflowStatus.RUNNING); when(executionDAOFacade.getTaskById(anyString())).thenReturn(task1); when(executionDAOFacade.getWorkflowById(anyString(), anyBoolean())).thenReturn(workflow); workflowExecutor.executeSystemTask(terminate, task1Id,30); assertEquals(Status.IN_PROGRESS, task1.getStatus()); }