WorkflowExecutor { @VisibleForTesting void rollbackTasks(String workflowId, List<Task> createdTasks) { String description = "rolling back task from DAO for " + workflowId; String operation = "rollbackTasks"; try { createdTasks.forEach(task -> new RetryUtil<>().retryOnException(() -> { if (task.getTaskType().equals(SUB_WORKFLOW.name())) { executionDAOFacade.removeWorkflow(task.getSubWorkflowId(), false); } executionDAOFacade.removeTask(task.getTaskId()); return null; }, null, null, 3, description, operation)); } catch (Exception e) { String errorMsg = String.format("Error scheduling/rolling back tasks for workflow: %s", workflowId); LOGGER.error(errorMsg, e); throw new TerminateWorkflowException(errorMsg); } } @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 testRollbackTasks() { String workflowId = "workflow-id"; Task task1 = new Task(); task1.setTaskType(TaskType.SIMPLE.name()); task1.setTaskDefName("simpleTask"); task1.setReferenceTaskName("simpleTask"); task1.setWorkflowInstanceId(workflowId); task1.setScheduledTime(System.currentTimeMillis()); task1.setTaskId(IDGenerator.generate()); task1.setStatus(Status.SCHEDULED); WorkflowTask waitTask = new WorkflowTask(); waitTask.setWorkflowTaskType(TaskType.WAIT); waitTask.setType(TaskType.WAIT.name()); waitTask.setTaskReferenceName("wait"); Task task2 = new Task(); task2.setTaskType(waitTask.getType()); task2.setTaskDefName(waitTask.getName()); task2.setReferenceTaskName(waitTask.getTaskReferenceName()); task2.setWorkflowInstanceId(workflowId); task2.setScheduledTime(System.currentTimeMillis()); task2.setTaskId(IDGenerator.generate()); task2.setStatus(Status.IN_PROGRESS); task2.setRetryCount(0); task2.setWorkflowTask(waitTask); WorkflowTask subWorkflowTask = new WorkflowTask(); subWorkflowTask.setWorkflowTaskType(TaskType.SUB_WORKFLOW); subWorkflowTask.setType(TaskType.SUB_WORKFLOW.name()); subWorkflowTask.setTaskReferenceName("sub-workflow"); Task task3 = new Task(); task3.setTaskType(subWorkflowTask.getType()); task3.setTaskDefName(subWorkflowTask.getName()); task3.setReferenceTaskName(subWorkflowTask.getTaskReferenceName()); task3.setWorkflowInstanceId(workflowId); task3.setScheduledTime(System.currentTimeMillis()); task3.setTaskId(IDGenerator.generate()); task3.setStatus(Status.IN_PROGRESS); task3.setRetryCount(0); task3.setWorkflowTask(subWorkflowTask); task3.setOutputData(new HashMap<>()); task3.setSubWorkflowId(IDGenerator.generate()); AtomicInteger removeWorkflowCalledCounter = new AtomicInteger(0); doAnswer(invocation -> { removeWorkflowCalledCounter.incrementAndGet(); return null; }).when(executionDAOFacade).removeWorkflow(anyString(), anyBoolean()); AtomicInteger removeTaskCalledCounter = new AtomicInteger(0); doAnswer(invocation -> { removeTaskCalledCounter.incrementAndGet(); return null; }).when(executionDAOFacade).removeTask(anyString()); workflowExecutor.rollbackTasks(workflowId, Arrays.asList(task1, task2, task3)); assertEquals(1, removeWorkflowCalledCounter.get()); assertEquals(3, removeTaskCalledCounter.get()); }