Java 类org.apache.hadoop.yarn.api.records.ApplicationReport 实例源码

/**
 * Scan all clusters to recover the soft state.
 */
@VisibleForTesting
void scanAll() throws IOException, YarnException {
  ConcurrentHashMap<UUID, InstanceInfo> newInstances = new ConcurrentHashMap<>();
  for (ClusterInfo cluster : clusters.values()) {
    List<ApplicationReport> reports = cluster.client()
        .getApplications(Collections.singleton(ATHENAX_APPLICATION_TYPE));
    for (ApplicationReport report : reports) {
      InstanceInfo instance = Utils.extractInstanceInfo(cluster.name(), report);
      if (instance == null) {
        LOG.warn("Failed to retrieve instance info for {}:{}", cluster.name(), report.getApplicationId());
      } else {
        newInstances.put(instance.metadata().uuid(), instance);
      }
    }
  }
  LOG.info("Inspected {} active instances", newInstances.size());
  instances.set(newInstances);
  listener.onUpdatedInstances(newInstances);
}

@Override
public YarnClusterClient deploy() {
  ApplicationSubmissionContext context = Records.newRecord(ApplicationSubmissionContext.class);
  context.setApplicationId(job.yarnAppId());
  ApplicationReport report;
  try {
    report = startAppMaster(context);

    Configuration conf = getFlinkConfiguration();
    conf.setString(JobManagerOptions.ADDRESS.key(), report.getHost());
    conf.setInteger(JobManagerOptions.PORT.key(), report.getRpcPort());

    return createYarnClusterClient(this, yarnClient, report, conf, false);
  } catch (Exception e) {
    throw new RuntimeException(e);
  }
}

@Override
public void run() {
  if (startTime == 0) {
    startTime = System.currentTimeMillis();
  }

  try {
    ApplicationReport report = poll();
    if (report == null) {
      YARN_POLL_EXECUTOR.schedule(this, RETRY_DELAY_MS, TimeUnit.MILLISECONDS);
    } else {
      result.complete(report);
    }
  } catch (YarnException | IOException e) {
    result.completeExceptionally(e);
  }
}

static InstanceInfo extractInstanceInfo(String clusterName, ApplicationReport report) {
  InstanceMetadata md = getMetadata(report.getApplicationTags());
  if (md == null) {
    return null;
  }

  ApplicationResourceUsageReport usage = report.getApplicationResourceUsageReport();
  InstanceStatus stat = new InstanceStatus()
      .allocatedVCores((long) usage.getUsedResources().getVirtualCores())
      .allocatedMB((long) usage.getUsedResources().getMemory())
      .clusterId(clusterName)
      .applicationId(report.getApplicationId().toString())
      .startedTime(report.getStartTime())
      .runningContainers((long) usage.getNumUsedContainers())
      .trackingUrl(report.getTrackingUrl())
      .state(InstanceStatus.StateEnum.fromValue(report.getYarnApplicationState().toString()));
  return new InstanceInfo(clusterName, report.getApplicationId(), md, stat);
}

@Override
public JobStatus submitJob(JobID jobId, String jobSubmitDir, Credentials ts)
        throws IOException, InterruptedException {

    addHistoryToken(ts);

    // Construct necessary information to start the MR AM
    ApplicationSubmissionContext appContext = createApplicationSubmissionContext(conf, jobSubmitDir, ts);

    // Submit to ResourceManager
    try {
        ApplicationId applicationId = resMgrDelegate.submitApplication(appContext);

        ApplicationReport appMaster = resMgrDelegate.getApplicationReport(applicationId);
        String diagnostics = (appMaster == null ? "application report is null" : appMaster.getDiagnostics());
        if (appMaster == null || appMaster.getYarnApplicationState() == YarnApplicationState.FAILED
                || appMaster.getYarnApplicationState() == YarnApplicationState.KILLED) {
            throw new IOException("Failed to run job : " + diagnostics);
        }
        return clientCache.getClient(jobId).getJobStatus(jobId);
    } catch (YarnException e) {
        throw new IOException(e);
    }
}

