MySQLExecutionDAO extends MySQLBaseDAO implements ExecutionDAO, RateLimitingDAO, PollDataDAO { @Override public List<Workflow> getWorkflowsByCorrelationId(String workflowName, String correlationId, boolean includeTasks) { Preconditions.checkNotNull(correlationId, "correlationId cannot be null"); String GET_WORKFLOWS_BY_CORRELATION_ID = "SELECT w.json_data FROM workflow w left join workflow_def_to_workflow wd on w.workflow_id = wd.workflow_id WHERE w.correlation_id = ? and wd.workflow_def = ?"; return queryWithTransaction(GET_WORKFLOWS_BY_CORRELATION_ID, q -> q.addParameter(correlationId).addParameter(workflowName).executeAndFetch(Workflow.class)); } @Inject  MySQLExecutionDAO(ObjectMapper objectMapper, DataSource dataSource); @Override List<Task> getPendingTasksByWorkflow(String taskDefName, String workflowId); @Override List<Task> getTasks(String taskDefName, String startKey, int count); @Override List<Task> createTasks(List<Task> tasks); @Override void updateTask(Task task); @Override boolean exceedsRateLimitPerFrequency(Task task, TaskDef taskDef); @Override boolean exceedsInProgressLimit(Task task); @Override boolean removeTask(String taskId); @Override Task getTask(String taskId); @Override List<Task> getTasks(List<String> taskIds); @Override List<Task> getPendingTasksForTaskType(String taskName); @Override List<Task> getTasksForWorkflow(String workflowId); @Override String createWorkflow(Workflow workflow); @Override String updateWorkflow(Workflow workflow); @Override boolean removeWorkflow(String workflowId); @Override boolean removeWorkflowWithExpiry(String workflowId, int ttlSeconds); @Override void removeFromPendingWorkflow(String workflowType, String workflowId); @Override Workflow getWorkflow(String workflowId); @Override Workflow getWorkflow(String workflowId, boolean includeTasks); @Override List<String> getRunningWorkflowIds(String workflowName, int version); @Override List<Workflow> getPendingWorkflowsByType(String workflowName, int version); @Override long getPendingWorkflowCount(String workflowName); @Override long getInProgressTaskCount(String taskDefName); @Override List<Workflow> getWorkflowsByType(String workflowName, Long startTime, Long endTime); @Override List<Workflow> getWorkflowsByCorrelationId(String workflowName, String correlationId, boolean includeTasks); @Override boolean canSearchAcrossWorkflows(); @Override boolean addEventExecution(EventExecution eventExecution); @Override void removeEventExecution(EventExecution eventExecution); @Override void updateEventExecution(EventExecution eventExecution); List<EventExecution> getEventExecutions(String eventHandlerName, String eventName, String messageId,
                                                   int max); @Override void updateLastPollData(String taskDefName, String domain, String workerId); @Override PollData getPollData(String taskDefName, String domain); @Override List<PollData> getPollData(String taskDefName); @Override List<PollData> getAllPollData();  }
@Test public void testPendingByCorrelationId() { WorkflowDef def = new WorkflowDef(); def.setName("pending_count_correlation_jtest"); Workflow workflow = createTestWorkflow(); workflow.setWorkflowDefinition(def); generateWorkflows(workflow, 10); List<Workflow> bycorrelationId = getExecutionDAO().getWorkflowsByCorrelationId("pending_count_correlation_jtest", "corr001", true); assertNotNull(bycorrelationId); assertEquals(10, bycorrelationId.size()); }