@Override
  public List<Container> allocateContainers(ResourceBlacklistRequest blackList,
      List<ResourceRequest> oppResourceReqs,
      ApplicationAttemptId applicationAttemptId,
      OpportunisticContainerContext opportContext, long rmIdentifier,
      String appSubmitter) throws YarnException {

    // Update black list.
    updateBlacklist(blackList, opportContext);

    // Add OPPORTUNISTIC requests to the outstanding ones.
    opportContext.addToOutstandingReqs(oppResourceReqs);
    Set<String> nodeBlackList = new HashSet<>(opportContext.getBlacklist());
    Set<String> allocatedNodes = new HashSet<>();
    List<Container> allocatedContainers = new ArrayList<>();

    // Satisfy the outstanding OPPORTUNISTIC requests.
    boolean continueLoop = true;
    while (continueLoop) {
      continueLoop = false;
      List<Map<Resource, List<Allocation>>> allocations = new ArrayList<>();
      for (SchedulerRequestKey schedulerKey :
          opportContext.getOutstandingOpReqs().descendingKeySet()) {
        // Allocated containers :
        //  Key = Requested Capability,
        //  Value = List of Containers of given cap (the actual container size
        //          might be different than what is requested, which is why
        //          we need the requested capability (key) to match against
        //          the outstanding reqs)
        int remAllocs = -1;
        int maxAllocationsPerAMHeartbeat = getMaxAllocationsPerAMHeartbeat();
        if (maxAllocationsPerAMHeartbeat > 0) {
          remAllocs =
              maxAllocationsPerAMHeartbeat - allocatedContainers.size()
                  - getTotalAllocations(allocations);
          if (remAllocs <= 0) {
            LOG.info("Not allocating more containers as we have reached max "
                    + "allocations per AM heartbeat {}",
                maxAllocationsPerAMHeartbeat);
            break;
          }
        }
        Map<Resource, List<Allocation>> allocation = allocate(
            rmIdentifier, opportContext, schedulerKey, applicationAttemptId,
            appSubmitter, nodeBlackList, allocatedNodes, remAllocs);
        if (allocation.size() > 0) {
          allocations.add(allocation);
          continueLoop = true;
        }
      }
      matchAllocation(allocations, allocatedContainers, opportContext);
    }

    return allocatedContainers;
  }
@Test
  public void testNodeLocalAllocation() throws Exception {
    ResourceBlacklistRequest blacklistRequest =
        ResourceBlacklistRequest.newInstance(
            new ArrayList<>(), new ArrayList<>());
    List<ResourceRequest> reqs =
        Arrays.asList(
            ResourceRequest.newBuilder().allocationRequestId(1)
                .priority(PRIORITY_NORMAL)
                .resourceName(ResourceRequest.ANY)
                .capability(CAPABILITY_1GB)
                .relaxLocality(true)
                .executionType(ExecutionType.OPPORTUNISTIC).build(),
            ResourceRequest.newBuilder().allocationRequestId(2)
                .priority(PRIORITY_NORMAL)
                .resourceName("/r1")
                .capability(CAPABILITY_1GB)
                .relaxLocality(true)
                .executionType(ExecutionType.OPPORTUNISTIC).build(),
            ResourceRequest.newBuilder().allocationRequestId(2)
                .priority(PRIORITY_NORMAL)
                .resourceName("h1")
                .capability(CAPABILITY_1GB)
                .relaxLocality(true)
                .executionType(ExecutionType.OPPORTUNISTIC).build(),
            ResourceRequest.newBuilder().allocationRequestId(2)
                .priority(PRIORITY_NORMAL)
                .resourceName(ResourceRequest.ANY)
                .capability(CAPABILITY_1GB)
                .relaxLocality(true)
                .executionType(ExecutionType.OPPORTUNISTIC).build(),
            ResourceRequest.newBuilder().allocationRequestId(3)
                .priority(PRIORITY_NORMAL)
                .resourceName("/r1")
                .capability(CAPABILITY_1GB)
                .relaxLocality(true)
                .executionType(ExecutionType.OPPORTUNISTIC).build(),
            ResourceRequest.newBuilder().allocationRequestId(3)
                .priority(PRIORITY_NORMAL)
                .resourceName("h1")
                .capability(CAPABILITY_1GB)
                .relaxLocality(true)
                .executionType(ExecutionType.OPPORTUNISTIC).build(),
            ResourceRequest.newBuilder().allocationRequestId(3)
                .priority(PRIORITY_NORMAL)
                .resourceName(ResourceRequest.ANY)
                .capability(CAPABILITY_1GB)
                .relaxLocality(true)
                .executionType(ExecutionType.OPPORTUNISTIC).build());
    ApplicationAttemptId appAttId = ApplicationAttemptId.newInstance(
        ApplicationId.newInstance(0L, 1), 1);

    oppCntxt.updateNodeList(
        Arrays.asList(
            RemoteNode.newInstance(
                NodeId.newInstance("h1", 1234), "h1:1234", "/r1"),
            RemoteNode.newInstance(
                NodeId.newInstance("h2", 1234), "h2:1234", "/r1"),
            RemoteNode.newInstance(
                NodeId.newInstance("h3", 1234), "h3:1234", "/r1")));

    List<Container> containers = allocator.allocateContainers(
        blacklistRequest, reqs, appAttId, oppCntxt, 1L, "luser");
    LOG.info("Containers: {}", containers);
    // all 3 containers should be allocated.
    Assert.assertEquals(3, containers.size());
    // container with allocation id 2 and 3 should be allocated on node h1
    for (Container c : containers) {
      if (c.getAllocationRequestId() == 2 || c.getAllocationRequestId() == 3) {
        Assert.assertEquals("h1:1234", c.getNodeHttpAddress());
      }
    }
  }