public void resyncWithRM() throws Exception {
  List<ApplicationReport> reports = null;

  try {
    // Only Get LAUNCHER ApplicationReport
    reports = yarnClient.getApplications(new HashSet<>(
        Collections.singletonList(GlobalConstants.LAUNCHER_APPLICATION_TYPE)));
  } catch (Exception e) {
    LOGGER.logWarning(e,
        "Exception occurred during GetApplications. It should be transient. " +
            "Will retry next time after %ss", conf.getServiceRMResyncIntervalSec());
  }

  if (reports != null) {
    // ApplicationId -> ApplicationReport
    HashMap<String, ApplicationReport> liveApplicationReports = new HashMap<>();
    for (ApplicationReport report : reports) {
      liveApplicationReports.put(report.getApplicationId().toString(), report);
    }

    service.onLiveApplicationsUpdated(liveApplicationReports);
  }

  service.queueResyncWithRM(conf.getServiceRMResyncIntervalSec());
}

@Test(timeout = 10000)
public void testGetApplicationAttempt() throws YarnException, IOException {
  Configuration conf = new Configuration();
  final YarnClient client = new MockYarnClient();
  client.init(conf);
  client.start();

  List<ApplicationReport> expectedReports = ((MockYarnClient) client)
      .getReports();

  ApplicationId applicationId = ApplicationId.newInstance(1234, 5);
  ApplicationAttemptId appAttemptId = ApplicationAttemptId.newInstance(
      applicationId, 1);
  ApplicationAttemptReport report = client
      .getApplicationAttemptReport(appAttemptId);
  Assert.assertNotNull(report);
  Assert.assertEquals(report.getApplicationAttemptId().toString(),
      expectedReports.get(0).getCurrentApplicationAttemptId().toString());
  client.stop();
}

boolean awaitApplication(ApplicationId appId) throws Exception {
  Set<YarnApplicationState> terminated = Sets.newHashSet(
      YarnApplicationState.FAILED,
      YarnApplicationState.FINISHED,
      YarnApplicationState.KILLED);
  while (true) {
    ApplicationReport report = yarnClient.getApplicationReport(appId);
    YarnApplicationState state = report.getYarnApplicationState();
    if (state.equals(YarnApplicationState.RUNNING)) {
      ClusterSpec clusterSpec = Client.getClusterSpec(yarnClient, appId);
      if (isClusterSpecSatisfied(clusterSpec)) {
        System.out.println("ClusterSpec: " + Utils.toJsonString(clusterSpec.getCluster()));
        return true;
      }
    } else if (terminated.contains(state)) {
      return false;
    } else {
      Thread.sleep(1000);
    }
  }
}

@Test(timeout=20000)
public void testJobSubmissionFailure() throws Exception {
  when(resourceMgrDelegate.submitApplication(any(ApplicationSubmissionContext.class))).
  thenReturn(appId);
  ApplicationReport report = mock(ApplicationReport.class);
  when(report.getApplicationId()).thenReturn(appId);
  when(report.getDiagnostics()).thenReturn(failString);
  when(report.getYarnApplicationState()).thenReturn(YarnApplicationState.FAILED);
  when(resourceMgrDelegate.getApplicationReport(appId)).thenReturn(report);
  Credentials credentials = new Credentials();
  File jobxml = new File(testWorkDir, "job.xml");
  OutputStream out = new FileOutputStream(jobxml);
  conf.writeXml(out);
  out.close();
  try {
    yarnRunner.submitJob(jobId, testWorkDir.getAbsolutePath().toString(), credentials);
  } catch(IOException io) {
    LOG.info("Logging exception:", io);
    assertTrue(io.getLocalizedMessage().contains(failString));
  }
}

