@Test public void testIsFlowRunning() { JobStatusRetriever jobStatusRetriever = Mockito.mock(JobStatusRetriever.class); String flowName = "testName"; String flowGroup = "testGroup"; Mockito.when(jobStatusRetriever.getLatestExecutionIdsForFlow(flowName, flowGroup, 1)).thenReturn(null); FlowStatusGenerator flowStatusGenerator = FlowStatusGenerator.builder().jobStatusRetriever(jobStatusRetriever).build(); Assert.assertFalse(flowStatusGenerator.isFlowRunning(flowName, flowGroup)); Long flowExecutionId = 1234L; Mockito.when(jobStatusRetriever.getLatestExecutionIdsForFlow(flowName, flowGroup, 1)).thenReturn( Lists.newArrayList(flowExecutionId)); JobStatus jobStatus = JobStatus.builder().flowGroup(flowGroup).flowName(flowName).flowExecutionId(flowExecutionId) .jobName(JobStatusRetriever.NA_KEY).jobGroup(JobStatusRetriever.NA_KEY).eventName("COMPILED").build(); Iterator<JobStatus> jobStatusIterator = Lists.newArrayList(jobStatus).iterator(); Mockito.when(jobStatusRetriever.getJobStatusesForFlowExecution(flowName, flowGroup, flowExecutionId)).thenReturn(jobStatusIterator); Assert.assertTrue(flowStatusGenerator.isFlowRunning(flowName, flowGroup)); String job1 = "job1"; String job2 = "job2"; String job3 = "job3"; JobStatus jobStatus1 = JobStatus.builder().flowGroup(flowGroup).flowName(flowName).flowExecutionId(flowExecutionId) .jobName(job1).eventName("COMPLETE").build(); JobStatus jobStatus2 = JobStatus.builder().flowGroup(flowGroup).flowName(flowName).flowExecutionId(flowExecutionId) .jobName(job2).eventName("FAILED").build(); JobStatus jobStatus3 = JobStatus.builder().flowGroup(flowGroup).flowName(flowName).flowExecutionId(flowExecutionId) .jobName(job3).eventName("CANCELLED").build(); JobStatus flowStatus = JobStatus.builder().flowGroup(flowGroup).flowName(flowName).flowExecutionId(flowExecutionId) .jobName(JobStatusRetriever.NA_KEY).jobGroup(JobStatusRetriever.NA_KEY).eventName("CANCELLED").build(); jobStatusIterator = Lists.newArrayList(jobStatus1, jobStatus2, jobStatus3, flowStatus).iterator(); Mockito.when(jobStatusRetriever.getJobStatusesForFlowExecution(flowName, flowGroup, flowExecutionId)).thenReturn(jobStatusIterator); Assert.assertFalse(flowStatusGenerator.isFlowRunning(flowName, flowGroup)); flowStatus = JobStatus.builder().flowGroup(flowGroup).flowName(flowName).flowExecutionId(flowExecutionId) .jobName(JobStatusRetriever.NA_KEY).jobGroup(JobStatusRetriever.NA_KEY).eventName("RUNNING").build(); jobStatusIterator = Lists.newArrayList(jobStatus1, jobStatus2, jobStatus3, flowStatus).iterator(); Mockito.when(jobStatusRetriever.getJobStatusesForFlowExecution(flowName, flowGroup, flowExecutionId)).thenReturn(jobStatusIterator); Assert.assertTrue(flowStatusGenerator.isFlowRunning(flowName, flowGroup)); }
public boolean isFlowRunning(String flowName, String flowGroup) { List<FlowStatus> flowStatusList = getLatestFlowStatus(flowName, flowGroup, 1, null); if (flowStatusList == null || flowStatusList.isEmpty()) { return false; } else { FlowStatus flowStatus = flowStatusList.get(0); Iterator<JobStatus> jobStatusIterator = flowStatus.getJobStatusIterator(); while (jobStatusIterator.hasNext()) { JobStatus jobStatus = jobStatusIterator.next(); if (JobStatusRetriever.isFlowStatus(jobStatus)) { return isJobRunning(jobStatus); } } return false; } }
FlowStatusGenerator { public boolean isFlowRunning(String flowName, String flowGroup) { List<FlowStatus> flowStatusList = getLatestFlowStatus(flowName, flowGroup, 1, null); if (flowStatusList == null || flowStatusList.isEmpty()) { return false; } else { FlowStatus flowStatus = flowStatusList.get(0); Iterator<JobStatus> jobStatusIterator = flowStatus.getJobStatusIterator(); while (jobStatusIterator.hasNext()) { JobStatus jobStatus = jobStatusIterator.next(); if (JobStatusRetriever.isFlowStatus(jobStatus)) { return isJobRunning(jobStatus); } } return false; } } }
FlowStatusGenerator { public boolean isFlowRunning(String flowName, String flowGroup) { List<FlowStatus> flowStatusList = getLatestFlowStatus(flowName, flowGroup, 1, null); if (flowStatusList == null || flowStatusList.isEmpty()) { return false; } else { FlowStatus flowStatus = flowStatusList.get(0); Iterator<JobStatus> jobStatusIterator = flowStatus.getJobStatusIterator(); while (jobStatusIterator.hasNext()) { JobStatus jobStatus = jobStatusIterator.next(); if (JobStatusRetriever.isFlowStatus(jobStatus)) { return isJobRunning(jobStatus); } } return false; } }  }
FlowStatusGenerator { public boolean isFlowRunning(String flowName, String flowGroup) { List<FlowStatus> flowStatusList = getLatestFlowStatus(flowName, flowGroup, 1, null); if (flowStatusList == null || flowStatusList.isEmpty()) { return false; } else { FlowStatus flowStatus = flowStatusList.get(0); Iterator<JobStatus> jobStatusIterator = flowStatus.getJobStatusIterator(); while (jobStatusIterator.hasNext()) { JobStatus jobStatus = jobStatusIterator.next(); if (JobStatusRetriever.isFlowStatus(jobStatus)) { return isJobRunning(jobStatus); } } return false; } }  List<FlowStatus> getLatestFlowStatus(String flowName, String flowGroup, int count, String tag); List<FlowStatus> getLatestFlowStatus(String flowName, String flowGroup, int count, String tag, String executionStatus); FlowStatus getFlowStatus(String flowName, String flowGroup, long flowExecutionId, String tag); boolean isFlowRunning(String flowName, String flowGroup); void killFlow(String flowGroup, String flowName, Long flowExecutionId); }
FlowStatusGenerator { public boolean isFlowRunning(String flowName, String flowGroup) { List<FlowStatus> flowStatusList = getLatestFlowStatus(flowName, flowGroup, 1, null); if (flowStatusList == null || flowStatusList.isEmpty()) { return false; } else { FlowStatus flowStatus = flowStatusList.get(0); Iterator<JobStatus> jobStatusIterator = flowStatus.getJobStatusIterator(); while (jobStatusIterator.hasNext()) { JobStatus jobStatus = jobStatusIterator.next(); if (JobStatusRetriever.isFlowStatus(jobStatus)) { return isJobRunning(jobStatus); } } return false; } }  List<FlowStatus> getLatestFlowStatus(String flowName, String flowGroup, int count, String tag); List<FlowStatus> getLatestFlowStatus(String flowName, String flowGroup, int count, String tag, String executionStatus); FlowStatus getFlowStatus(String flowName, String flowGroup, long flowExecutionId, String tag); boolean isFlowRunning(String flowName, String flowGroup); void killFlow(String flowGroup, String flowName, Long flowExecutionId); static final List<String> FINISHED_STATUSES; static final int MAX_LOOKBACK; }