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

    // the maps are concurrent, but we need to protect from reset()
    // reinitialization mid-execution by creating a new reference local to this
    // method.
    Map<String, SubClusterPolicyConfiguration> cachedConfs = globalConfMap;
    Map<String, FederationRouterPolicy> policyMap = globalPolicyMap;

    if (appSubmissionContext == null) {
      throw new FederationPolicyException(
          "The ApplicationSubmissionContext cannot be null.");
    }

    String queue = appSubmissionContext.getQueue();

    // respecting YARN behavior we assume default queue if the queue is not
    // specified. This also ensures that "null" can be used as a key to get the
    // default behavior.
    if (queue == null) {
      queue = YarnConfiguration.DEFAULT_QUEUE_NAME;
    }

    FederationRouterPolicy policy = getFederationRouterPolicy(cachedConfs, policyMap, queue);
    if (policy == null) {
      // this should never happen, as the to maps are updated together
      throw new FederationPolicyException("No FederationRouterPolicy found "
          + "for queue: " + appSubmissionContext.getQueue() + " (for "
          + "application: " + appSubmissionContext.getApplicationId() + ") "
          + "and no default specified.");
    }

    return policy.getHomeSubcluster(appSubmissionContext, blackListSubClusters);
  }
@Test
  public void testFallbacks() throws YarnException {

    // this tests the behavior of the system when the queue requested is
    // not configured (or null) and there is no default policy configured
    // for DEFAULT_FEDERATION_POLICY_KEY (*). This is our second line of
    // defense.

    ApplicationSubmissionContext applicationSubmissionContext =
        mock(ApplicationSubmissionContext.class);

    // The facade answers also for non-initialized policies (using the
    // defaultPolicy)
    String uninitQueue = "non-initialized-queue";
    when(applicationSubmissionContext.getQueue()).thenReturn(uninitQueue);
    SubClusterId chosen =
        routerFacade.getHomeSubcluster(applicationSubmissionContext, null);
    Assert.assertTrue(subClusterIds.contains(chosen));
    Assert.assertFalse(routerFacade.globalPolicyMap.containsKey(uninitQueue));

    // empty string
    when(applicationSubmissionContext.getQueue()).thenReturn("");
    chosen = routerFacade.getHomeSubcluster(applicationSubmissionContext, null);
    Assert.assertTrue(subClusterIds.contains(chosen));
    Assert.assertFalse(routerFacade.globalPolicyMap.containsKey(uninitQueue));

    // null queue also falls back to default
    when(applicationSubmissionContext.getQueue()).thenReturn(null);
    chosen = routerFacade.getHomeSubcluster(applicationSubmissionContext, null);
    Assert.assertTrue(subClusterIds.contains(chosen));
    Assert.assertFalse(routerFacade.globalPolicyMap.containsKey(uninitQueue));

  }