@Test
public void testApplicationReport() {
  long timestamp = System.currentTimeMillis();
  ApplicationReport appReport1 =
      createApplicationReport(1, 1, timestamp);
  ApplicationReport appReport2 =
      createApplicationReport(1, 1, timestamp);
  ApplicationReport appReport3 =
      createApplicationReport(1, 1, timestamp);
  Assert.assertEquals(appReport1, appReport2);
  Assert.assertEquals(appReport2, appReport3);
  appReport1.setApplicationId(null);
  Assert.assertNull(appReport1.getApplicationId());
  Assert.assertNotSame(appReport1, appReport2);
  appReport2.setCurrentApplicationAttemptId(null);
  Assert.assertNull(appReport2.getCurrentApplicationAttemptId());
  Assert.assertNotSame(appReport2, appReport3);
  Assert.assertNull(appReport1.getAMRMToken());
}

@Override
public ApplicationReport getApplicationReport(ApplicationId applicationId)
    throws IOException {
  List<ApplicationReport> reports = null;
  try {
    reports = getApplications(applicationId.getId());
  } catch (Throwable e) {
    LOG.info("Couldn't get application report for " + applicationId
        + ", might be completed already.");
  }
  if (reports == null || reports.isEmpty()) {
    return ApplicationReport.newInstance(applicationId, null, "", "default",
        "", "", 0, null, YarnApplicationState.FINISHED, "", "", 0, 0,
        FinalApplicationStatus.SUCCEEDED, null, "", 100, null, null);
  }
  return reports.get(0);
}

public AppInfo(ApplicationReport app) {
  appId = app.getApplicationId().toString();
  if (app.getCurrentApplicationAttemptId() != null) {
    currentAppAttemptId = app.getCurrentApplicationAttemptId().toString();
  }
  user = app.getUser();
  queue = app.getQueue();
  name = app.getName();
  type = app.getApplicationType();
  host = app.getHost();
  rpcPort = app.getRpcPort();
  appState = app.getYarnApplicationState();
  diagnosticsInfo = app.getDiagnostics();
  trackingUrl = app.getTrackingUrl();
  originalTrackingUrl = app.getOriginalTrackingUrl();
  submittedTime = app.getStartTime();
  startedTime = app.getStartTime();
  finishedTime = app.getFinishTime();
  elapsedTime = Times.elapsed(startedTime, finishedTime);
  finalAppStatus = app.getFinalApplicationStatus();
  progress = app.getProgress() * 100; // in percent
  if (app.getApplicationTags() != null && !app.getApplicationTags().isEmpty()) {
    this.applicationTags = CSV_JOINER.join(app.getApplicationTags());
  }
}

@Test
public void tesAllJobs() throws Exception {
  final ApplicationClientProtocol applicationsManager = Mockito.mock(ApplicationClientProtocol.class);
  GetApplicationsResponse allApplicationsResponse = Records
      .newRecord(GetApplicationsResponse.class);
  List<ApplicationReport> applications = new ArrayList<ApplicationReport>();
  applications.add(getApplicationReport(YarnApplicationState.FINISHED,
      FinalApplicationStatus.FAILED));
  applications.add(getApplicationReport(YarnApplicationState.FINISHED,
      FinalApplicationStatus.SUCCEEDED));
  applications.add(getApplicationReport(YarnApplicationState.FINISHED,
      FinalApplicationStatus.KILLED));
  applications.add(getApplicationReport(YarnApplicationState.FAILED,
      FinalApplicationStatus.FAILED));
  allApplicationsResponse.setApplicationList(applications);
  Mockito.when(
      applicationsManager.getApplications(Mockito
          .any(GetApplicationsRequest.class))).thenReturn(
      allApplicationsResponse);
  ResourceMgrDelegate resourceMgrDelegate = new ResourceMgrDelegate(
    new YarnConfiguration()) {
    @Override
    protected void serviceStart() throws Exception {
      Assert.assertTrue(this.client instanceof YarnClientImpl);
      ((YarnClientImpl) this.client).setRMClient(applicationsManager);
    }
  };
  JobStatus[] allJobs = resourceMgrDelegate.getAllJobs();

  Assert.assertEquals(State.FAILED, allJobs[0].getState());
  Assert.assertEquals(State.SUCCEEDED, allJobs[1].getState());
  Assert.assertEquals(State.KILLED, allJobs[2].getState());
  Assert.assertEquals(State.FAILED, allJobs[3].getState());
}

