@Override
  public SubClusterId getHomeSubcluster(
      ApplicationSubmissionContext appSubmissionContext,
      List<SubClusterId> blackListSubClusters) throws YarnException {

    // null checks and default-queue behavior
    validate(appSubmissionContext);

    List<ResourceRequest> rrList =
        appSubmissionContext.getAMContainerResourceRequests();

    // Fast path for FailForward to WeightedRandomRouterPolicy
    if (rrList == null || rrList.isEmpty() || (rrList.size() == 1
        && ResourceRequest.isAnyLocation(rrList.get(0).getResourceName()))) {
      return super.getHomeSubcluster(appSubmissionContext, blackListSubClusters);
    }

    if (rrList.size() != 3) {
      throw new FederationPolicyException(
          "Invalid number of resource requests: " + rrList.size());
    }

    Map<SubClusterId, SubClusterInfo> activeSubClusters = getActiveSubclusters();
    Set<SubClusterId> validSubClusters = activeSubClusters.keySet();
    FederationPolicyUtils.validateSubClusterAvailability(activeSubClusters.keySet(),
        blackListSubClusters);

    if (blackListSubClusters != null) {
      // Remove from the active SubClusters from StateStore the blacklisted ones
      validSubClusters.removeAll(blackListSubClusters);
    }

    try {
      // With three requests, this has been processed by the
      // ResourceRequestInterceptorREST, and should have
      // node, rack, and any
      SubClusterId targetId = null;
      ResourceRequest nodeRequest = null;
      ResourceRequest rackRequest = null;
      ResourceRequest anyRequest = null;

      for (ResourceRequest rr : rrList) {
        // Handle "node" requests
        try {
          targetId = resolver.getSubClusterForNode(rr.getResourceName());
          nodeRequest = rr;
        } catch (YarnException e) {
          LOG.error("Cannot resolve node : {}.", e.getMessage());
        }
        // Handle "rack" requests
        try {
          resolver.getSubClustersForRack(rr.getResourceName());
          rackRequest = rr;
        } catch (YarnException e) {
          LOG.error("Cannot resolve rack : {}.", e.getMessage());
        }
        // Handle "ANY" requests
        if (ResourceRequest.isAnyLocation(rr.getResourceName())) {
          anyRequest = rr;
          continue;
        }
      }

      if (nodeRequest == null) {
        throw new YarnException("Missing node request.");
      }
      if (rackRequest == null) {
        throw new YarnException("Missing rack request.");
      }
      if (anyRequest == null) {
        throw new YarnException("Missing any request.");
      }

      LOG.info("Node request: {} , Rack request: {} , Any request: {}.",
          nodeRequest.getResourceName(), rackRequest.getResourceName(),
          anyRequest.getResourceName());

      // Handle "node" requests
      if (validSubClusters.contains(targetId) && enabledSCs
          .contains(targetId)) {
        LOG.info("Node {} is in SubCluster: {}.", nodeRequest.getResourceName(), targetId);
        return targetId;
      } else {
        throw new YarnException("The node " + nodeRequest.getResourceName()
            + " is in a blacklist SubCluster or not active. ");
      }
    } catch (YarnException e) {
      LOG.error("Validating resource requests failed, " +
          "Falling back to WeightedRandomRouterPolicy placement : {}.", e.getMessage());
      // FailForward to WeightedRandomRouterPolicy
      // Overwrite request to use a default ANY
      ResourceRequest amReq = Records.newRecord(ResourceRequest.class);
      amReq.setPriority(appSubmissionContext.getPriority());
      amReq.setResourceName(ResourceRequest.ANY);
      amReq.setCapability(appSubmissionContext.getResource());
      amReq.setNumContainers(1);
      amReq.setRelaxLocality(true);
      amReq.setNodeLabelExpression(appSubmissionContext.getNodeLabelExpression());
      amReq.setExecutionTypeRequest(ExecutionTypeRequest.newInstance(ExecutionType.GUARANTEED));
      appSubmissionContext.setAMContainerResourceRequests(Collections.singletonList(amReq));
      return super.getHomeSubcluster(appSubmissionContext, blackListSubClusters);
    }
  }
@Test
  public void testMultipleResourceRequests() throws YarnException {
    List<ResourceRequest> requests = new ArrayList<ResourceRequest>();
    requests.add(ResourceRequest
        .newInstance(Priority.UNDEFINED, "node1", Resource.newInstance(10, 1),
            1));
    requests.add(ResourceRequest
        .newInstance(Priority.UNDEFINED, "node2", Resource.newInstance(10, 1),
            1));
    ApplicationSubmissionContext asc = ApplicationSubmissionContext
        .newInstance(null, null, null, null, null, false, false, 0,
            Resources.none(), null, false, null, null);
    asc.setAMContainerResourceRequests(requests);
    try {
      ((FederationRouterPolicy) getPolicy()).getHomeSubcluster(asc, null);
      Assert.fail();
    } catch (FederationPolicyException e) {
      Assert.assertTrue(
          e.getMessage().startsWith("Invalid number of resource requests: "));
    }
  }