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

@Override
public Priority getPriority() {
  NMContainerStatusProtoOrBuilder p = viaProto ? proto : builder;
  if (this.priority != null) {
    return this.priority;
  }
  if (!p.hasPriority()) {
    return null;
  }
  this.priority = convertFromProtoFormat(p.getPriority());
  return this.priority;
}

@Override
public void handle(ContainerAllocatorEvent event) {
  ContainerId cId =
      ContainerId.newContainerId(getContext().getApplicationAttemptId(),
        containerCount++);
  NodeId nodeId = NodeId.newInstance(NM_HOST, NM_PORT);
  Resource resource = Resource.newInstance(1234, 2, 2);
  ContainerTokenIdentifier containerTokenIdentifier =
      new ContainerTokenIdentifier(cId, nodeId.toString(), "user",
      resource, System.currentTimeMillis() + 10000, 42, 42,
      Priority.newInstance(0), 0);
  Token containerToken = newContainerToken(nodeId, "password".getBytes(),
        containerTokenIdentifier);
  Container container = Container.newInstance(cId, nodeId,
      NM_HOST + ":" + NM_HTTP_PORT, resource, null, containerToken);
  JobID id = TypeConverter.fromYarn(applicationId);
  JobId jobId = TypeConverter.toYarn(id);
  getContext().getEventHandler().handle(new JobHistoryEvent(jobId, 
      new NormalizedResourceEvent(
          org.apache.hadoop.mapreduce.TaskType.REDUCE,
      100)));
  getContext().getEventHandler().handle(new JobHistoryEvent(jobId, 
      new NormalizedResourceEvent(
          org.apache.hadoop.mapreduce.TaskType.MAP,
      100)));
  getContext().getEventHandler().handle(
      new TaskAttemptContainerAssignedEvent(event.getAttemptID(),
          container, null));
}

@Public
@Unstable
public static ContainerHistoryData newInstance(ContainerId containerId,
    Resource allocatedResource, NodeId assignedNode, Priority priority,
    long startTime, long finishTime, String diagnosticsInfo,
    int containerExitCode, ContainerState containerState) {
  ContainerHistoryData containerHD = new ContainerHistoryData();
  containerHD.setContainerId(containerId);
  containerHD.setAllocatedResource(allocatedResource);
  containerHD.setAssignedNode(assignedNode);
  containerHD.setPriority(priority);
  containerHD.setStartTime(startTime);
  containerHD.setFinishTime(finishTime);
  containerHD.setDiagnosticsInfo(diagnosticsInfo);
  containerHD.setContainerExitStatus(containerExitCode);
  containerHD.setContainerState(containerState);
  return containerHD;
}

@Override
protected RMContainerTokenSecretManager createContainerTokenSecretManager(
    Configuration conf) {
  return new RMContainerTokenSecretManager(conf) {

    @Override
    public Token createContainerToken(ContainerId containerId,
        NodeId nodeId, String appSubmitter, Resource capability,
        Priority priority, long createTime,
        LogAggregationContext logAggregationContext) {
      numRetries++;
      return super.createContainerToken(containerId, nodeId, appSubmitter,
        capability, priority, createTime, logAggregationContext);
    }
  };
}

@Override
public void updateDemand() {
  demand = Resources.createResource(0);
  // Demand is current consumption plus outstanding requests
  Resources.addTo(demand, getCurrentConsumption());

  // Add up outstanding resource requests
  synchronized (this) {
    for (Priority p : getPriorities()) {
      for (ResourceRequest r : getResourceRequests(p).values()) {
        Resource total = Resources.multiply(r.getCapability(), r.getNumContainers());
        Resources.addTo(demand, total);
      }
    }
  }
}

