@Override
  public Map<T, Double> getWeightedSubset(Map<T, Double> weightMap, DeterministicSubsettingMetadata metadata)
  {
    if (metadata != null)
    {
      List<T> points = new ArrayList<>(weightMap.keySet());
      Collections.sort(points);
      Collections.shuffle(points, new Random(_randomSeed));
      List<Double> weights = points.stream().map(weightMap::get).collect(Collectors.toList());
      double totalWeight = weights.stream().mapToDouble(Double::doubleValue).sum();
      if (totalWeight == 0)
      {
        return null;
      }

      Ring ring = new Ring(weights, totalWeight);

      double offset = metadata.getInstanceId() / (double) metadata.getTotalInstanceCount();
      double subsetSliceWidth = getSubsetSliceWidth(metadata.getTotalInstanceCount(), points.size());
      List<Integer> indices = ring.getIndices(offset, subsetSliceWidth);

      return indices.stream().collect(
          Collectors.toMap(points::get, i -> round(ring.getWeight(i, offset, subsetSliceWidth), WEIGHT_DECIMAL_PLACE)));
    }
    else
    {
      _log.warn("Cannot retrieve metadata required for D2 subsetting. Revert to use all available hosts.");
      return null;
    }
  }
@Test(dataProvider = "uniformWeightData")
  public void testDistributionWithUniformWeight(int clientNum, int hostNum, int minSubsetSize)
  {
    double[] weights = new double[hostNum];
    Arrays.fill(weights, 1D);
    Map<String, Double> pointsMap = constructPointsMap(weights);

    Map<String, Double> distributionMap = new HashMap<>();

    for (int i = 0; i < clientNum; i++)
    {
      _deterministicSubsettingStrategy = new DeterministicSubsettingStrategy<>("test", minSubsetSize);
      Map<String, Double> weightedSubset = _deterministicSubsettingStrategy.getWeightedSubset(pointsMap,
          new DeterministicSubsettingMetadata(i, clientNum, 0));
      assertTrue(weightedSubset.size() >= Math.min(minSubsetSize, hostNum));

      for (Map.Entry<String, Double> entry: weightedSubset.entrySet())
      {
        distributionMap.put(entry.getKey(), distributionMap.getOrDefault(entry.getKey(), 0D) + entry.getValue());
      }
    }

    double host0WeightSum = distributionMap.getOrDefault("host0", 0D);
    for (double weightSum: distributionMap.values())
    {
      assertEquals(weightSum, host0WeightSum, DELTA_DIFF);
    }
  }