WorkflowExecutor { public void retry(String workflowId) { Workflow workflow = executionDAOFacade.getWorkflowById(workflowId, true); if (!workflow.getStatus().isTerminal()) { throw new ApplicationException(CONFLICT, "Workflow is still running. status=" + workflow.getStatus()); } if (workflow.getTasks().isEmpty()) { throw new ApplicationException(CONFLICT, "Workflow has not started yet"); } Map<String, Task> retriableMap = new HashMap<>(); for (Task task : workflow.getTasks()) { switch (task.getStatus()) { case FAILED: case FAILED_WITH_TERMINAL_ERROR: case TIMED_OUT: retriableMap.put(task.getReferenceTaskName(), task); break; case CANCELED: if (task.getTaskType().equalsIgnoreCase(TaskType.JOIN.toString()) || task.getTaskType().equalsIgnoreCase(TaskType.DO_WHILE.toString())) { task.setStatus(IN_PROGRESS); } else { retriableMap.put(task.getReferenceTaskName(), task); } break; default: retriableMap.remove(task.getReferenceTaskName()); break; } } if (retriableMap.values().size() == 0) { throw new ApplicationException(CONFLICT, "There are no retriable tasks! Use restart if you want to attempt entire workflow execution again."); } workflow.setStatus(WorkflowStatus.RUNNING); workflow.setLastRetriedTime(System.currentTimeMillis()); queueDAO.push(DECIDER_QUEUE, workflow.getWorkflowId(), workflow.getPriority(), config.getSweepFrequency()); executionDAOFacade.updateWorkflow(workflow); List<Task> retriableTasks = retriableMap.values().stream() .sorted(Comparator.comparingInt(Task::getSeq)) .map(task -> taskToBeRescheduled(workflow, task)) .collect(Collectors.toList()); dedupAndAddTasks(workflow, retriableTasks); executionDAOFacade.updateTasks(workflow.getTasks()); scheduleTask(workflow, retriableTasks); decide(workflowId); if (StringUtils.isNotEmpty(workflow.getParentWorkflowId())) { updateParentWorkflow(workflow); decide(workflow.getParentWorkflowId()); } } @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(expected = ApplicationException.class) public void testRetryNonTerminalWorkflow() { Workflow workflow = new Workflow(); workflow.setWorkflowId("testRetryNonTerminalWorkflow"); workflow.setStatus(Workflow.WorkflowStatus.RUNNING); when(executionDAOFacade.getWorkflowById(anyString(), anyBoolean())).thenReturn(workflow); workflowExecutor.retry(workflow.getWorkflowId()); }
@Test(expected = ApplicationException.class) public void testRetryWorkflowNoTasks() { Workflow workflow = new Workflow(); workflow.setWorkflowId("ApplicationException"); workflow.setStatus(Workflow.WorkflowStatus.FAILED); workflow.setTasks(Collections.emptyList()); when(executionDAOFacade.getWorkflowById(anyString(), anyBoolean())).thenReturn(workflow); workflowExecutor.retry(workflow.getWorkflowId()); }