@Override
  public void getJobStatuses(Collection<Job> jobs)
      throws IOException, InterruptedException {
    if (jobs == null || jobs.isEmpty()) {
      return;
    }

    // We will identify the YARN application states by their qualified names, so build a map
    // to translate back from that name to the JobInfo we wish to populate.
    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);

      // There can be multiple Yarn apps for each qualified job name, so we iterate the former and match with latter.
      for (YarnApplication app: yarnApplicationInfo.getYarnApplications()) {
        Job job = qualifiedJobToInfo.get(app.getName());
        JobStatus samzaStatus = yarnStateToSamzaStatus(YarnApplicationState.valueOf(app.getState().toUpperCase()));

        // If job is null, it wasn't requested.  The default statusDetail is null so always update in that case.
        // Only update the job status if the current status is not STOPPED because there could be many
        // application attempts for the job, and we're interested in the RUNNING one if it exists.
        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);
    }
  }
An implementation of the {@link JobStatusProvider} that retrieves
the job status from the YARN REST api.
samza-rest/src/main/java/org/apache/samza/rest/proxy/job/YarnRestJobStatusProvider.java