/**
 * Instantiates a {@link ContainerRequest} with the given constraints.
 * 
 * @param capability
 *          The {@link Resource} to be requested for each container.
 * @param nodes
 *          Any hosts to request that the containers are placed on.
 * @param racks
 *          Any racks to request that the containers are placed on. The
 *          racks corresponding to any hosts requested will be automatically
 *          added to this list.
 * @param priority
 *          The priority at which to request the containers. Higher
 *          priorities have lower numerical values.
 * @param relaxLocality
 *          If true, containers for this request may be assigned on hosts
 *          and racks other than the ones explicitly requested.
 * @param nodeLabelsExpression
 *          Set node labels to allocate resource, now we only support
 *          asking for only a single node label
 */
public ContainerRequest(Resource capability, String[] nodes,
    String[] racks, Priority priority, boolean relaxLocality,
    String nodeLabelsExpression) {
  // Validate request
  Preconditions.checkArgument(capability != null,
      "The Resource to be requested for each container " +
          "should not be null ");
  Preconditions.checkArgument(priority != null,
      "The priority at which to request containers should not be null ");
  Preconditions.checkArgument(
          !(!relaxLocality && (racks == null || racks.length == 0) 
              && (nodes == null || nodes.length == 0)),
          "Can't turn off locality relaxation on a " + 
          "request with no location constraints");
  this.capability = capability;
  this.nodes = (nodes != null ? ImmutableList.copyOf(nodes) : null);
  this.racks = (racks != null ? ImmutableList.copyOf(racks) : null);
  this.priority = priority;
  this.relaxLocality = relaxLocality;
  this.nodeLabelsExpression = nodeLabelsExpression;
}

public ResourceRequest createResourceReq(String resource, int memory, int priority,
    int containers, String labelExpression) throws Exception {
  ResourceRequest req = Records.newRecord(ResourceRequest.class);
  req.setResourceName(resource);
  req.setNumContainers(containers);
  Priority pri = Records.newRecord(Priority.class);
  pri.setPriority(priority);
  req.setPriority(pri);
  Resource capability = Records.newRecord(Resource.class);
  capability.setMemory(memory);
  req.setCapability(capability);
  if (labelExpression != null) {
   req.setNodeLabelExpression(labelExpression); 
  }
  return req;
}

/**
 * ContainerRequests with locality relaxation cannot be made at the same
 * priority as ContainerRequests without locality relaxation.
 */
private void checkLocalityRelaxationConflict(Priority priority,
    Collection<String> locations, boolean relaxLocality) {
  Map<String, TreeMap<Resource, ResourceRequestInfo>> remoteRequests =
      this.remoteRequestsTable.get(priority);
  if (remoteRequests == null) {
    return;
  }
  // Locality relaxation will be set to relaxLocality for all implicitly
  // requested racks. Make sure that existing rack requests match this.
  for (String location : locations) {
      TreeMap<Resource, ResourceRequestInfo> reqs =
          remoteRequests.get(location);
      if (reqs != null && !reqs.isEmpty()) {
        boolean existingRelaxLocality =
            reqs.values().iterator().next().remoteRequest.getRelaxLocality();
        if (relaxLocality != existingRelaxLocality) {
          throw new InvalidContainerRequestException("Cannot submit a "
              + "ContainerRequest asking for location " + location
              + " with locality relaxation " + relaxLocality + " when it has "
              + "already been requested with locality relaxation " + existingRelaxLocality);
        }
      }
    }
}

@Test (expected = InvalidContainerRequestException.class)
public void testDifferentLocalityRelaxationSamePriority() {
  AMRMClientImpl<ContainerRequest> client =
      new AMRMClientImpl<ContainerRequest>();
  Configuration conf = new Configuration();
  conf.setClass(
      CommonConfigurationKeysPublic.NET_TOPOLOGY_NODE_SWITCH_MAPPING_IMPL_KEY,
      MyResolver.class, DNSToSwitchMapping.class);
  client.init(conf);

  Resource capability = Resource.newInstance(1024, 1, 1);
  ContainerRequest request1 =
      new ContainerRequest(capability, new String[] {"host1", "host2"},
          null, Priority.newInstance(1), false);
  client.addContainerRequest(request1);
  ContainerRequest request2 =
      new ContainerRequest(capability, new String[] {"host3"},
          null, Priority.newInstance(1), true);
  client.addContainerRequest(request2);
}

