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 testConfigurationUpdate() throws YarnException {

    // in this test we see what happens when the configuration is changed
    // between calls. We achieve this by changing what is in the store.

    ApplicationSubmissionContext applicationSubmissionContext =
        mock(ApplicationSubmissionContext.class);
    when(applicationSubmissionContext.getQueue()).thenReturn(queue1);

    // first call runs using standard UniformRandomRouterPolicy
    SubClusterId chosen =
        routerFacade.getHomeSubcluster(applicationSubmissionContext, null);
    Assert.assertTrue(subClusterIds.contains(chosen));
    Assert.assertTrue(routerFacade.globalPolicyMap
        .get(queue1) instanceof UniformRandomRouterPolicy);

    // then the operator changes how queue1 is routed setting it to
    // PriorityRouterPolicy with weights favoring the first subcluster in
    // subClusterIds.
    store.setPolicyConfiguration(SetSubClusterPolicyConfigurationRequest
        .newInstance(getPriorityPolicy(queue1)));

    // second call is routed by new policy PriorityRouterPolicy
    chosen = routerFacade.getHomeSubcluster(applicationSubmissionContext, null);
    Assert.assertTrue(chosen.equals(subClusterIds.get(0)));
    Assert.assertTrue(routerFacade.globalPolicyMap
        .get(queue1) instanceof PriorityRouterPolicy);
  }