WorkflowExecutor { @VisibleForTesting void setTaskDomains(List<Task> tasks, Workflow workflow) { Map<String, String> taskToDomain = workflow.getTaskToDomain(); if (taskToDomain != null) { String domainstr = taskToDomain.get("*"); if (StringUtils.isNotBlank(domainstr)) { String[] domains = domainstr.split(","); tasks.forEach(task -> { if (!TaskType.isSystemTask(task.getTaskType())) { task.setDomain(getActiveDomain(task.getTaskType(), domains)); } }); } tasks.forEach(task -> { if (!TaskType.isSystemTask(task.getTaskType())) { String taskDomainstr = taskToDomain.get(task.getTaskType()); if (taskDomainstr != null) { task.setDomain(getActiveDomain(task.getTaskType(), taskDomainstr.split(","))); } } }); } } @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 testTaskToDomain() { Workflow workflow = generateSampleWorkflow(); List<Task> tasks = generateSampleTasks(3); Map<String, String> taskToDomain = new HashMap<>(); taskToDomain.put("*", "mydomain"); workflow.setTaskToDomain(taskToDomain); PollData pollData1 = new PollData("queue1", "mydomain", "worker1", System.currentTimeMillis() - 99 * 100); when(executionDAOFacade.getTaskPollDataByDomain(anyString(), anyString())).thenReturn(pollData1); workflowExecutor.setTaskDomains(tasks, workflow); assertNotNull(tasks); tasks.forEach(task -> assertEquals("mydomain", task.getDomain())); }
@Test public void testTaskToDomainsPerTask() { Workflow workflow = generateSampleWorkflow(); List<Task> tasks = generateSampleTasks(2); Map<String, String> taskToDomain = new HashMap<>(); taskToDomain.put("*", "mydomain, NO_DOMAIN"); workflow.setTaskToDomain(taskToDomain); PollData pollData1 = new PollData("queue1", "mydomain", "worker1", System.currentTimeMillis() - 99 * 100); when(executionDAOFacade.getTaskPollDataByDomain(eq("task1"), anyString())).thenReturn(pollData1); when(executionDAOFacade.getTaskPollDataByDomain(eq("task2"), anyString())).thenReturn(null); workflowExecutor.setTaskDomains(tasks, workflow); assertEquals("mydomain", tasks.get(0).getDomain()); assertNull(tasks.get(1).getDomain()); }
@Test public void testTaskToDomainOverrides() { Workflow workflow = generateSampleWorkflow(); List<Task> tasks = generateSampleTasks(4); Map<String, String> taskToDomain = new HashMap<>(); taskToDomain.put("*", "mydomain"); taskToDomain.put("task2", "someInactiveDomain, NO_DOMAIN"); taskToDomain.put("task3", "someActiveDomain, NO_DOMAIN"); taskToDomain.put("task4", "someInactiveDomain, someInactiveDomain2"); workflow.setTaskToDomain(taskToDomain); PollData pollData1 = new PollData("queue1", "mydomain", "worker1", System.currentTimeMillis() - 99 * 100); PollData pollData2 = new PollData("queue2", "someActiveDomain", "worker2", System.currentTimeMillis() - 99 * 100); when(executionDAOFacade.getTaskPollDataByDomain(anyString(), eq("mydomain"))).thenReturn(pollData1); when(executionDAOFacade.getTaskPollDataByDomain(anyString(), eq("someInactiveDomain"))).thenReturn(null); when(executionDAOFacade.getTaskPollDataByDomain(anyString(), eq("someActiveDomain"))).thenReturn(pollData2); when(executionDAOFacade.getTaskPollDataByDomain(anyString(), eq("someInactiveDomain"))).thenReturn(null); workflowExecutor.setTaskDomains(tasks, workflow); assertEquals("mydomain", tasks.get(0).getDomain()); assertNull(tasks.get(1).getDomain()); assertEquals("someActiveDomain", tasks.get(2).getDomain()); assertEquals("someInactiveDomain2", tasks.get(3).getDomain()); }