@Test (timeout = 60000)
public void testInvalidatedAMHostPortOnAMRestart() throws Exception {
  MockRM rm1 = new MockRM(conf);
  rm1.start();
  MockNM nm1 =
      new MockNM("127.0.0.1:1234", 15120, rm1.getResourceTrackerService());
  nm1.registerNode();

  // a failed app
  RMApp app2 = rm1.submitApp(200);
  MockAM am2 = MockRM.launchAndRegisterAM(app2, rm1, nm1);
  nm1
    .nodeHeartbeat(am2.getApplicationAttemptId(), 1, ContainerState.COMPLETE);
  am2.waitForState(RMAppAttemptState.FAILED);
  rm1.waitForState(app2.getApplicationId(), RMAppState.ACCEPTED);

  // before new attempt is launched, the app report returns the invalid AM
  // host and port.
  GetApplicationReportRequest request1 =
      GetApplicationReportRequest.newInstance(app2.getApplicationId());
  ApplicationReport report1 =
      rm1.getClientRMService().getApplicationReport(request1)
        .getApplicationReport();
  Assert.assertEquals("N/A", report1.getHost());
  Assert.assertEquals(-1, report1.getRpcPort());
}

private void verifyEnemyAppReport(ApplicationReport appReport) {
  Assert.assertEquals("Enemy should not see app host!",
      UNAVAILABLE, appReport.getHost());
  Assert.assertEquals("Enemy should not see app rpc port!",
      -1, appReport.getRpcPort());
  Assert.assertEquals("Enemy should not see app client token!",
      null, appReport.getClientToAMToken());
  Assert.assertEquals("Enemy should not see app diagnostics!",
      UNAVAILABLE, appReport.getDiagnostics());
  Assert.assertEquals("Enemy should not see app tracking url!",
      UNAVAILABLE, appReport.getTrackingUrl());
  Assert.assertEquals("Enemy should not see app original tracking url!",
      UNAVAILABLE, appReport.getOriginalTrackingUrl());
  ApplicationResourceUsageReport usageReport =
      appReport.getApplicationResourceUsageReport();
  Assert.assertEquals("Enemy should not see app used containers",
      -1, usageReport.getNumUsedContainers());
  Assert.assertEquals("Enemy should not see app reserved containers",
      -1, usageReport.getNumReservedContainers());
  Assert.assertEquals("Enemy should not see app used resources",
      -1, usageReport.getUsedResources().getMemory());
  Assert.assertEquals("Enemy should not see app reserved resources",
      -1, usageReport.getReservedResources().getMemory());
  Assert.assertEquals("Enemy should not see app needed resources",
      -1, usageReport.getNeededResources().getMemory());
}

private NotRunningJob getNotRunningJob(ApplicationReport applicationReport,
    JobState state) {
  synchronized (notRunningJobs) {
    HashMap<String, NotRunningJob> map = notRunningJobs.get(state);
    if (map == null) {
      map = new HashMap<String, NotRunningJob>();
      notRunningJobs.put(state, map);
    }
    String user =
        (applicationReport == null) ?
            UNKNOWN_USER : applicationReport.getUser();
    NotRunningJob notRunningJob = map.get(user);
    if (notRunningJob == null) {
      notRunningJob = new NotRunningJob(applicationReport, state);
      map.put(user, notRunningJob);
    }
    return notRunningJob;
  }
}

@Override
public Map<ContainerId, ContainerReport> getContainers(
    ApplicationAttemptId appAttemptId) throws IOException {
  ApplicationReport app =
      getApplication(appAttemptId.getApplicationId());
  Map<ContainerId, ContainerHistoryData> histData =
      historyStore.getContainers(appAttemptId);
  HashMap<ContainerId, ContainerReport> containersReport =
      new HashMap<ContainerId, ContainerReport>();
  for (Entry<ContainerId, ContainerHistoryData> entry : histData.entrySet()) {
    containersReport.put(entry.getKey(),
      convertToContainerReport(entry.getValue(),
          app == null ? null : app.getUser()));
  }
  return containersReport;
}