public static Token createContainerToken(ContainerId cId, long rmIdentifier,
    NodeId nodeId, String user,
    NMContainerTokenSecretManager containerTokenSecretManager,
    LogAggregationContext logAggregationContext)
    throws IOException {
  Resource r = BuilderUtils.newResource(1024, 1);
  ContainerTokenIdentifier containerTokenIdentifier =
      new ContainerTokenIdentifier(cId, nodeId.toString(), user, r,
        System.currentTimeMillis() + 100000L, 123, rmIdentifier,
        Priority.newInstance(0), 0, logAggregationContext);
  Token containerToken =
      BuilderUtils
        .newContainerToken(nodeId, containerTokenSecretManager
          .retrievePassword(containerTokenIdentifier),
          containerTokenIdentifier);
  return containerToken;
}

@Test
public void testFillInRacks() {
  AMRMClientImpl<ContainerRequest> client =
      new AMRMClientImpl<ContainerRequest>();

  Configuration conf = new Configuration();
  conf.setClass(
      CommonConfigurationKeysPublic.NET_TOPOLOGY_NODE_SWITCH_MAPPING_IMPL_KEY,
      MyResolver.class, DNSToSwitchMapping.class);
  client.init(conf);

  Resource capability = Resource.newInstance(1024, 1, 1);
  ContainerRequest request =
      new ContainerRequest(capability, new String[] {"host1", "host2"},
          new String[] {"/rack2"}, Priority.newInstance(1));
  client.addContainerRequest(request);
  verifyResourceRequest(client, request, "host1", true);
  verifyResourceRequest(client, request, "host2", true);
  verifyResourceRequest(client, request, "/rack1", true);
  verifyResourceRequest(client, request, "/rack2", true);
  verifyResourceRequest(client, request, ResourceRequest.ANY, true);
}

private int assignNodeLocalContainers(FiCaSchedulerNode node, 
    FiCaSchedulerApp application, Priority priority) {
  int assignedContainers = 0;
  ResourceRequest request = 
    application.getResourceRequest(priority, node.getNodeName());
  if (request != null) {
    // Don't allocate on this node if we don't need containers on this rack
    ResourceRequest rackRequest =
        application.getResourceRequest(priority, 
            node.getRMNode().getRackName());
    if (rackRequest == null || rackRequest.getNumContainers() <= 0) {
      return 0;
    }

    int assignableContainers = 
      Math.min(
          getMaxAllocatableContainers(application, priority, node, 
              NodeType.NODE_LOCAL), 
              request.getNumContainers());
    assignedContainers = 
      assignContainer(node, application, priority, 
          assignableContainers, request, NodeType.NODE_LOCAL);
  }
  return assignedContainers;
}

private int assignRackLocalContainers(FiCaSchedulerNode node, 
    FiCaSchedulerApp application, Priority priority) {
  int assignedContainers = 0;
  ResourceRequest request = 
    application.getResourceRequest(priority, node.getRMNode().getRackName());
  if (request != null) {
    // Don't allocate on this rack if the application doens't need containers
    ResourceRequest offSwitchRequest =
        application.getResourceRequest(priority, ResourceRequest.ANY);
    if (offSwitchRequest.getNumContainers() <= 0) {
      return 0;
    }

    int assignableContainers = 
      Math.min(
          getMaxAllocatableContainers(application, priority, node, 
              NodeType.RACK_LOCAL), 
              request.getNumContainers());
    assignedContainers = 
      assignContainer(node, application, priority, 
          assignableContainers, request, NodeType.RACK_LOCAL);
  }
  return assignedContainers;
}

