class LoadBasedRouterPolicy extends AbstractRouterPolicy {

  @Override
  public SubClusterId getHomeSubcluster(
      ApplicationSubmissionContext appSubmissionContext,
      List<SubClusterId> blacklist) throws YarnException {

    // null checks and default-queue behavior
    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();
  }

}