PriorityRouterPolicy extends AbstractRouterPolicy { @Override public SubClusterId getHomeSubcluster( ApplicationSubmissionContext appSubmissionContext, List<SubClusterId> blacklist) throws YarnException { validate(appSubmissionContext); Map<SubClusterId, SubClusterInfo> activeSubclusters = getActiveSubclusters(); FederationPolicyUtils.validateSubClusterAvailability( new ArrayList<SubClusterId>(activeSubclusters.keySet()), blacklist); Map<SubClusterIdInfo, Float> weights = getPolicyInfo().getRouterPolicyWeights(); SubClusterId chosen = null; Float currentBest = Float.MIN_VALUE; for (SubClusterId id : activeSubclusters.keySet()) { SubClusterIdInfo idInfo = new SubClusterIdInfo(id); if (blacklist != null && blacklist.contains(id)) { continue; } if (weights.containsKey(idInfo) && weights.get(idInfo) > currentBest) { currentBest = weights.get(idInfo); chosen = id; } } if (chosen == null) { throw new FederationPolicyException( "No Active Subcluster with weight vector greater than zero"); } return chosen; }  @Override SubClusterId getHomeSubcluster(
      ApplicationSubmissionContext appSubmissionContext,
      List<SubClusterId> blacklist);  }
@Test public void testPickLowestWeight() throws YarnException { SubClusterId chosen = ((FederationRouterPolicy) getPolicy()) .getHomeSubcluster(getApplicationSubmissionContext(), null); Assert.assertEquals("sc5", chosen.getId()); }
@Test public void testZeroSubClustersWithPositiveWeight() throws Exception { Map<SubClusterIdInfo, Float> routerWeights = new HashMap<>(); Map<SubClusterIdInfo, Float> amrmWeights = new HashMap<>(); for (int i = 0; i < 5; i++) { SubClusterIdInfo sc = new SubClusterIdInfo("sc" + i); SubClusterInfo sci = mock(SubClusterInfo.class); when(sci.getState()).thenReturn(SubClusterState.SC_RUNNING); when(sci.getSubClusterId()).thenReturn(sc.toId()); getActiveSubclusters().put(sc.toId(), sci); routerWeights.put(sc, 0.0f); amrmWeights.put(sc, -1.0f); } getPolicyInfo().setRouterPolicyWeights(routerWeights); getPolicyInfo().setAMRMPolicyWeights(amrmWeights); FederationPoliciesTestUtil.initializePolicyContext(getPolicy(), getPolicyInfo(), getActiveSubclusters()); intercept(FederationPolicyException.class, "No Active Subcluster with weight vector greater than zero", () -> ((FederationRouterPolicy) getPolicy()) .getHomeSubcluster(getApplicationSubmissionContext(), null)); }