@Test (expected = InvalidContainerRequestException.class)
public void testLocalityRelaxationDifferentLevels() {
  AMRMClientImpl<ContainerRequest> client =
      new AMRMClientImpl<ContainerRequest>();
  Configuration conf = new Configuration();
  conf.setClass(
      CommonConfigurationKeysPublic.NET_TOPOLOGY_NODE_SWITCH_MAPPING_IMPL_KEY,
      MyResolver.class, DNSToSwitchMapping.class);
  client.init(conf);

  Resource capability = Resource.newInstance(1024, 1, 1);
  ContainerRequest request1 =
      new ContainerRequest(capability, new String[] {"host1", "host2"},
          null, Priority.newInstance(1), false);
  client.addContainerRequest(request1);
  ContainerRequest request2 =
      new ContainerRequest(capability, null,
          new String[] {"rack1"}, Priority.newInstance(1), true);
  client.addContainerRequest(request2);
}

public synchronized void showRequests() {
  if (LOG.isDebugEnabled()) {
    for (Priority priority : getPriorities()) {
      Map<String, ResourceRequest> requests = getResourceRequests(priority);
      if (requests != null) {
        LOG.debug("showRequests:" + " application=" + getApplicationId()
            + " headRoom=" + getHeadroom() + " currentConsumption="
            + attemptResourceUsage.getUsed().getMemory());
        for (ResourceRequest request : requests.values()) {
          LOG.debug("showRequests:" + " application=" + getApplicationId()
              + " request=" + request);
        }
      }
    }
  }
}

private synchronized void unreserveInternal(
    Priority priority, FSSchedulerNode node) {
  Map<NodeId, RMContainer> reservedContainers = 
      this.reservedContainers.get(priority);
  RMContainer reservedContainer = reservedContainers.remove(node.getNodeID());
  if (reservedContainers.isEmpty()) {
    this.reservedContainers.remove(priority);
  }

  // Reset the re-reservation count
  resetReReservations(priority);

  Resource resource = reservedContainer.getContainer().getResource();
  this.attemptResourceUsage.decReserved(resource);

  LOG.info("Application " + getApplicationId() + " unreserved " + " on node "
      + node + ", currently has " + reservedContainers.size()
      + " at priority " + priority + "; currentReservation "
      + this.attemptResourceUsage.getReserved());
}

/**
 * Return the level at which we are allowed to schedule containers, given the
 * current size of the cluster and thresholds indicating how many nodes to
 * fail at (as a fraction of cluster size) before relaxing scheduling
 * constraints.
 */
public synchronized NodeType getAllowedLocalityLevel(Priority priority,
    int numNodes, double nodeLocalityThreshold, double rackLocalityThreshold) {
  // upper limit on threshold
  if (nodeLocalityThreshold > 1.0) { nodeLocalityThreshold = 1.0; }
  if (rackLocalityThreshold > 1.0) { rackLocalityThreshold = 1.0; }

  // If delay scheduling is not being used, can schedule anywhere
  if (nodeLocalityThreshold < 0.0 || rackLocalityThreshold < 0.0) {
    return NodeType.OFF_SWITCH;
  }

  // Default level is NODE_LOCAL
  if (!allowedLocalityLevel.containsKey(priority)) {
    allowedLocalityLevel.put(priority, NodeType.NODE_LOCAL);
    return NodeType.NODE_LOCAL;
  }

  NodeType allowed = allowedLocalityLevel.get(priority);

  // If level is already most liberal, we're done
  if (allowed.equals(NodeType.OFF_SWITCH)) return NodeType.OFF_SWITCH;

  double threshold = allowed.equals(NodeType.NODE_LOCAL) ? nodeLocalityThreshold :
    rackLocalityThreshold;

  // Relax locality constraints once we've surpassed threshold.
  if (getSchedulingOpportunities(priority) > (numNodes * threshold)) {
    if (allowed.equals(NodeType.NODE_LOCAL)) {
      allowedLocalityLevel.put(priority, NodeType.RACK_LOCAL);
      resetSchedulingOpportunities(priority);
    }
    else if (allowed.equals(NodeType.RACK_LOCAL)) {
      allowedLocalityLevel.put(priority, NodeType.OFF_SWITCH);
      resetSchedulingOpportunities(priority);
    }
  }
  return allowedLocalityLevel.get(priority);
}

