@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 = "differentWeightsData")
  public void testDistributionWithDifferentWeights(int clientNum, double[] weights, int minSubsetSize)
  {
    Map<String, Double> pointsMap = constructPointsMap(weights);
    Map<String, Double> distributionMap = new HashMap<>();
    double minSubsetWeight = minSubsetSize / (double) weights.length;
    double totalHostWeights = Arrays.stream(weights).sum();

    for (int i = 0; i < clientNum; i++)
    {
      _deterministicSubsettingStrategy = new DeterministicSubsettingStrategy<>("test", minSubsetSize);
      Map<String, Double> weightedSubset = _deterministicSubsettingStrategy.getWeightedSubset(pointsMap,
          new DeterministicSubsettingMetadata(i, clientNum, 0));
      double totalWeights = 0;
      for (Map.Entry<String, Double> entry: weightedSubset.entrySet())
      {
        String hostName = entry.getKey();
        double weight = entry.getValue();
        distributionMap.put(hostName,
            distributionMap.getOrDefault(hostName, 0D) + weight * pointsMap.get(hostName));
        totalWeights += weights[Integer.parseInt(hostName.substring("test".length()))] / totalHostWeights * weight;
      }
      assertTrue(totalWeights + DELTA_DIFF >= Math.min(minSubsetWeight, 1D));
    }

    double totalWeights = distributionMap.values().stream().mapToDouble(Double::doubleValue).sum();
    for (Map.Entry<String, Double> entry: distributionMap.entrySet())
    {
      String hostName = entry.getKey();
      double hostWeight = weights[Integer.parseInt(hostName.substring("test".length()))];
      assertEquals(entry.getValue() / totalWeights, hostWeight / totalHostWeights, DELTA_DIFF);
    }
  }