@VisibleForTesting
  protected Map<SubClusterIdInfo, Float> getTargetWeights(
      Map<SubClusterId, ClusterMetricsInfo> clusterMetrics) {
    Map<SubClusterIdInfo, Float> weights = GPGUtils.createUniformWeights(clusterMetrics.keySet());

    List<SubClusterId> scs = new ArrayList<>(clusterMetrics.keySet());
    // Sort the sub clusters into descending order based on pending load
    scs.sort(new SortByDescendingLoad(clusterMetrics));

    // Keep the top N loaded sub clusters
    scs = scs.subList(0, Math.min(maxEdit, scs.size()));

    for (SubClusterId sc : scs) {
      LOG.info("Updating weight for sub cluster {}", sc.toString());
      int pending = clusterMetrics.get(sc).getAppsPending();
      if (pending <= minPending) {
        LOG.info("Load ({}) is lower than minimum ({}), skipping", pending, minPending);
      } else if (pending < maxPending) {
        // The different scaling strategies should all map values from the
        // range min_pending+1 to max_pending to the range min_weight to 1.0f
        // so we pre-process and simplify the domain to some value [1, MAX-MIN)
        int val = pending - minPending;
        int maxVal = maxPending - minPending;

        // Scale the weights to respect the config minimum
        float weight = getWeightByScaling(maxVal, val);
        weight = weight * (1.0f - minWeight);
        weight += minWeight;
        weights.put(new SubClusterIdInfo(sc), weight);
        LOG.info("Load ({}) is within maximum ({}), setting weights via {} "
            + "scale to {}", pending, maxPending, scaling, weight);
      } else {
        weights.put(new SubClusterIdInfo(sc), minWeight);
        LOG.info("Load ({}) exceeded maximum ({}), setting weight to minimum: {}",
            pending, maxPending, minWeight);
      }
    }
    validateWeights(weights);
    return weights;
  }
@Test
  public void testSimpleTargetWeights() {
    weights = policyGenerator.getTargetWeights(clusterMetricsInfos);
    assertEquals(weights.size(), 3);
    assertEquals(1.0, getWeight(0), DELTA);
    assertEquals(1.0, getWeight(1), DELTA);
    assertEquals(1.0, getWeight(2), DELTA);
  }