/**
 * Create a ContainerRequest
 * @param id the task that asks for the container
 * @param capability container resource quota
 * @param hosts the expected host addresses
 * @param racks the rack addresses for the expected hosts
 * @param priority resource priority
 */
public ContainerRequest(Id id, Resource capability, String[] hosts, String[] racks,
    Priority priority) {
  this.id = id;
  this.capability = capability;
  this.hosts = hosts;
  this.racks = racks;
  this.priority = priority;
}

@Test(timeout=20000)
public void testAppSubmissionWithInvalidDelegationToken() throws Exception {
  Configuration conf = new Configuration();
  conf.set(
      CommonConfigurationKeysPublic.HADOOP_SECURITY_AUTHENTICATION,
      "kerberos");
  UserGroupInformation.setConfiguration(conf);
  MockRM rm = new MockRM(conf) {
    @Override
    protected void doSecureLogin() throws IOException {
      // Skip the login.
    }
  };
  ByteBuffer tokens = ByteBuffer.wrap("BOGUS".getBytes()); 
  ContainerLaunchContext amContainer =
      ContainerLaunchContext.newInstance(
          new HashMap<String, LocalResource>(), new HashMap<String, String>(),
          new ArrayList<String>(), new HashMap<String, ByteBuffer>(), tokens,
          new HashMap<ApplicationAccessType, String>());
  ApplicationSubmissionContext appSubContext =
      ApplicationSubmissionContext.newInstance(
          ApplicationId.newInstance(1234121, 0),
          "BOGUS", "default", Priority.UNDEFINED, amContainer, false,
          true, 1, Resource.newInstance(1024, 1, 1), "BOGUS");
  SubmitApplicationRequest request =
      SubmitApplicationRequest.newInstance(appSubContext);
  try {
    rm.getClientRMService().submitApplication(request);
    fail("Error was excepted.");
  } catch (YarnException e) {
    Assert.assertTrue(e.getMessage().contains(
        "Bad header found in token storage"));
  }
}

protected ResourceRequest createResourceRequest(
        Resource resource, String host, int priority, int numContainers) {
  ResourceRequest request = recordFactory
      .newRecordInstance(ResourceRequest.class);
  request.setCapability(resource);
  request.setResourceName(host);
  request.setNumContainers(numContainers);
  Priority prio = recordFactory.newRecordInstance(Priority.class);
  prio.setPriority(priority);
  request.setPriority(prio);
  return request;
}

@SuppressWarnings("deprecation")
protected void writeContainerStartData(ContainerId containerId)
    throws IOException {
  store.containerStarted(ContainerStartData.newInstance(containerId,
    Resource.newInstance(0, 0, 0), NodeId.newInstance("localhost", 0),
    Priority.newInstance(containerId.getId()), 0));
}

public ContainerTokenIdentifier(ContainerId containerID, String hostName,
    String appSubmitter, Resource r, long expiryTimeStamp, int masterKeyId,
    long rmIdentifier, Priority priority, long creationTime,
    LogAggregationContext logAggregationContext) {
  ContainerTokenIdentifierProto.Builder builder = 
      ContainerTokenIdentifierProto.newBuilder();
  if (containerID != null) {
    builder.setContainerId(((ContainerIdPBImpl)containerID).getProto());
  }
  builder.setNmHostAddr(hostName);
  builder.setAppSubmitter(appSubmitter);
  if (r != null) {
    builder.setResource(((ResourcePBImpl)r).getProto());
  }
  builder.setExpiryTimeStamp(expiryTimeStamp);
  builder.setMasterKeyId(masterKeyId);
  builder.setRmIdentifier(rmIdentifier);
  if (priority != null) {
    builder.setPriority(((PriorityPBImpl)priority).getProto());
  }
  builder.setCreationTime(creationTime);

  if (logAggregationContext != null) {
    builder.setLogAggregationContext(
        ((LogAggregationContextPBImpl)logAggregationContext).getProto());
  }
  proto = builder.build();
}

