@Override
  public Map<SubClusterId, List<ResourceRequest>> splitResourceRequests(
      List<ResourceRequest> resourceRequests,
      Set<SubClusterId> timedOutSubClusters) throws YarnException {

    // object used to accumulate statistics about the answer, initialize with
    // active subclusters. Create a new instance per call because this method
    // can be called concurrently.
    AllocationBookkeeper bookkeeper = new AllocationBookkeeper();
    bookkeeper.reinitialize(getActiveSubclusters(), timedOutSubClusters, conf);

    List<ResourceRequest> nonLocalizedRequests = new ArrayList<>();

    SubClusterId targetId = null;
    Set<SubClusterId> targetIds = null;

    // if the RR is resolved to a local subcluster add it directly (node and
    // resolvable racks)
    for (ResourceRequest rr : resourceRequests) {
      targetId = null;
      targetIds = null;

      // Handle: ANY (accumulated for later)
      if (ResourceRequest.isAnyLocation(rr.getResourceName())) {
        nonLocalizedRequests.add(rr);
        continue;
      }

      // Handle "node" requests
      try {
        targetId = resolver.getSubClusterForNode(rr.getResourceName());

        // If needed, re-reroute node requests base on SC load
        boolean loadBasedSCSelectorEnabled =
            conf.getBoolean(LOAD_BASED_SC_SELECTOR_ENABLED, DEFAULT_LOAD_BASED_SC_SELECTOR_ENABLED);
        if (loadBasedSCSelectorEnabled) {
          int maxPendingThreshold = conf.getInt(LOAD_BASED_SC_SELECTOR_THRESHOLD,
              DEFAULT_LOAD_BASED_SC_SELECTOR_THRESHOLD);
          targetId = routeNodeRequestIfNeeded(targetId, maxPendingThreshold,
              bookkeeper.getActiveAndEnabledSC());
        }
        LOG.debug("Node request {}", rr.getResourceName());
      } catch (YarnException e) {
        // this might happen as we can't differentiate node from rack names
        // we log altogether later
      }
      if (bookkeeper.isActiveAndEnabled(targetId)) {
        bookkeeper.addLocalizedNodeRR(targetId, rr);
        continue;
      }

      // Handle "rack" requests
      try {
        targetIds = resolver.getSubClustersForRack(rr.getResourceName());
      } catch (YarnException e) {
        // this might happen as we can't differentiate node from rack names
        // we log altogether later
      }
      if (targetIds != null && targetIds.size() > 0) {
        boolean hasActive = false;
        for (SubClusterId tid : targetIds) {
          if (bookkeeper.isActiveAndEnabled(tid)) {
            bookkeeper.addRackRR(tid, rr);
            hasActive = true;
          }
        }
        if (hasActive) {
          continue;
        }
      }

      // Handle node/rack requests that the SubClusterResolver cannot map to
      // any cluster. Pick a random sub-cluster from active and enabled ones.
      targetId = getSubClusterForUnResolvedRequest(bookkeeper,
          rr.getAllocationRequestId());
      LOG.debug("ERROR resolving sub-cluster for resourceName: {}, picked a "
          + "random subcluster to forward:{}", rr.getResourceName(), targetId);
      if (targetIds != null && targetIds.size() > 0) {
        bookkeeper.addRackRR(targetId, rr);
      } else {
        bookkeeper.addLocalizedNodeRR(targetId, rr);
      }
    }

    // handle all non-localized requests (ANY)
    splitAnyRequests(nonLocalizedRequests, bookkeeper);

    // Take the split result, feed into the askBalancer
    Map<SubClusterId, List<ResourceRequest>> answer = bookkeeper.getAnswer();
    LOG.info("Before split {} RRs: {}", resourceRequests.size(),
        prettyPrintRequests(resourceRequests, this.printRRMax));

    for (Map.Entry<SubClusterId, List<ResourceRequest>> entry : bookkeeper.getAnswer().entrySet()) {
      LOG.info("After split {} has {} RRs: {}", entry.getKey(), entry.getValue().size(),
          prettyPrintRequests(entry.getValue(), this.printRRMax));
    }
    return answer;
  }
@Test(timeout = 5000)
  public void testStressPolicy() throws Exception {

    // Tests how the headroom info are used to split based on the capacity
    // each RM claims to give us.
    // Configure policy to be 100% headroom based
    getPolicyInfo().setHeadroomAlpha(1.0f);

    initializePolicy();
    addHomeSubClusterAsActive();

    int numRR = 1000;
    List<ResourceRequest> resourceRequests = createLargeRandomList(numRR);

    prepPolicyWithHeadroom(true);

    int numIterations = 1000;
    long tstart = System.currentTimeMillis();
    for (int i = 0; i < numIterations; i++) {
      Map<SubClusterId, List<ResourceRequest>> response =
          ((FederationAMRMProxyPolicy) getPolicy()).splitResourceRequests(
              resourceRequests, new HashSet<SubClusterId>());
      validateSplit(response, resourceRequests);
    }
    long tend = System.currentTimeMillis();

    LOG.info("Performed " + numIterations + " policy invocations (and "
        + "validations) in " + (tend - tstart) + "ms");
  }