class LocalityMulticastAMRMProxyPolicy extends AbstractAMRMProxyPolicy {

  @Override
  public Map<SubClusterId, List<ResourceRequest>> splitResourceRequests(
      List<ResourceRequest> resourceRequests) 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.
    bookkeeper = new AllocationBookkeeper();
    bookkeeper.reinitialize(federationFacade.getSubClusters(true));

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

    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());
      } 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());
      if (LOG.isDebugEnabled()) {
        LOG.debug("ERROR resolving sub-cluster for resourceName: "
            + rr.getResourceName() + ", picked a random subcluster to forward:"
            + 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);

    for (Map.Entry<SubClusterId, List<ResourceRequest>> entry : bookkeeper
        .getAnswer().entrySet()) {
      // A new-cluster here will trigger new UAM luanch, which might take a long
      // time. We don't want too many requests stuck in this UAM before it is
      // ready and starts heartbeating
      if (!lastHeartbeatTimeStamp.containsKey(entry.getKey())) {
        lastHeartbeatTimeStamp.put(entry.getKey(), System.currentTimeMillis());
      }
    }
    return bookkeeper.getAnswer();
  }

}