public ContainerId assignContainer(String nodeName, boolean isReduce) {
  ContainerId containerId =
      ContainerId.newContainerId(attemptId, nextContainerId++);
  Priority priority = isReduce ? RMContainerAllocator.PRIORITY_REDUCE
      : RMContainerAllocator.PRIORITY_MAP;
  Container container = Container.newInstance(containerId,
      NodeId.newInstance(nodeName, 1234), nodeName + ":5678",
    Resource.newInstance(1024, 1, 1), priority, null);
  containersToAllocate.add(container);
  return containerId;
}

@Test(timeout=30000)
public void testAskWithInvalidNodeLabels() {
  AMRMClientImpl<ContainerRequest> client =
      new AMRMClientImpl<ContainerRequest>();

  // specified exp with more than one node labels
  verifyAddRequestFailed(client,
      new ContainerRequest(Resource.newInstance(1024, 1), null, null,
          Priority.UNDEFINED, true, "x && y"));
}

public ContainerRequest(TaskAttemptId attemptID,
    Resource capability, String[] hosts, String[] racks,
    Priority priority, long requestTimeMs,String nodeLabelExpression) {
  this.attemptID = attemptID;
  this.capability = capability;
  this.hosts = hosts;
  this.racks = racks;
  this.priority = priority;
  this.requestTimeMs = requestTimeMs;
  this.nodeLabelExpression = nodeLabelExpression;
}

@Override
public Priority getPriority() {
  ResourceRequestProtoOrBuilder p = viaProto ? proto : builder;
  if (this.priority != null) {
    return this.priority;
  }
  if (!p.hasPriority()) {
    return null;
  }
  this.priority = convertFromProtoFormat(p.getPriority());
  return this.priority;
}

@Override
public Priority getPriority() {
  ApplicationSubmissionContextProtoOrBuilder p = viaProto ? proto : builder;
  if (this.priority != null) {
    return this.priority;
  }
  if (!p.hasPriority()) {
    return null;
  }
  this.priority = convertFromProtoFormat(p.getPriority());
  return this.priority;
}

@Public
@Unstable
public static ContainerStartData newInstance(ContainerId containerId,
    Resource allocatedResource, NodeId assignedNode, Priority priority,
    long startTime) {
  ContainerStartData containerSD =
      Records.newRecord(ContainerStartData.class);
  containerSD.setContainerId(containerId);
  containerSD.setAllocatedResource(allocatedResource);
  containerSD.setAssignedNode(assignedNode);
  containerSD.setPriority(priority);
  containerSD.setStartTime(startTime);
  return containerSD;
}

@Test
public void testGetContainerReport() throws Exception {
  ApplicationCLI cli = createAndGetAppCLI();
  ApplicationId applicationId = ApplicationId.newInstance(1234, 5);
  ApplicationAttemptId attemptId = ApplicationAttemptId.newInstance(
      applicationId, 1);
  ContainerId containerId = ContainerId.newContainerId(attemptId, 1);
  ContainerReport container = ContainerReport.newInstance(containerId, null,
      NodeId.newInstance("host", 1234), Priority.UNDEFINED, 1234, 5678,
      "diagnosticInfo", "logURL", 0, ContainerState.COMPLETE,
      "http://" + NodeId.newInstance("host", 2345).toString());
  when(client.getContainerReport(any(ContainerId.class))).thenReturn(
      container);
  int result = cli.run(new String[] { "container", "-status",
      containerId.toString() });
  assertEquals(0, result);
  verify(client).getContainerReport(containerId);
  ByteArrayOutputStream baos = new ByteArrayOutputStream();
  PrintWriter pw = new PrintWriter(baos);
  pw.println("Container Report : ");
  pw.println("\tContainer-Id : container_1234_0005_01_000001");
  pw.println("\tStart-Time : 1234");
  pw.println("\tFinish-Time : 5678");
  pw.println("\tState : COMPLETE");
  pw.println("\tLOG-URL : logURL");
  pw.println("\tHost : host:1234");
  pw.println("\tNodeHttpAddress : http://host:2345");
  pw.println("\tDiagnostics : diagnosticInfo");
  pw.close();
  String appReportStr = baos.toString("UTF-8");
  Assert.assertEquals(appReportStr, sysOutStream.toString());
  verify(sysOut, times(1)).println(isA(String.class));
}