@Test
public void testApplicationReport() throws IOException, YarnException {
  ApplicationId appId = null;
  appId = ApplicationId.newInstance(0, 1);
  GetApplicationReportRequest request =
      GetApplicationReportRequest.newInstance(appId);
  GetApplicationReportResponse response =
      clientService.getApplicationReport(request);
  ApplicationReport appReport = response.getApplicationReport();
  Assert.assertNotNull(appReport);
  Assert.assertEquals(123, appReport.getApplicationResourceUsageReport()
      .getMemorySeconds());
  Assert.assertEquals(345, appReport.getApplicationResourceUsageReport()
      .getVcoreSeconds());
  Assert.assertEquals(345, appReport.getApplicationResourceUsageReport()
      .getGcoreSeconds());
  Assert.assertEquals("application_0_0001", appReport.getApplicationId()
    .toString());
  Assert.assertEquals("test app type",
      appReport.getApplicationType().toString());
  Assert.assertEquals("test queue", appReport.getQueue().toString());
}

private void waitTillAccepted(YarnClient rmClient, ApplicationId appId)
  throws Exception {
  try {
    long start = System.currentTimeMillis();
    ApplicationReport report = rmClient.getApplicationReport(appId);
    while (YarnApplicationState.ACCEPTED != report.getYarnApplicationState()) {
      if (System.currentTimeMillis() - start > 20 * 1000) {
        throw new Exception("App '" + appId + 
          "' time out, failed to reach ACCEPTED state");
      }
      Thread.sleep(200);
      report = rmClient.getApplicationReport(appId);
    }
  } catch (Exception ex) {
    throw new Exception(ex);
  }
}

private int verifyApplicationState(ApplicationId appId) throws IOException,
    YarnException {
  YarnClient yarnClient = createYarnClient();

  try {
    ApplicationReport appReport = yarnClient.getApplicationReport(appId);
    switch (appReport.getYarnApplicationState()) {
    case NEW:
    case NEW_SAVING:
    case SUBMITTED:
      return -1;
    case ACCEPTED:
    case RUNNING:
    case FAILED:
    case FINISHED:
    case KILLED:
    default:
      break;

    }
  } finally {
    yarnClient.close();
  }
  return 0;
}

private ApplicationReport poll() throws IOException, YarnException {
  ApplicationReport report;
  report = yarnClient.getApplicationReport(appId);
  YarnApplicationState appState = report.getYarnApplicationState();
  LOG.debug("Application State: {}", appState);

  switch (appState) {
    case FAILED:
    case FINISHED:
      //TODO: the finished state may be valid in flip-6
    case KILLED:
      throw new IOException("The YARN application unexpectedly switched to state "
          + appState + " during deployment. \n"
          + "Diagnostics from YARN: " + report.getDiagnostics() + "\n"
          + "If log aggregation is enabled on your cluster, use this command to further investigate the issue:\n"
          + "yarn logs -applicationId " + appId);
      //break ..
    case RUNNING:
      LOG.info("YARN application has been deployed successfully.");
      break;
    default:
      if (appState != lastAppState) {
        LOG.info("Deploying cluster, current state " + appState);
      }
      lastAppState = appState;
      if (System.currentTimeMillis() - startTime > DEPLOY_TIMEOUT_MS) {
        throw new RuntimeException(String.format("Deployment took more than %d seconds. "
            + "Please check if the requested resources are available in the YARN cluster", DEPLOY_TIMEOUT_MS));
      }
      return null;
  }
  return report;
}

public static JobStatus[] fromYarnApps(List<ApplicationReport> applications,
    Configuration conf) {
  List<JobStatus> jobStatuses = new ArrayList<JobStatus>();
  for (ApplicationReport application : applications) {
    // each applicationReport has its own jobFile
    org.apache.hadoop.mapreduce.JobID jobId =
        TypeConverter.fromYarn(application.getApplicationId());
    jobStatuses.add(TypeConverter.fromYarn(application,
        MRApps.getJobFile(conf, application.getUser(), jobId)));
  }
  return jobStatuses.toArray(new JobStatus[jobStatuses.size()]);
}

