ExecutionDAOFacade { public List<Workflow> getWorkflowsByCorrelationId(String workflowName, String correlationId, boolean includeTasks) { if (!executionDAO.canSearchAcrossWorkflows()) { String query = "correlationId='" + correlationId + "' AND workflowType='" + workflowName + "'"; SearchResult<String> result = indexDAO.searchWorkflows(query, "*", 0, 1000, null); return result.getResults().stream() .parallel() .map(workflowId -> { try { return getWorkflowById(workflowId, includeTasks); } catch (ApplicationException e) { LOGGER.error("Error getting the workflow: {} for correlationId: {} from datastore/index", workflowId, correlationId, e); return null; } }) .filter(Objects::nonNull) .collect(Collectors.toList()); } return executionDAO.getWorkflowsByCorrelationId(workflowName,correlationId, includeTasks); } @Inject  ExecutionDAOFacade(ExecutionDAO executionDAO, QueueDAO queueDAO, IndexDAO indexDAO,
        RateLimitingDAO rateLimitingDao, PollDataDAO pollDataDAO, ObjectMapper objectMapper, Configuration config); @PreDestroy void shutdownExecutorService(); Workflow getWorkflowById(String workflowId, boolean includeTasks); List<Workflow> getWorkflowsByCorrelationId(String workflowName, String correlationId, boolean includeTasks); List<Workflow> getWorkflowsByName(String workflowName, Long startTime, Long endTime); List<Workflow> getPendingWorkflowsByName(String workflowName, int version); List<String> getRunningWorkflowIds(String workflowName, int version); long getPendingWorkflowCount(String workflowName); String createWorkflow(Workflow workflow); String updateWorkflow(Workflow workflow); void removeFromPendingWorkflow(String workflowType, String workflowId); void removeWorkflow(String workflowId, boolean archiveWorkflow); void removeWorkflowWithExpiry(String workflowId, boolean archiveWorkflow, int ttlSeconds); void resetWorkflow(String workflowId); List<Task> createTasks(List<Task> tasks); List<Task> getTasksForWorkflow(String workflowId); Task getTaskById(String taskId); List<Task> getTasksByName(String taskName, String startKey, int count); List<Task> getPendingTasksForTaskType(String taskType); long getInProgressTaskCount(String taskDefName); void updateTask(Task task); void updateTasks(List<Task> tasks); void removeTask(String taskId); List<PollData> getTaskPollData(String taskName); List<PollData> getAllPollData(); PollData getTaskPollDataByDomain(String taskName, String domain); void updateTaskLastPoll(String taskName, String domain, String workerId); boolean addEventExecution(EventExecution eventExecution); void updateEventExecution(EventExecution eventExecution); void removeEventExecution(EventExecution eventExecution); boolean exceedsInProgressLimit(Task task); boolean exceedsRateLimitPerFrequency(Task task, TaskDef taskDef); void addTaskExecLog(List<TaskExecLog> logs); void addMessage(String queue, Message message); SearchResult<String> searchWorkflows(String query, String freeText, int start, int count, List<String> sort); SearchResult<String> searchTasks(String query, String freeText, int start, int count, List<String> sort); List<TaskExecLog> getTaskExecutionLogs(String taskId);  }
@Test public void testGetWorkflowsByCorrelationId() { when(executionDAO.canSearchAcrossWorkflows()).thenReturn(true); when(executionDAO.getWorkflowsByCorrelationId(any(), any(), anyBoolean())).thenReturn(Collections.singletonList(new Workflow())); List<Workflow> workflows = executionDAOFacade.getWorkflowsByCorrelationId("workflowName", "correlationId", true); assertNotNull(workflows); assertEquals(1, workflows.size()); verify(indexDAO, never()).searchWorkflows(anyString(), anyString(), anyInt(), anyInt(), any()); when(executionDAO.canSearchAcrossWorkflows()).thenReturn(false); List<String> workflowIds = new ArrayList<>(); workflowIds.add("workflowId"); SearchResult<String> searchResult = new SearchResult<>(); searchResult.setResults(workflowIds); when(indexDAO.searchWorkflows(anyString(), anyString(), anyInt(), anyInt(), any())).thenReturn(searchResult); when(executionDAO.getWorkflow("workflowId", true)).thenReturn(new Workflow()); workflows = executionDAOFacade.getWorkflowsByCorrelationId("workflowName", "correlationId", true); assertNotNull(workflows); assertEquals(1, workflows.size()); }