public static NMContainerStatus newInstance(ContainerId containerId,
    ContainerState containerState, Resource allocatedResource,
    String diagnostics, int containerExitStatus, Priority priority,
    long creationTime) {
  NMContainerStatus status =
      Records.newRecord(NMContainerStatus.class);
  status.setContainerId(containerId);
  status.setContainerState(containerState);
  status.setAllocatedResource(allocatedResource);
  status.setDiagnostics(diagnostics);
  status.setContainerExitStatus(containerExitStatus);
  status.setPriority(priority);
  status.setCreationTime(creationTime);
  return status;
}

@Override
public void setPriority(Priority priority) {
  maybeInitBuilder();
  if (priority == null) 
    builder.clearPriority();
  this.priority = priority;
}

@Test
public void testInvalidValidWhenOldRemoved() {
  AMRMClientImpl<ContainerRequest> client =
      new AMRMClientImpl<ContainerRequest>();
  Configuration conf = new Configuration();
  conf.setClass(
      CommonConfigurationKeysPublic.NET_TOPOLOGY_NODE_SWITCH_MAPPING_IMPL_KEY,
      MyResolver.class, DNSToSwitchMapping.class);
  client.init(conf);

  Resource capability = Resource.newInstance(1024, 1, 1);
  ContainerRequest request1 =
      new ContainerRequest(capability, new String[] {"host1", "host2"},
          null, Priority.newInstance(1), false);
  client.addContainerRequest(request1);

  client.removeContainerRequest(request1);

  ContainerRequest request2 =
      new ContainerRequest(capability, new String[] {"host3"},
          null, Priority.newInstance(1), true);
  client.addContainerRequest(request2);

  client.removeContainerRequest(request2);

  ContainerRequest request3 =
      new ContainerRequest(capability, new String[] {"host1", "host2"},
          null, Priority.newInstance(1), false);
  client.addContainerRequest(request3);

  client.removeContainerRequest(request3);

  ContainerRequest request4 =
      new ContainerRequest(capability, null,
          new String[] {"rack1"}, Priority.newInstance(1), true);
  client.addContainerRequest(request4);

}

/**
 * Create and return a container object reflecting an allocation for the
 * given appliction on the given node with the given capability and
 * priority.
 */
public Container createContainer(
    FSSchedulerNode node, Resource capability, Priority priority) {

  NodeId nodeId = node.getRMNode().getNodeID();
  ContainerId containerId = BuilderUtils.newContainerId(
      getApplicationAttemptId(), getNewContainerId());

  // Create the container
  Container container =
      BuilderUtils.newContainer(containerId, nodeId, node.getRMNode()
          .getHttpAddress(), capability, priority, null);

  return container;
}

synchronized private void checkForDeactivation() {
  boolean deactivate = true;
  for (Priority priority : getPriorities()) {
    ResourceRequest request = getResourceRequest(priority, ResourceRequest.ANY);
    if (request != null) {
      if (request.getNumContainers() > 0) {
        deactivate = false;
        break;
      }
    }
  }
  if (deactivate) {
    activeUsersManager.deactivateApplication(user, applicationId);
  }
}

public static Priority createMockPriority( int priority) {
//    Priority p = mock(Priority.class);
//    when(p.getPriority()).thenReturn(priority);
    Priority p = recordFactory.newRecordInstance(Priority.class);
    p.setPriority(priority);
    return p;
  }

public static ApplicationSubmissionContext newApplicationSubmissionContext(
    ApplicationId applicationId, String applicationName, String queue,
    Priority priority, ContainerLaunchContext amContainer,
    boolean isUnmanagedAM, boolean cancelTokensWhenComplete,
    int maxAppAttempts, Resource resource) {
  return newApplicationSubmissionContext(applicationId, applicationName,
    queue, priority, amContainer, isUnmanagedAM, cancelTokensWhenComplete,
    maxAppAttempts, resource, null);
}

