WorkflowExecutor { @VisibleForTesting void completeWorkflow(Workflow wf) { LOGGER.debug("Completing workflow execution for {}", wf.getWorkflowId()); Workflow workflow = executionDAOFacade.getWorkflowById(wf.getWorkflowId(), false); if (workflow.getStatus().equals(WorkflowStatus.COMPLETED)) { queueDAO.remove(DECIDER_QUEUE, workflow.getWorkflowId()); executionDAOFacade.removeFromPendingWorkflow(workflow.getWorkflowName(), workflow.getWorkflowId()); LOGGER.debug("Workflow: {} has already been completed.", wf.getWorkflowId()); return; } if (workflow.getStatus().isTerminal()) { String msg = "Workflow is already in terminal state. Current status: " + workflow.getStatus(); throw new ApplicationException(CONFLICT, msg); } if (workflow.getWorkflowDefinition() == null) { workflow = metadataMapperService.populateWorkflowWithDefinitions(workflow); } deciderService.updateWorkflowOutput(wf, null); workflow.setStatus(WorkflowStatus.COMPLETED); workflow.setTasks(wf.getTasks()); workflow.setOutput(wf.getOutput()); workflow.setExternalOutputPayloadStoragePath(wf.getExternalOutputPayloadStoragePath()); executionDAOFacade.updateWorkflow(workflow); executionDAOFacade.updateTasks(workflow.getTasks()); LOGGER.debug("Completed workflow execution for {}", workflow.getWorkflowId()); if (StringUtils.isNotEmpty(workflow.getParentWorkflowId())) { updateParentWorkflowTask(workflow); decide(workflow.getParentWorkflowId()); } Monitors.recordWorkflowCompletion(workflow.getWorkflowName(), workflow.getEndTime() - workflow.getStartTime(), workflow.getOwnerApp()); queueDAO.remove(DECIDER_QUEUE, workflow.getWorkflowId()); if (workflow.getWorkflowDefinition().isWorkflowStatusListenerEnabled()) { workflowStatusListener.onWorkflowCompleted(workflow); } executionLockService.releaseLock(workflow.getWorkflowId()); executionLockService.deleteLock(workflow.getWorkflowId()); } @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 @SuppressWarnings("unchecked") public void testCompleteWorkflow() { WorkflowDef def = new WorkflowDef(); def.setName("test"); Workflow workflow = new Workflow(); workflow.setWorkflowDefinition(def); workflow.setWorkflowId("1"); workflow.setStatus(Workflow.WorkflowStatus.RUNNING); workflow.setOwnerApp("junit_test"); workflow.setStartTime(10L); workflow.setEndTime(100L); workflow.setOutput(Collections.EMPTY_MAP); when(executionDAOFacade.getWorkflowById(anyString(), anyBoolean())).thenReturn(workflow); AtomicInteger updateWorkflowCalledCounter = new AtomicInteger(0); doAnswer(invocation -> { updateWorkflowCalledCounter.incrementAndGet(); return null; }).when(executionDAOFacade).updateWorkflow(any()); AtomicInteger updateTasksCalledCounter = new AtomicInteger(0); doAnswer(invocation -> { updateTasksCalledCounter.incrementAndGet(); return null; }).when(executionDAOFacade).updateTasks(any()); AtomicInteger removeQueueEntryCalledCounter = new AtomicInteger(0); doAnswer(invocation -> { removeQueueEntryCalledCounter.incrementAndGet(); return null; }).when(queueDAO).remove(anyString(), anyString()); workflowExecutor.completeWorkflow(workflow); assertEquals(Workflow.WorkflowStatus.COMPLETED, workflow.getStatus()); assertEquals(1, updateWorkflowCalledCounter.get()); assertEquals(1, updateTasksCalledCounter.get()); assertEquals(1, removeQueueEntryCalledCounter.get()); verify(workflowStatusListener, times(0)).onWorkflowCompleted(any(Workflow.class)); def.setWorkflowStatusListenerEnabled(true); workflow.setStatus(Workflow.WorkflowStatus.RUNNING); workflowExecutor.completeWorkflow(workflow); verify(workflowStatusListener, times(1)).onWorkflowCompleted(any(Workflow.class)); }