ExecutionFlowDao { public List<Pair<ExecutionReference, ExecutableFlow>> fetchQueuedFlows() throws ExecutorManagerException { try { return this.dbOperator.query(FetchQueuedExecutableFlows.FETCH_QUEUED_EXECUTABLE_FLOW, new FetchQueuedExecutableFlows()); } catch (final SQLException e) { throw new ExecutorManagerException("Error fetching active flows", e); } } @Inject  ExecutionFlowDao(final DatabaseOperator dbOperator, final MysqlNamedLock mysqlNamedLock); void uploadExecutableFlow(final ExecutableFlow flow); List<ExecutableFlow> fetchAgedQueuedFlows(final Duration minAge); List<Pair<ExecutionReference, ExecutableFlow>> fetchQueuedFlows(); List<ExecutableFlow> fetchFlowHistory(final int projectId, final String flowId, final
  long startTime); ExecutableFlow fetchExecutableFlow(final int execId); void unsetExecutorIdForExecution(final int executionId); int selectAndUpdateExecution(final int executorId, final boolean isActive); int selectAndUpdateExecutionWithLocking(final int executorId, final boolean isActive);  }
@Test public void testFetchQueuedFlows() throws Exception { final ExecutableFlow flow1 = submitNewFlow("exectest1", "exec1", System.currentTimeMillis(), ExecutionOptions.DEFAULT_FLOW_PRIORITY); final ExecutableFlow flow2 = submitNewFlow("exectest1", "exec2", System.currentTimeMillis(), ExecutionOptions.DEFAULT_FLOW_PRIORITY); final List<Pair<ExecutionReference, ExecutableFlow>> fetchedQueuedFlows = this.executionFlowDao.fetchQueuedFlows(); assertThat(fetchedQueuedFlows.size()).isEqualTo(2); final Pair<ExecutionReference, ExecutableFlow> fetchedFlow1 = fetchedQueuedFlows.get(0); final Pair<ExecutionReference, ExecutableFlow> fetchedFlow2 = fetchedQueuedFlows.get(1); assertTwoFlowSame(flow1, fetchedFlow1.getSecond()); assertTwoFlowSame(flow2, fetchedFlow2.getSecond()); }