@Test public void testGetJobStatusesForFlowExecution() throws IOException { super.testGetJobStatusesForFlowExecution(); }
@Override public Iterator<JobStatus> getJobStatusesForFlowExecution(String flowName, String flowGroup, long flowExecutionId) { Preconditions.checkArgument(flowName != null, "FlowName cannot be null"); Preconditions.checkArgument(flowGroup != null, "FlowGroup cannot be null"); Predicate<String> flowExecutionIdPredicate = input -> input.startsWith(String.valueOf(flowExecutionId) + "."); String storeName = KafkaJobStatusMonitor.jobStatusStoreName(flowGroup, flowName); try { List<JobStatus> jobStatuses = new ArrayList<>(); List<String> tableNames = this.stateStore.getTableNames(storeName, flowExecutionIdPredicate); for (String tableName: tableNames) { List<State> jobStates = this.stateStore.getAll(storeName, tableName); if (jobStates.isEmpty()) { return Iterators.emptyIterator(); } jobStatuses.add(getJobStatus(jobStates.get(0))); } return jobStatuses.iterator(); } catch (IOException e) { log.error("IOException encountered when retrieving job statuses for flow: {},{},{}", flowGroup, flowName, flowExecutionId, e); return Iterators.emptyIterator(); } }
FsJobStatusRetriever extends JobStatusRetriever { @Override public Iterator<JobStatus> getJobStatusesForFlowExecution(String flowName, String flowGroup, long flowExecutionId) { Preconditions.checkArgument(flowName != null, "FlowName cannot be null"); Preconditions.checkArgument(flowGroup != null, "FlowGroup cannot be null"); Predicate<String> flowExecutionIdPredicate = input -> input.startsWith(String.valueOf(flowExecutionId) + "."); String storeName = KafkaJobStatusMonitor.jobStatusStoreName(flowGroup, flowName); try { List<JobStatus> jobStatuses = new ArrayList<>(); List<String> tableNames = this.stateStore.getTableNames(storeName, flowExecutionIdPredicate); for (String tableName: tableNames) { List<State> jobStates = this.stateStore.getAll(storeName, tableName); if (jobStates.isEmpty()) { return Iterators.emptyIterator(); } jobStatuses.add(getJobStatus(jobStates.get(0))); } return jobStatuses.iterator(); } catch (IOException e) { log.error("IOException encountered when retrieving job statuses for flow: {},{},{}", flowGroup, flowName, flowExecutionId, e); return Iterators.emptyIterator(); } } }
FsJobStatusRetriever extends JobStatusRetriever { @Override public Iterator<JobStatus> getJobStatusesForFlowExecution(String flowName, String flowGroup, long flowExecutionId) { Preconditions.checkArgument(flowName != null, "FlowName cannot be null"); Preconditions.checkArgument(flowGroup != null, "FlowGroup cannot be null"); Predicate<String> flowExecutionIdPredicate = input -> input.startsWith(String.valueOf(flowExecutionId) + "."); String storeName = KafkaJobStatusMonitor.jobStatusStoreName(flowGroup, flowName); try { List<JobStatus> jobStatuses = new ArrayList<>(); List<String> tableNames = this.stateStore.getTableNames(storeName, flowExecutionIdPredicate); for (String tableName: tableNames) { List<State> jobStates = this.stateStore.getAll(storeName, tableName); if (jobStates.isEmpty()) { return Iterators.emptyIterator(); } jobStatuses.add(getJobStatus(jobStates.get(0))); } return jobStatuses.iterator(); } catch (IOException e) { log.error("IOException encountered when retrieving job statuses for flow: {},{},{}", flowGroup, flowName, flowExecutionId, e); return Iterators.emptyIterator(); } } FsJobStatusRetriever(Config config); }
FsJobStatusRetriever extends JobStatusRetriever { @Override public Iterator<JobStatus> getJobStatusesForFlowExecution(String flowName, String flowGroup, long flowExecutionId) { Preconditions.checkArgument(flowName != null, "FlowName cannot be null"); Preconditions.checkArgument(flowGroup != null, "FlowGroup cannot be null"); Predicate<String> flowExecutionIdPredicate = input -> input.startsWith(String.valueOf(flowExecutionId) + "."); String storeName = KafkaJobStatusMonitor.jobStatusStoreName(flowGroup, flowName); try { List<JobStatus> jobStatuses = new ArrayList<>(); List<String> tableNames = this.stateStore.getTableNames(storeName, flowExecutionIdPredicate); for (String tableName: tableNames) { List<State> jobStates = this.stateStore.getAll(storeName, tableName); if (jobStates.isEmpty()) { return Iterators.emptyIterator(); } jobStatuses.add(getJobStatus(jobStates.get(0))); } return jobStatuses.iterator(); } catch (IOException e) { log.error("IOException encountered when retrieving job statuses for flow: {},{},{}", flowGroup, flowName, flowExecutionId, e); return Iterators.emptyIterator(); } } FsJobStatusRetriever(Config config); @Override Iterator<JobStatus> getJobStatusesForFlowExecution(String flowName, String flowGroup, long flowExecutionId); @Override Iterator<JobStatus> getJobStatusesForFlowExecution(String flowName, String flowGroup, long flowExecutionId,
      String jobName, String jobGroup); @Override List<Long> getLatestExecutionIdsForFlow(String flowName, String flowGroup, int count); }
FsJobStatusRetriever extends JobStatusRetriever { @Override public Iterator<JobStatus> getJobStatusesForFlowExecution(String flowName, String flowGroup, long flowExecutionId) { Preconditions.checkArgument(flowName != null, "FlowName cannot be null"); Preconditions.checkArgument(flowGroup != null, "FlowGroup cannot be null"); Predicate<String> flowExecutionIdPredicate = input -> input.startsWith(String.valueOf(flowExecutionId) + "."); String storeName = KafkaJobStatusMonitor.jobStatusStoreName(flowGroup, flowName); try { List<JobStatus> jobStatuses = new ArrayList<>(); List<String> tableNames = this.stateStore.getTableNames(storeName, flowExecutionIdPredicate); for (String tableName: tableNames) { List<State> jobStates = this.stateStore.getAll(storeName, tableName); if (jobStates.isEmpty()) { return Iterators.emptyIterator(); } jobStatuses.add(getJobStatus(jobStates.get(0))); } return jobStatuses.iterator(); } catch (IOException e) { log.error("IOException encountered when retrieving job statuses for flow: {},{},{}", flowGroup, flowName, flowExecutionId, e); return Iterators.emptyIterator(); } } FsJobStatusRetriever(Config config); @Override Iterator<JobStatus> getJobStatusesForFlowExecution(String flowName, String flowGroup, long flowExecutionId); @Override Iterator<JobStatus> getJobStatusesForFlowExecution(String flowName, String flowGroup, long flowExecutionId,
      String jobName, String jobGroup); @Override List<Long> getLatestExecutionIdsForFlow(String flowName, String flowGroup, int count); static final String CONF_PREFIX; }