import org.apache.hadoop.yarn.api.protocolrecords.GetClusterMetricsRequest; //导入依赖的package包/类
/**
 * Wait for all the NodeManagers to connect to the ResourceManager.
 *
 * @param timeout Time to wait (sleeps in 100 ms intervals) in milliseconds.
 * @return true if all NodeManagers connect to the (Active)
 * ResourceManager, false otherwise.
 * @throws YarnException
 * @throws InterruptedException
 */
public boolean waitForNodeManagersToConnect(long timeout)
    throws YarnException, InterruptedException {
  GetClusterMetricsRequest req = GetClusterMetricsRequest.newInstance();
  for (int i = 0; i < timeout / 100; i++) {
    ResourceManager rm = getResourceManager();
    if (rm == null) {
      throw new YarnException("Can not find the active RM.");
    }
    else if (nodeManagers.length == rm.getClientRMService()
          .getClusterMetrics(req).getClusterMetrics().getNumNodeManagers()) {
      return true;
    }
    Thread.sleep(100);
  }
  return false;
}
 

import org.apache.hadoop.yarn.api.protocolrecords.GetClusterMetricsRequest; //导入依赖的package包/类
@Override
public GetClusterMetricsResponse getClusterMetrics(
    GetClusterMetricsRequest request) throws YarnException {
  GetClusterMetricsResponse response = recordFactory
      .newRecordInstance(GetClusterMetricsResponse.class);
  YarnClusterMetrics ymetrics = recordFactory
      .newRecordInstance(YarnClusterMetrics.class);
  ymetrics.setNumNodeManagers(this.rmContext.getRMNodes().size());
  ClusterMetrics clusterMetrics = ClusterMetrics.getMetrics();
  ymetrics.setNumDecommissionedNodeManagers(clusterMetrics
    .getNumDecommisionedNMs());
  ymetrics.setNumActiveNodeManagers(clusterMetrics.getNumActiveNMs());
  ymetrics.setNumLostNodeManagers(clusterMetrics.getNumLostNMs());
  ymetrics.setNumUnhealthyNodeManagers(clusterMetrics.getUnhealthyNMs());
  ymetrics.setNumRebootedNodeManagers(clusterMetrics.getNumRebootedNMs());
  response.setClusterMetrics(ymetrics);
  return response;
}
 

import org.apache.hadoop.yarn.api.protocolrecords.GetClusterMetricsRequest; //导入依赖的package包/类
/**
 * Wait for all the NodeManagers to connect to the ResourceManager.
 *
 * @param timeout Time to wait (sleeps in 10 ms intervals) in milliseconds.
 * @return true if all NodeManagers connect to the (Active)
 * ResourceManager, false otherwise.
 * @throws YarnException if there is no active RM
 * @throws InterruptedException if any thread has interrupted
 * the current thread
 */
public boolean waitForNodeManagersToConnect(long timeout)
    throws YarnException, InterruptedException {
  GetClusterMetricsRequest req = GetClusterMetricsRequest.newInstance();
  for (int i = 0; i < timeout / 10; i++) {
    ResourceManager rm = getResourceManager();
    if (rm == null) {
      throw new YarnException("Can not find the active RM.");
    }
    else if (nodeManagers.length == rm.getClientRMService()
        .getClusterMetrics(req).getClusterMetrics().getNumNodeManagers()) {
      LOG.info("All Node Managers connected in MiniYARNCluster");
      return true;
    }
    Thread.sleep(10);
  }
  LOG.info("Node Managers did not connect within 5000ms");
  return false;
}
 

import org.apache.hadoop.yarn.api.protocolrecords.GetClusterMetricsRequest; //导入依赖的package包/类
/**
 * Wait for all the NodeManagers to connect to the ResourceManager.
 *
 * @param timeout Time to wait (sleeps in 100 ms intervals) in milliseconds.
 * @return true if all NodeManagers connect to the (Active)
 * ResourceManager, false otherwise.
 * @throws YarnException
 * @throws InterruptedException
 */
