ExecutionFlowDao { List<ExecutableFlow> fetchFlowHistory(final int skip, final int num) throws ExecutorManagerException { try { return this.dbOperator.query(FetchExecutableFlows.FETCH_ALL_EXECUTABLE_FLOW_HISTORY, new FetchExecutableFlows(), skip, num); } catch (final SQLException e) { throw new ExecutorManagerException("Error fetching flow History", 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 fetchFlowHistory() throws Exception { final ExecutableFlow flow = createTestFlow(); this.executionFlowDao.uploadExecutableFlow(flow); final List<ExecutableFlow> flowList1 = this.executionFlowDao.fetchFlowHistory(0, 2); assertThat(flowList1.size()).isEqualTo(1); final List<ExecutableFlow> flowList2 = this.executionFlowDao .fetchFlowHistory(flow.getProjectId(), flow.getId(), 0, 2); assertThat(flowList2.size()).isEqualTo(1); final ExecutableFlow fetchFlow = this.executionFlowDao.fetchExecutableFlow(flow.getExecutionId()); assertTwoFlowSame(flowList1.get(0), flowList2.get(0)); assertTwoFlowSame(flowList1.get(0), fetchFlow); }
@Test public void fetchFlowHistoryWithStartTime() throws Exception { final ExecutableFlow flow1 = createExecution( TimeUtils.convertDateTimeToUTCMillis("2018-09-01 10:00:00"), Status.PREPARING); final ExecutableFlow flow2 = createExecution( TimeUtils.convertDateTimeToUTCMillis("2018-09-01 09:00:00"), Status.PREPARING); final ExecutableFlow flow3 = createExecution( TimeUtils.convertDateTimeToUTCMillis("2018-09-01 09:00:00"), Status.PREPARING); final ExecutableFlow flow4 = createExecution( TimeUtils.convertDateTimeToUTCMillis("2018-09-01 08:00:00"), Status.PREPARING); final List<ExecutableFlow> expectedFlows = new ArrayList<>(); expectedFlows.add(flow1); expectedFlows.add(flow2); expectedFlows.add(flow3); final List<ExecutableFlow> flowList = this.executionFlowDao.fetchFlowHistory( flow1.getProjectId(), flow1.getFlowId(), TimeUtils.convertDateTimeToUTCMillis("2018-09-01 09:00:00")); assertThat(flowList).hasSize(3); for (int i = 0; i < flowList.size(); i++) { assertTwoFlowSame(flowList.get(i), expectedFlows.get(i)); } }
@Test public void fetchFlowStatusWithFlowHistoryByStartTime() throws Exception { final ExecutableFlow flow1 = createExecution( TimeUtils.convertDateTimeToUTCMillis("2018-09-01 10:00:00"), Status.PREPARING); final ExecutableFlow flow2 = createExecution( TimeUtils.convertDateTimeToUTCMillis("2018-09-01 09:00:00"), Status.PREPARING); final ExecutableFlow flow3 = createExecution( TimeUtils.convertDateTimeToUTCMillis("2018-09-01 09:00:00"), Status.PREPARING); final ExecutableFlow flow4 = createExecution( TimeUtils.convertDateTimeToUTCMillis("2018-09-01 08:00:00"), Status.QUEUED); makeFlowStatusInconsistent(flow1.getExecutionId(), Status.FAILED_FINISHING); final List<ExecutableFlow> flowList = this.executionFlowDao.fetchFlowHistory( flow1.getProjectId(), flow1.getFlowId(), TimeUtils.convertDateTimeToUTCMillis("2018-09-01 08:00:00")); assertThat(flowList).hasSize(4); assertThat(flowList.get(0).getExecutionId()).isEqualTo(flow1.getExecutionId()); assertThat(flowList.get(0).getStatus()).isEqualTo(Status.FAILED_FINISHING); assertThat(flowList.get(3).getExecutionId()).isEqualTo(flow4.getExecutionId()); assertThat(flowList.get(3).getStatus()).isEqualTo(Status.QUEUED); }