YarnRestJobStatusProvider implements JobStatusProvider { @Override public void getJobStatuses(Collection<Job> jobs) throws IOException, InterruptedException { if (jobs == null || jobs.isEmpty()) { return; } final Map<String, Job> qualifiedJobToInfo = new HashMap<>(); for (Job job : jobs) { qualifiedJobToInfo.put(YarnApplicationInfo.getQualifiedJobName(new JobInstance(job.getJobName(), job.getJobId())), job); } try { byte[] response = httpGet(apiEndpoint); YarnApplicationInfo yarnApplicationInfo = OBJECT_MAPPER.readValue(response, YarnApplicationInfo.class); for (YarnApplication app: yarnApplicationInfo.getYarnApplications()) { Job job = qualifiedJobToInfo.get(app.getName()); JobStatus samzaStatus = yarnStateToSamzaStatus(YarnApplicationState.valueOf(app.getState().toUpperCase())); if (job != null && (job.getStatusDetail() == null || samzaStatus != JobStatus.STOPPED)) { job.setStatusDetail(app.getState()); job.setStatus(samzaStatus); } } } catch (IOException e) { throw new RuntimeException("Failed to retrieve node info.", e); } } YarnRestJobStatusProvider(JobsResourceConfig config); @Override void getJobStatuses(Collection<Job> jobs); @Override Job getJobStatus(JobInstance jobInstance);  }
@Test public void testGetJobStatuses() throws IOException, InterruptedException { doReturn(APPS_RESPONSE.getBytes()).when(provider).httpGet(anyString()); List<Job> jobs = Lists.newArrayList( new Job("job1", "1"), new Job("job2", "1"), new Job("job3", "1"), new Job("job4", "1")); provider.getJobStatuses(jobs); Collections.sort(jobs, (o1, o2) -> o1.getJobName().compareTo(o2.getJobName())); assertEquals(4, jobs.size()); verifyJobStatus(jobs.get(0), "job1", JobStatus.STARTED, "RUNNING"); verifyJobStatus(jobs.get(1), "job2", JobStatus.STOPPED, "KILLED"); verifyJobStatus(jobs.get(2), "job3", JobStatus.STARTED, "RUNNING"); verifyJobStatus(jobs.get(3), "job4", JobStatus.UNKNOWN, null); }