@Test
public void testRegisterNodeManagerRequest() {
  RegisterNodeManagerRequest request =
      RegisterNodeManagerRequest.newInstance(
        NodeId.newInstance("host", 1234), 1234, Resource.newInstance(0, 0),
        "version", Arrays.asList(NMContainerStatus.newInstance(
          ContainerId.newContainerId(
            ApplicationAttemptId.newInstance(
              ApplicationId.newInstance(1234L, 1), 1), 1),
          ContainerState.RUNNING, Resource.newInstance(1024, 1), "good", -1,
          Priority.newInstance(0), 1234)), Arrays.asList(
          ApplicationId.newInstance(1234L, 1),
          ApplicationId.newInstance(1234L, 2)));

  // serialze to proto, and get request from proto
  RegisterNodeManagerRequest request1 =
      new RegisterNodeManagerRequestPBImpl(
          ((RegisterNodeManagerRequestPBImpl) request).getProto());

  // check values
  Assert.assertEquals(request1.getNMContainerStatuses().size(), request
      .getNMContainerStatuses().size());
  Assert.assertEquals(request1.getNMContainerStatuses().get(0).getContainerId(),
      request.getNMContainerStatuses().get(0).getContainerId());
  Assert.assertEquals(request1.getRunningApplications().size(), request
      .getRunningApplications().size());
  Assert.assertEquals(request1.getRunningApplications().get(0), request
      .getRunningApplications().get(0));
  Assert.assertEquals(request1.getRunningApplications().get(1), request
      .getRunningApplications().get(1));
}

/**
 * The {@link ResourceScheduler} is allocating data-local resources to the
 * application.
 * 
 * @param allocatedContainers
 *          resources allocated to the application
 */
synchronized private void allocateRackLocal(SchedulerNode node,
    Priority priority, ResourceRequest rackLocalRequest, Container container,
    List<ResourceRequest> resourceRequests) {
  // Update future requirements
  decResourceRequest(node.getRackName(), priority, rackLocalRequest);

  ResourceRequest offRackRequest = requests.get(priority).get(
      ResourceRequest.ANY);
  decrementOutstanding(offRackRequest);

  // Update cloned RackLocal and OffRack requests for recovery
  resourceRequests.add(cloneResourceRequest(rackLocalRequest));
  resourceRequests.add(cloneResourceRequest(offRackRequest));
}

private static ContainerTokenIdentifier createContainerTokenId(
    ContainerId cid, NodeId nodeId, String user,
    NMContainerTokenSecretManager secretMgr) throws IOException {
  long rmid = cid.getApplicationAttemptId().getApplicationId()
      .getClusterTimestamp();
  ContainerTokenIdentifier ctid = new ContainerTokenIdentifier(cid,
      nodeId.toString(), user, BuilderUtils.newResource(1024, 1),
      System.currentTimeMillis() + 100000L,
      secretMgr.getCurrentKey().getKeyId(), rmid,
      Priority.newInstance(0), 0);
  Token token = BuilderUtils.newContainerToken(nodeId,
      secretMgr.createPassword(ctid), ctid);
  return BuilderUtils.newContainerTokenIdentifier(token);
}

protected Token createContainerToken(ContainerId cId, Priority priority,
    long createTime) throws InvalidToken {
  Resource r = BuilderUtils.newResource(1024, 1);
  ContainerTokenIdentifier containerTokenIdentifier =
      new ContainerTokenIdentifier(cId, context.getNodeId().toString(), user,
        r, System.currentTimeMillis() + 10000L, 123, DUMMY_RM_IDENTIFIER,
        priority, createTime);
  Token containerToken =
      BuilderUtils.newContainerToken(
        context.getNodeId(),
        context.getContainerTokenSecretManager().retrievePassword(
          containerTokenIdentifier), containerTokenIdentifier);
  return containerToken;
}