protected List<ResourceRequest> packageRequests(
          List<ContainerSimulator> csList, int priority) {
    // create requests
    Map<Long, Map<String, ResourceRequest>> rackLocalRequests =
        new HashMap<>();
    Map<Long, Map<String, ResourceRequest>> nodeLocalRequests =
        new HashMap<>();
    Map<Long, ResourceRequest> anyRequests = new HashMap<>();
    for (ContainerSimulator cs : csList) {
      long allocationId = cs.getAllocationId();
      ResourceRequest anyRequest = anyRequests.get(allocationId);
      if (cs.getHostname() != null) {
        Map<String, ResourceRequest> rackLocalRequestMap;
        if (rackLocalRequests.containsKey(allocationId)) {
          rackLocalRequestMap = rackLocalRequests.get(allocationId);
        } else {
          rackLocalRequestMap = new HashMap<>();
          rackLocalRequests.put(allocationId, rackLocalRequestMap);
        }
        String[] rackHostNames = SLSUtils.getRackHostName(cs.getHostname());
        // check rack local
        String rackname = "/" + rackHostNames[0];
        if (rackLocalRequestMap.containsKey(rackname)) {
          rackLocalRequestMap.get(rackname).setNumContainers(
              rackLocalRequestMap.get(rackname).getNumContainers() + 1);
        } else {
          ResourceRequest request = createResourceRequest(cs.getResource(),
              cs.getExecutionType(), rackname, priority,
              cs.getAllocationId(), 1);
          rackLocalRequestMap.put(rackname, request);
        }
        // check node local
        Map<String, ResourceRequest> nodeLocalRequestMap;
        if (nodeLocalRequests.containsKey(allocationId)) {
          nodeLocalRequestMap = nodeLocalRequests.get(allocationId);
        } else {
          nodeLocalRequestMap = new HashMap<>();
          nodeLocalRequests.put(allocationId, nodeLocalRequestMap);
        }
        String hostname = rackHostNames[1];
        if (nodeLocalRequestMap.containsKey(hostname)) {
          nodeLocalRequestMap.get(hostname).setNumContainers(
              nodeLocalRequestMap.get(hostname).getNumContainers() + 1);
        } else {
          ResourceRequest request = createResourceRequest(cs.getResource(),
              cs.getExecutionType(), hostname, priority,
              cs.getAllocationId(), 1);
          nodeLocalRequestMap.put(hostname, request);
        }
      }
      // any
      if (anyRequest == null) {
        anyRequest = createResourceRequest(cs.getResource(),
            cs.getExecutionType(), ResourceRequest.ANY, priority,
            cs.getAllocationId(), 1);
        anyRequests.put(allocationId, anyRequest);
      } else {
        anyRequest.setNumContainers(anyRequest.getNumContainers() + 1);
      }
    }
    List<ResourceRequest> ask = new ArrayList<ResourceRequest>();
    for (Map<String, ResourceRequest> nodeLocalRequestMap :
        nodeLocalRequests.values()) {
      ask.addAll(nodeLocalRequestMap.values());
    }
    for (Map<String, ResourceRequest> rackLocalRequestMap :
        rackLocalRequests.values()) {
      ask.addAll(rackLocalRequestMap.values());
    }
    ask.addAll(anyRequests.values());
    return ask;
  }
@Test
  public void testPackageRequests() throws YarnException {
    MockAMSimulator app = new MockAMSimulator();
    List<ContainerSimulator> containerSimulators = new ArrayList<>();
    Resource resource = Resources.createResource(1024);
    int priority = 1;
    ExecutionType execType = ExecutionType.GUARANTEED;
    String type = "map";

    TaskContainerDefinition.Builder builder =
        TaskContainerDefinition.Builder.create()
        .withResource(resource)
        .withDuration(100)
        .withPriority(1)
        .withType(type)
        .withExecutionType(execType)
        .withAllocationId(-1)
        .withRequestDelay(0);

    ContainerSimulator s1 = ContainerSimulator
        .createFromTaskContainerDefinition(
            builder.withHostname("/default-rack/h1").build());
    ContainerSimulator s2 = ContainerSimulator
        .createFromTaskContainerDefinition(
            builder.withHostname("/default-rack/h1").build());
    ContainerSimulator s3 = ContainerSimulator
        .createFromTaskContainerDefinition(
            builder.withHostname("/default-rack/h2").build());

    containerSimulators.add(s1);
    containerSimulators.add(s2);
    containerSimulators.add(s3);

    List<ResourceRequest> res = app.packageRequests(containerSimulators,
        priority);

    // total 4 resource requests: any -> 1, rack -> 1, node -> 2
    // All resource requests for any would be packaged into 1.
    // All resource requests for racks would be packaged into 1 as all of them
    // are for same rack.
    // All resource requests for nodes would be packaged into 2 as there are
    // two different nodes.
    Assert.assertEquals(4, res.size());
    int anyRequestCount = 0;
    int rackRequestCount = 0;
    int nodeRequestCount = 0;

    for (ResourceRequest request : res) {
      String resourceName = request.getResourceName();
      if (resourceName.equals("*")) {
        anyRequestCount++;
      } else if (resourceName.equals("/default-rack")) {
        rackRequestCount++;
      } else {
        nodeRequestCount++;
      }
    }

    Assert.assertEquals(1, anyRequestCount);
    Assert.assertEquals(1, rackRequestCount);
    Assert.assertEquals(2, nodeRequestCount);

    containerSimulators.clear();
    s1 = ContainerSimulator.createFromTaskContainerDefinition(
        createDefaultTaskContainerDefMock(resource, priority, execType, type,
            "/default-rack/h1", 1));
    s2 = ContainerSimulator.createFromTaskContainerDefinition(
        createDefaultTaskContainerDefMock(resource, priority, execType, type,
            "/default-rack/h1", 2));
    s3 = ContainerSimulator.createFromTaskContainerDefinition(
        createDefaultTaskContainerDefMock(resource, priority, execType, type,
            "/default-rack/h2", 1));

    containerSimulators.add(s1);
    containerSimulators.add(s2);
    containerSimulators.add(s3);

    res = app.packageRequests(containerSimulators, priority);

    // total 7 resource requests: any -> 2, rack -> 2, node -> 3
    // All resource requests for any would be packaged into 2 as there are
    // two different allocation id.
    // All resource requests for racks would be packaged into 2 as all of them
    // are for same rack but for two different allocation id.
    // All resource requests for nodes would be packaged into 3 as either node
    // or allocation id is different for each request.
    Assert.assertEquals(7, res.size());

    anyRequestCount = 0;
    rackRequestCount = 0;
    nodeRequestCount = 0;

    for (ResourceRequest request : res) {
      String resourceName = request.getResourceName();
      long allocationId = request.getAllocationRequestId();
      // allocation id should be either 1 or 2
      Assert.assertTrue(allocationId == 1 || allocationId == 2);
      if (resourceName.equals("*")) {
        anyRequestCount++;
      } else if (resourceName.equals("/default-rack")) {
        rackRequestCount++;
      } else {
        nodeRequestCount++;
      }
    }

    Assert.assertEquals(2, anyRequestCount);
    Assert.assertEquals(2, rackRequestCount);
    Assert.assertEquals(3, nodeRequestCount);
  }