@Override
public List<ApplicationReport> getApplications(
    Set<String> applicationTypes, EnumSet<YarnApplicationState> applicationStates)
    throws YarnException, IOException {
  when(mockAppResponse.getApplicationList()).thenReturn(
      getApplicationReports(reports, applicationTypes, applicationStates));
  return super.getApplications(applicationTypes, applicationStates);
}

public static String retrieveDiagnostics(YarnClient yarnClient, String applicationId) throws Exception {
  ApplicationReport report = yarnClient.getApplicationReport(ConverterUtils.toApplicationId(applicationId));
  String diagnostics = report.getDiagnostics();
  if (isDiagnosticsEmpty(diagnostics)) {
    throw new Exception("Retrieved Empty Diagnostics for " + applicationId);
  }
  return diagnostics;
}

public synchronized void updateApplicationStatus(String frameworkName, ApplicationReport applicationReport) throws Exception {
  FrameworkStatus frameworkStatus = getFrameworkStatus(frameworkName);
  String applicationId = applicationReport.getApplicationId().toString();
  String logPrefix = String.format(
      "[%s][%s]: UpdateFrameworkStatus: ", frameworkName, frameworkStatus.getApplicationId());

  assert applicationId.equals(frameworkStatus.getApplicationId());

  boolean frameworkStatusChanged = false;
  if (frameworkStatus.getApplicationProgress() == null ||
      Math.abs(frameworkStatus.getApplicationProgress() - applicationReport.getProgress()) >= 0.1) {
    LOGGER.logInfo(
        logPrefix + "Update ApplicationProgress from [%s] to [%s]",
        frameworkStatus.getApplicationProgress(), applicationReport.getProgress());
    frameworkStatus.setApplicationProgress(applicationReport.getProgress());
    frameworkStatusChanged = true;
  }

  // Only update ApplicationTrackingUrl at the first time, since after Application
  // completed in RM, it will be redirect to non-proxy url.
  if (frameworkStatus.getApplicationTrackingUrl() == null ||
      frameworkStatus.getApplicationTrackingUrl().trim().isEmpty()) {
    LOGGER.logInfo(
        logPrefix + "Update ApplicationTrackingUrl from [%s] to [%s]",
        frameworkStatus.getApplicationTrackingUrl(), applicationReport.getTrackingUrl());
    frameworkStatus.setApplicationTrackingUrl(applicationReport.getTrackingUrl());
    frameworkStatusChanged = true;
  }

  if (frameworkStatusChanged) {
    zkStore.setFrameworkStatus(frameworkName, frameworkStatus);
  }
}

public ApplicationReport createFakeAppReport() {
  ApplicationId appId = ApplicationId.newInstance(1000l, 1);
  ApplicationAttemptId attemptId =
      ApplicationAttemptId.newInstance(appId, 1);
  // create a fake application report
  ApplicationReport report =
      ApplicationReport.newInstance(appId, attemptId, "fakeUser",
          "fakeQueue", "fakeApplicationName", "localhost", 0, null,
          YarnApplicationState.FINISHED, "fake an application report", "",
          1000l, 1200l, FinalApplicationStatus.FAILED, null, "", 50f,
          "fakeApplicationType", null);
  return report;
}

@Override
public void setApplicationList(List<ApplicationReport> applications) {
  maybeInitBuilder();
  if (applications == null)
    builder.clearApplications();
  this.applicationList = applications;
}

@Override
public GetApplicationReportResponse getApplicationReport(
    GetApplicationReportRequest request) throws YarnException {
  resetStartFailoverFlag(true);

  // make sure failover has been triggered
  Assert.assertTrue(waittingForFailOver());

  // create a fake application report
  ApplicationReport report = createFakeAppReport();
  GetApplicationReportResponse response =
      GetApplicationReportResponse.newInstance(report);
  return response;
}

@Override
public void setApplications(List<ApplicationReport> applications) {
  if (applications == null) {
    builder.clearApplications();
  }
  this.applicationsList = applications;
}

