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; } } } 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()); }