LoadBasedRouterPolicy 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(); SubClusterIdInfo chosen = null; long currBestMem = -1; for (Map.Entry<SubClusterId, SubClusterInfo> entry : activeSubclusters .entrySet()) { if (blacklist != null && blacklist.contains(entry.getKey())) { continue; } SubClusterIdInfo id = new SubClusterIdInfo(entry.getKey()); if (weights.containsKey(id) && weights.get(id) > 0) { long availableMemory = getAvailableMemory(entry.getValue()); if (availableMemory > currBestMem) { currBestMem = availableMemory; chosen = id; } } } if (chosen == null) { throw new FederationPolicyException( "Zero Active Subcluster with weight 1."); } return chosen.toId(); }  @Override void reinitialize(FederationPolicyInitializationContext policyContext); @Override SubClusterId getHomeSubcluster(
      ApplicationSubmissionContext appSubmissionContext,
      List<SubClusterId> blacklist);  }
@Test public void testLoadIsRespected() throws YarnException { SubClusterId chosen = ((FederationRouterPolicy) getPolicy()) .getHomeSubcluster(getApplicationSubmissionContext(), null); Assert.assertEquals("sc05", chosen.getId()); }
@Test public void testIfNoSubclustersWithWeightOne() { setPolicy(new LoadBasedRouterPolicy()); setPolicyInfo(new WeightedPolicyInfo()); Map<SubClusterIdInfo, Float> routerWeights = new HashMap<>(); Map<SubClusterIdInfo, Float> amrmWeights = new HashMap<>(); SubClusterIdInfo sc = new SubClusterIdInfo(String.format("sc%02d", 0)); SubClusterInfo federationSubClusterInfo = SubClusterInfo.newInstance( sc.toId(), null, null, null, null, -1, SubClusterState.SC_RUNNING, -1, generateClusterMetricsInfo(0)); getActiveSubclusters().clear(); getActiveSubclusters().put(sc.toId(), federationSubClusterInfo); routerWeights.put(sc, 0.0f); amrmWeights.put(sc, 0.0f); getPolicyInfo().setRouterPolicyWeights(routerWeights); getPolicyInfo().setAMRMPolicyWeights(amrmWeights); try { FederationPoliciesTestUtil.initializePolicyContext(getPolicy(), getPolicyInfo(), getActiveSubclusters()); ((FederationRouterPolicy) getPolicy()) .getHomeSubcluster(getApplicationSubmissionContext(), null); fail(); } catch (YarnException ex) { Assert.assertTrue( ex.getMessage().contains("Zero Active Subcluster with weight 1")); } }