RouterPolicyFacade { public SubClusterId getHomeSubcluster( ApplicationSubmissionContext appSubmissionContext, List<SubClusterId> blackListSubClusters) throws YarnException { 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(); if (queue == null) { queue = YarnConfiguration.DEFAULT_QUEUE_NAME; } SubClusterPolicyConfiguration configuration = null; try { configuration = federationFacade.getPolicyConfiguration(queue); } catch (YarnException e) { String errMsg = "There is no policy configured for the queue: " + queue + ", falling back to defaults."; LOG.warn(errMsg, e); } if (configuration == null) { LOG.warn("There is no policies configured for queue: " + queue + " we" + " fallback to default policy for: " + YarnConfiguration.DEFAULT_FEDERATION_POLICY_KEY); queue = YarnConfiguration.DEFAULT_FEDERATION_POLICY_KEY; try { configuration = federationFacade.getPolicyConfiguration(queue); } catch (YarnException e) { String errMsg = "Cannot retrieve policy configured for the queue: " + queue + ", falling back to defaults."; LOG.warn(errMsg, e); } } if (configuration == null) { configuration = cachedConfs.get(YarnConfiguration.DEFAULT_FEDERATION_POLICY_KEY); } if (!cachedConfs.containsKey(queue) || !cachedConfs.get(queue).equals(configuration)) { singlePolicyReinit(policyMap, cachedConfs, queue, configuration); } FederationRouterPolicy policy = policyMap.get(queue); if (policy == null) { throw new FederationPolicyException("No FederationRouterPolicy found " + "for queue: " + appSubmissionContext.getQueue() + " (for " + "application: " + appSubmissionContext.getApplicationId() + ") " + "and no default specified."); } return policy.getHomeSubcluster(appSubmissionContext, blackListSubClusters); } RouterPolicyFacade(Configuration conf,
      FederationStateStoreFacade facade, SubClusterResolver resolver,
      SubClusterId homeSubcluster); SubClusterId getHomeSubcluster(
      ApplicationSubmissionContext appSubmissionContext,
      List<SubClusterId> blackListSubClusters); synchronized void reset();  }
@Test public void testConfigurationUpdate() throws YarnException { ApplicationSubmissionContext applicationSubmissionContext = mock(ApplicationSubmissionContext.class); when(applicationSubmissionContext.getQueue()).thenReturn(queue1); SubClusterId chosen = routerFacade.getHomeSubcluster(applicationSubmissionContext, null); Assert.assertTrue(subClusterIds.contains(chosen)); Assert.assertTrue(routerFacade.globalPolicyMap .get(queue1) instanceof UniformRandomRouterPolicy); store.setPolicyConfiguration(SetSubClusterPolicyConfigurationRequest .newInstance(getPriorityPolicy(queue1))); chosen = routerFacade.getHomeSubcluster(applicationSubmissionContext, null); Assert.assertTrue(chosen.equals(subClusterIds.get(0))); Assert.assertTrue(routerFacade.globalPolicyMap .get(queue1) instanceof PriorityRouterPolicy); }
@Test public void testGetHomeSubcluster() throws YarnException { ApplicationSubmissionContext applicationSubmissionContext = mock(ApplicationSubmissionContext.class); when(applicationSubmissionContext.getQueue()).thenReturn(queue1); Assert.assertTrue(routerFacade.globalPolicyMap.containsKey(defQueueKey) && routerFacade.globalPolicyMap.size() == 1); SubClusterId chosen = routerFacade.getHomeSubcluster(applicationSubmissionContext, null); Assert.assertTrue(subClusterIds.contains(chosen)); Assert.assertTrue(routerFacade.globalPolicyMap.size() == 2); Assert.assertTrue(routerFacade.globalPolicyMap .get(queue1) instanceof UniformRandomRouterPolicy); routerFacade.reset(); Assert.assertTrue(routerFacade.globalPolicyMap.containsKey(defQueueKey) && routerFacade.globalPolicyMap.size() == 1); }
@Test public void testFallbacks() throws YarnException { ApplicationSubmissionContext applicationSubmissionContext = mock(ApplicationSubmissionContext.class); 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)); when(applicationSubmissionContext.getQueue()).thenReturn(""); chosen = routerFacade.getHomeSubcluster(applicationSubmissionContext, null); Assert.assertTrue(subClusterIds.contains(chosen)); Assert.assertFalse(routerFacade.globalPolicyMap.containsKey(uninitQueue)); when(applicationSubmissionContext.getQueue()).thenReturn(null); chosen = routerFacade.getHomeSubcluster(applicationSubmissionContext, null); Assert.assertTrue(subClusterIds.contains(chosen)); Assert.assertFalse(routerFacade.globalPolicyMap.containsKey(uninitQueue)); }