WorkflowExecutor { @VisibleForTesting String getActiveDomain(String taskType, String[] domains) { if (domains == null || domains.length == 0) { return null; } return Arrays.stream(domains) .filter(domain -> !domain.equalsIgnoreCase("NO_DOMAIN")) .map(domain -> executionDAOFacade.getTaskPollDataByDomain(taskType, domain.trim())) .filter(Objects::nonNull) .filter(validateLastPolledTime) .findFirst() .map(PollData::getDomain) .orElse(domains[domains.length - 1].trim().equalsIgnoreCase("NO_DOMAIN") ? null : domains[domains.length - 1].trim()); } @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 testGetActiveDomain() { String taskType = "test-task"; String[] domains = new String[]{"domain1", "domain2"}; PollData pollData1 = new PollData("queue1", domains[0], "worker1", System.currentTimeMillis() - 99 * 1000); when(executionDAOFacade.getTaskPollDataByDomain(taskType, domains[0])).thenReturn(pollData1); String activeDomain = workflowExecutor.getActiveDomain(taskType, domains); assertEquals(domains[0], activeDomain); Uninterruptibles.sleepUninterruptibly(2, TimeUnit.SECONDS); PollData pollData2 = new PollData("queue2", domains[1], "worker2", System.currentTimeMillis() - 99 * 1000); when(executionDAOFacade.getTaskPollDataByDomain(taskType, domains[1])).thenReturn(pollData2); activeDomain = workflowExecutor.getActiveDomain(taskType, domains); assertEquals(domains[1], activeDomain); Uninterruptibles.sleepUninterruptibly(2, TimeUnit.SECONDS); activeDomain = workflowExecutor.getActiveDomain(taskType, domains); assertEquals(domains[1], activeDomain); domains = new String[]{""}; when(executionDAOFacade.getTaskPollDataByDomain(any(), any())).thenReturn(new PollData()); activeDomain = workflowExecutor.getActiveDomain(taskType, domains); assertNotNull(activeDomain); assertEquals("", activeDomain); domains = new String[]{}; activeDomain = workflowExecutor.getActiveDomain(taskType, domains); assertNull(activeDomain); activeDomain = workflowExecutor.getActiveDomain(taskType, null); assertNull(activeDomain); domains = new String[]{"test-domain"}; when(executionDAOFacade.getTaskPollDataByDomain(anyString(), anyString())).thenReturn(null); activeDomain = workflowExecutor.getActiveDomain(taskType, domains); assertNotNull(activeDomain); assertEquals("test-domain", activeDomain); }
@Test public void testInactiveDomains() { String taskType = "test-task"; String[] domains = new String[]{"domain1", "domain2"}; PollData pollData1 = new PollData("queue1", domains[0], "worker1", System.currentTimeMillis() - 99 * 10000); when(executionDAOFacade.getTaskPollDataByDomain(taskType, domains[0])).thenReturn(pollData1); when(executionDAOFacade.getTaskPollDataByDomain(taskType, domains[1])).thenReturn(null); String activeDomain = workflowExecutor.getActiveDomain(taskType, domains); assertEquals("domain2", activeDomain); }
@Test public void testDefaultDomain() { String taskType = "test-task"; String[] domains = new String[]{"domain1", "domain2", "NO_DOMAIN"}; PollData pollData1 = new PollData("queue1", domains[0], "worker1", System.currentTimeMillis() - 99 * 10000); when(executionDAOFacade.getTaskPollDataByDomain(taskType, domains[0])).thenReturn(pollData1); when(executionDAOFacade.getTaskPollDataByDomain(taskType, domains[1])).thenReturn(null); String activeDomain = workflowExecutor.getActiveDomain(taskType, domains); assertNull(activeDomain); }