public boolean waitForNodeManagersToConnect(long timeout)
    throws YarnException, InterruptedException {
    GetClusterMetricsRequest req = GetClusterMetricsRequest.newInstance();
    for (int i = 0; i < timeout / 100; i++) {
        ResourceManager rm = getResourceManager();
        if (rm == null) {
            throw new YarnException("Can not find the active RM.");
        }
        else if (nodeManagers.length == rm.getClientRMService()
                                          .getClusterMetrics(req).getClusterMetrics().getNumNodeManagers()) {
            return true;
        }
        Thread.sleep(100);
    }
    return false;
}
 

import org.apache.hadoop.yarn.api.protocolrecords.GetClusterMetricsRequest; //导入依赖的package包/类
@Override
public GetClusterMetricsResponse getClusterMetrics(
    GetClusterMetricsRequest request) throws YarnException,
    IOException {
  GetClusterMetricsRequestProto requestProto =
      ((GetClusterMetricsRequestPBImpl) request).getProto();
  try {
    return new GetClusterMetricsResponsePBImpl(proxy.getClusterMetrics(null,
      requestProto));
  } catch (ServiceException e) {
    RPCUtil.unwrapAndThrowException(e);
    return null;
  }
}
 

import org.apache.hadoop.yarn.api.protocolrecords.GetClusterMetricsRequest; //导入依赖的package包/类
@Override
public GetClusterMetricsResponse getClusterMetrics(
    GetClusterMetricsRequest request) throws YarnException {
  GetClusterMetricsResponse response = recordFactory
      .newRecordInstance(GetClusterMetricsResponse.class);
  YarnClusterMetrics ymetrics = recordFactory
      .newRecordInstance(YarnClusterMetrics.class);
  ymetrics.setNumNodeManagers(this.rmContext.getRMNodes().size());
  response.setClusterMetrics(ymetrics);
  return response;
}
 

import org.apache.hadoop.yarn.api.protocolrecords.GetClusterMetricsRequest; //导入依赖的package包/类
@Override
public YarnClusterMetrics getYarnClusterMetrics() throws YarnException,
    IOException {
  GetClusterMetricsRequest request =
      Records.newRecord(GetClusterMetricsRequest.class);
  GetClusterMetricsResponse response = rmClient.getClusterMetrics(request);
  return response.getClusterMetrics();
}
 

import org.apache.hadoop.yarn.api.protocolrecords.GetClusterMetricsRequest; //导入依赖的package包/类
@Override
public GetClusterMetricsResponse getClusterMetrics(
    GetClusterMetricsRequest request) throws YarnException {
  resetStartFailoverFlag(true);

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

  // create GetClusterMetricsResponse with fake YarnClusterMetrics
  GetClusterMetricsResponse response =
      GetClusterMetricsResponse.newInstance(
          createFakeYarnClusterMetrics());
  return response;
}
 

import org.apache.hadoop.yarn.api.protocolrecords.GetClusterMetricsRequest; //导入依赖的package包/类
/**
 * Wait for all the NodeManagers to connect to the ResourceManager.
 *
 * @param timeout Time to wait (sleeps in 100 ms intervals) in milliseconds.
 * @return true if all NodeManagers connect to the (Active)
 * ResourceManager, false otherwise.
 * @throws YarnException
 * @throws InterruptedException
 */
public boolean waitForNodeManagersToConnect(long timeout)
    throws YarnException, InterruptedException {
  ResourceManager rm = getResourceManager();
  GetClusterMetricsRequest req = GetClusterMetricsRequest.newInstance();

  for (int i = 0; i < timeout / 100; i++) {
    if (nodeManagers.length == rm.getClientRMService().getClusterMetrics(req)
        .getClusterMetrics().getNumNodeManagers()) {
      return true;
    }
    Thread.sleep(100);
  }
  return false;
}