private void initLocalApplicationsList() {
  if (this.applicationsList != null) {
    return;
  }
  QueueInfoProtoOrBuilder p = viaProto ? proto : builder;
  List<ApplicationReportProto> list = p.getApplicationsList();
  applicationsList = new ArrayList<ApplicationReport>();

  for (ApplicationReportProto a : list) {
    applicationsList.add(convertFromProtoFormat(a));
  }
}

private void addApplicationsToProto() {
  maybeInitBuilder();
  builder.clearApplications();
  if (applicationsList == null)
    return;
  Iterable<ApplicationReportProto> iterable = new Iterable<ApplicationReportProto>() {
    @Override
    public Iterator<ApplicationReportProto> iterator() {
      return new Iterator<ApplicationReportProto>() {

        Iterator<ApplicationReport> iter = applicationsList.iterator();

        @Override
        public boolean hasNext() {
          return iter.hasNext();
        }

        @Override
        public ApplicationReportProto next() {
          return convertToProtoFormat(iter.next());
        }

        @Override
        public void remove() {
          throw new UnsupportedOperationException();

        }
      };

    }
  };
  builder.addAllApplications(iterable);
}

private static GetApplicationReportResponse
    createApplicationReportWithFinishedApplication() {
  ApplicationReport report = mock(ApplicationReport.class);
  when(report.getYarnApplicationState()).thenReturn(
    YarnApplicationState.FINISHED);
  GetApplicationReportResponse response =
      mock(GetApplicationReportResponse.class);
  when(response.getApplicationReport()).thenReturn(report);
  return response;
}

/**
 * Get a report for the specified app.
 * @param appId the id of the application to get. 
 * @return the ApplicationReport for that app.
 * @throws YarnException on any error.
 * @throws IOException
 */
public ApplicationReport getApplicationReport(ApplicationId appId)
throws YarnException, IOException {
  GetApplicationReportRequest request = recordFactory
      .newRecordInstance(GetApplicationReportRequest.class);
  request.setApplicationId(appId);

  GetApplicationReportResponse response = applicationsManager
      .getApplicationReport(request);
  return response.getApplicationReport();
}

public static ApplicationReport newApplicationReport(
    ApplicationId applicationId, ApplicationAttemptId applicationAttemptId,
    String user, String queue, String name, String host, int rpcPort,
    Token clientToAMToken, YarnApplicationState state, String diagnostics,
    String url, long startTime, long finishTime,
    FinalApplicationStatus finalStatus,
    ApplicationResourceUsageReport appResources, String origTrackingUrl,
    float progress, String appType, Token amRmToken, Set<String> tags) {
  ApplicationReport report = recordFactory
      .newRecordInstance(ApplicationReport.class);
  report.setApplicationId(applicationId);
  report.setCurrentApplicationAttemptId(applicationAttemptId);
  report.setUser(user);
  report.setQueue(queue);
  report.setName(name);
  report.setHost(host);
  report.setRpcPort(rpcPort);
  report.setClientToAMToken(clientToAMToken);
  report.setYarnApplicationState(state);
  report.setDiagnostics(diagnostics);
  report.setTrackingUrl(url);
  report.setStartTime(startTime);
  report.setFinishTime(finishTime);
  report.setFinalApplicationStatus(finalStatus);
  report.setApplicationResourceUsageReport(appResources);
  report.setOriginalTrackingUrl(origTrackingUrl);
  report.setProgress(progress);
  report.setApplicationType(appType);
  report.setAMRMToken(amRmToken);
  report.setApplicationTags(tags);
  return report;
}

@Override
@Private
public Collection<ApplicationId> getActiveApplications() throws YarnException {
  try {
    List<ApplicationId> activeApps = new ArrayList<ApplicationId>();
    List<ApplicationReport> apps = client.getApplications(ACTIVE_STATES);
    for (ApplicationReport app: apps) {
      activeApps.add(app.getApplicationId());
    }
    return activeApps;
  } catch (IOException e) {
    throw new YarnException(e);
  }
}

