WorkflowExecutor { private boolean updateParentWorkflow(Workflow subWorkflow) { Task subWorkflowTask = executionDAOFacade.getTaskById(subWorkflow.getParentWorkflowTaskId()); Workflow parentWorkflow = executionDAOFacade.getWorkflowById(subWorkflow.getParentWorkflowId(), false); return updateParentWorkflow(subWorkflowTask, subWorkflow, parentWorkflow); } @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 testUpdateParentWorkflow() { String workflowId = "test-workflow-Id"; String subWorkflowId = "test-subWorkflow-Id"; String parentWorkflowSubWFTaskId = "test-subworkflow-taskId"; WorkflowTask subWorkflowTask = new WorkflowTask(); subWorkflowTask.setWorkflowTaskType(TaskType.SUB_WORKFLOW); subWorkflowTask.setType(TaskType.SUB_WORKFLOW.name()); subWorkflowTask.setTaskReferenceName("sub-workflow"); Task task = new Task(); task.setTaskType(subWorkflowTask.getType()); task.setTaskDefName(subWorkflowTask.getName()); task.setReferenceTaskName(subWorkflowTask.getTaskReferenceName()); task.setWorkflowInstanceId(workflowId); task.setScheduledTime(System.currentTimeMillis()); task.setTaskId(parentWorkflowSubWFTaskId); task.setStatus(Status.IN_PROGRESS); task.setRetryCount(0); task.setWorkflowTask(subWorkflowTask); task.setOutputData(new HashMap<>()); task.setSubWorkflowId(subWorkflowId); WorkflowDef def = new WorkflowDef(); def.setName("test"); Workflow parentWorkflow = new Workflow(); parentWorkflow.setWorkflowId(workflowId); parentWorkflow.setWorkflowDefinition(def); parentWorkflow.setStatus(Workflow.WorkflowStatus.RUNNING); parentWorkflow.setTasks(Arrays.asList(task)); Workflow subWorkflow = new Workflow(); subWorkflow.setWorkflowId("subworkflowId"); subWorkflow.setStatus(Workflow.WorkflowStatus.COMPLETED); subWorkflow.setParentWorkflowTaskId(parentWorkflowSubWFTaskId); subWorkflow.setWorkflowId(subWorkflowId); when(executionDAOFacade.getTaskById(anyString())).thenReturn(task); when(workflowExecutor.getWorkflow(subWorkflowId, false)).thenReturn(subWorkflow); workflowExecutor.updateParentWorkflow(task, subWorkflow, parentWorkflow); assertEquals(Status.COMPLETED, task.getStatus()); assertEquals(Workflow.WorkflowStatus.COMPLETED, subWorkflow.getStatus()); assertEquals(Workflow.WorkflowStatus.RUNNING, parentWorkflow.getStatus()); parentWorkflow.setStatus(Workflow.WorkflowStatus.FAILED); assertFalse(workflowExecutor.updateParentWorkflow(task, subWorkflow, parentWorkflow)); subWorkflow.setStatus(Workflow.WorkflowStatus.RUNNING); parentWorkflow.setStatus(Workflow.WorkflowStatus.FAILED); workflowExecutor.updateParentWorkflow(task, subWorkflow, parentWorkflow); assertEquals(Workflow.WorkflowStatus.RUNNING, subWorkflow.getStatus()); assertEquals(Status.IN_PROGRESS, task.getStatus()); assertEquals(Workflow.WorkflowStatus.RUNNING, parentWorkflow.getStatus()); subWorkflow.setStatus(Workflow.WorkflowStatus.RUNNING); parentWorkflow.setStatus(Workflow.WorkflowStatus.RUNNING); task.setStatus(Status.IN_PROGRESS); workflowExecutor.updateParentWorkflow(task, subWorkflow, parentWorkflow); assertEquals(Workflow.WorkflowStatus.RUNNING, subWorkflow.getStatus()); assertEquals(Status.IN_PROGRESS, task.getStatus()); assertEquals(Workflow.WorkflowStatus.RUNNING, parentWorkflow.getStatus()); }