@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);
}