@Test
public void testRunningApp() throws Exception {
  YarnClient client = createCheckerWithMockedClient();
  ApplicationId id = ApplicationId.newInstance(1, 1);

  // create a report and set the state to an active one
  ApplicationReport report = new ApplicationReportPBImpl();
  report.setYarnApplicationState(YarnApplicationState.ACCEPTED);
  doReturn(report).when(client).getApplicationReport(id);

  assertTrue(checker.isApplicationActive(id));
}

private ApplicationReport getUnknownApplicationReport() {
  ApplicationId unknownAppId = recordFactory
      .newRecordInstance(ApplicationId.class);
  ApplicationAttemptId unknownAttemptId = recordFactory
      .newRecordInstance(ApplicationAttemptId.class);

  // Setting AppState to NEW and finalStatus to UNDEFINED as they are never
  // used for a non running job
  return ApplicationReport.newInstance(unknownAppId, unknownAttemptId,
    "N/A", "N/A", "N/A", "N/A", 0, null, YarnApplicationState.NEW, "N/A",
    "N/A", 0, 0, FinalApplicationStatus.UNDEFINED, null, "N/A", 0.0f,
    YarnConfiguration.DEFAULT_APPLICATION_TYPE, null);
}

private List<ApplicationReport> getApplicationReports(
    List<ApplicationReport> applicationReports,
    Set<String> appTypes, EnumSet<YarnApplicationState> appStates,
    boolean allStates) {

  List<ApplicationReport> appReports = new ArrayList<ApplicationReport>();

  if (allStates) {
    for(YarnApplicationState state : YarnApplicationState.values()) {
      appStates.add(state);
    }
  }
  for (ApplicationReport appReport : applicationReports) {
    if (appTypes != null && !appTypes.isEmpty()) {
      if (!appTypes.contains(appReport.getApplicationType())) {
        continue;
      }
    }

    if (appStates != null && !appStates.isEmpty()) {
      if (!appStates.contains(appReport.getYarnApplicationState())) {
        continue;
      }
    }

    appReports.add(appReport);
  }
  return appReports;
}

public static ClientRMService mockClientRMService(RMContext rmContext) {
  ClientRMService clientRMService = mock(ClientRMService.class);
  List<ApplicationReport> appReports = new ArrayList<ApplicationReport>();
  for (RMApp app : rmContext.getRMApps().values()) {
    ApplicationReport appReport =
        ApplicationReport.newInstance(
            app.getApplicationId(), (ApplicationAttemptId) null,
            app.getUser(), app.getQueue(),
            app.getName(), (String) null, 0, (Token) null,
            app.createApplicationState(),
            app.getDiagnostics().toString(), (String) null,
            app.getStartTime(), app.getFinishTime(),
            app.getFinalApplicationStatus(),
            (ApplicationResourceUsageReport) null, app.getTrackingUrl(),
            app.getProgress(), app.getApplicationType(), (Token) null);
    appReports.add(appReport);
  }
  GetApplicationsResponse response = mock(GetApplicationsResponse.class);
  when(response.getApplicationList()).thenReturn(appReports);
  try {
    when(clientRMService.getApplications(any(GetApplicationsRequest.class)))
        .thenReturn(response);
  } catch (YarnException e) {
    Assert.fail("Exception is not expteced.");
  }
  return clientRMService;
}

private ApplicationReport verifyAppReportAfterRMRestart(RMApp app, MockRM rm)
    throws Exception {
  GetApplicationReportRequest reportRequest =
      GetApplicationReportRequest.newInstance(app.getApplicationId());
  GetApplicationReportResponse response =
      rm.getClientRMService().getApplicationReport(reportRequest);
  ApplicationReport report = response.getApplicationReport();
  Assert.assertEquals(app.getStartTime(), report.getStartTime());
  Assert.assertEquals(app.getFinishTime(), report.getFinishTime());
  Assert.assertEquals(app.createApplicationState(),
    report.getYarnApplicationState());
  Assert.assertTrue(1 == report.getProgress());
  return response.getApplicationReport();
}