class SegmentGenerator {

  @VisibleForTesting
  List<Segment> coalesceTokenRanges(
      BigInteger targetSegmentSize, Map<List<String>, List<RingRange>> replicasToRange) {

    List<Segment> coalescedRepairSegments = Lists.newArrayList();
    List<RingRange> tokenRangesForCurrentSegment = Lists.newArrayList();
    BigInteger tokenCount = BigInteger.ZERO;

    for (Entry<List<String>, List<RingRange>> tokenRangesByReplica : replicasToRange.entrySet()) {
      LOG.info("Coalescing segments for nodes {}", tokenRangesByReplica.getKey());
      for (RingRange tokenRange : tokenRangesByReplica.getValue()) {
        if (tokenRange.span(rangeSize).add(tokenCount).compareTo(targetSegmentSize) > 0
            && !tokenRangesForCurrentSegment.isEmpty()) {
          // enough tokens in that segment
          LOG.info(
              "Got enough tokens for one segment ({}) : {}",
              tokenCount,
              tokenRangesForCurrentSegment);
          coalescedRepairSegments.add(
              Segment.builder().withTokenRanges(tokenRangesForCurrentSegment).build());
          tokenRangesForCurrentSegment = Lists.newArrayList();
          tokenCount = BigInteger.ZERO;
        }

        tokenCount = tokenCount.add(tokenRange.span(rangeSize));
        tokenRangesForCurrentSegment.add(tokenRange);

      }

      if (!tokenRangesForCurrentSegment.isEmpty()) {
        coalescedRepairSegments.add(
            Segment.builder().withTokenRanges(tokenRangesForCurrentSegment).build());
        tokenRangesForCurrentSegment = Lists.newArrayList();
      }
    }


    // Check that we haven't left any token range outside of the resulting segments
    Preconditions.checkState(
        allTokensHaveBeenCoalesced(coalescedRepairSegments, replicasToRange),
        "Number of coalesced tokens doesn't match with the total number of tokens");

    return coalescedRepairSegments;
  }

    SegmentGenerator(String partitioner);
    SegmentGenerator(BigInteger rangeMin, BigInteger rangeMax);

}