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()) { 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(); } } 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);   }
@Test public void coalesceTokenRangesTests() throws ReaperException { SegmentGenerator sg = new SegmentGenerator(BigInteger.valueOf(1), BigInteger.valueOf(1600)); Map<List<String>, List<String>> rangeToEndpoint = Maps.newHashMap(); rangeToEndpoint.put(Arrays.asList("1", "200"), Arrays.asList("node1", "node2", "node3")); rangeToEndpoint.put(Arrays.asList("200", "400"), Arrays.asList("node1", "node2", "node3")); rangeToEndpoint.put(Arrays.asList("400", "600"), Arrays.asList("node1")); rangeToEndpoint.put(Arrays.asList("600", "800"), Arrays.asList("node1", "node2")); rangeToEndpoint.put(Arrays.asList("800", "1000"), Arrays.asList("node1", "node2")); rangeToEndpoint.put(Arrays.asList("1100", "1200"), Arrays.asList("node2", "node3", "node1")); Map<List<String>, List<RingRange>> replicasToRangeMap = RepairRunService.buildReplicasToRangeMap(rangeToEndpoint); List<Segment> segments = sg.coalesceTokenRanges(BigInteger.valueOf(1200), replicasToRangeMap); assertEquals(3, segments.size()); }
@Test public void coalesceTokenRangesTooFewTokensPerSegmentTests() throws ReaperException { SegmentGenerator sg = new SegmentGenerator(BigInteger.valueOf(1), BigInteger.valueOf(1600)); Map<List<String>, List<String>> rangeToEndpoint = Maps.newHashMap(); rangeToEndpoint.put(Arrays.asList("1", "200"), Arrays.asList("node1", "node2", "node3")); rangeToEndpoint.put(Arrays.asList("200", "400"), Arrays.asList("node1", "node2", "node3")); rangeToEndpoint.put(Arrays.asList("400", "600"), Arrays.asList("node1", "node2", "node3")); rangeToEndpoint.put(Arrays.asList("600", "800"), Arrays.asList("node1", "node2", "node3")); rangeToEndpoint.put(Arrays.asList("800", "1000"), Arrays.asList("node1", "node2", "node3")); rangeToEndpoint.put(Arrays.asList("1100", "1200"), Arrays.asList("node2", "node3", "node1")); rangeToEndpoint.put(Arrays.asList("1200", "1300"), Arrays.asList("node2", "node3", "node1")); rangeToEndpoint.put(Arrays.asList("1300", "1400"), Arrays.asList("node2", "node3", "node1")); rangeToEndpoint.put(Arrays.asList("1400", "1500"), Arrays.asList("node2", "node3", "node1")); Map<List<String>, List<RingRange>> replicasToRangeMap = RepairRunService.buildReplicasToRangeMap(rangeToEndpoint); List<Segment> segments = sg.coalesceTokenRanges(BigInteger.valueOf(1), replicasToRangeMap); assertEquals(9, segments.size()); }
@Test public void coalesceTokenRangesBy200TokensPerSegmentTests() throws ReaperException { SegmentGenerator sg = new SegmentGenerator(BigInteger.valueOf(1), BigInteger.valueOf(1500)); Map<List<String>, List<String>> rangeToEndpoint = Maps.newHashMap(); rangeToEndpoint.put(Arrays.asList("1", "200"), Arrays.asList("node1", "node2", "node3")); rangeToEndpoint.put(Arrays.asList("200", "400"), Arrays.asList("node1", "node2", "node3")); rangeToEndpoint.put(Arrays.asList("400", "600"), Arrays.asList("node1", "node2", "node3")); rangeToEndpoint.put(Arrays.asList("600", "800"), Arrays.asList("node1", "node2", "node3")); rangeToEndpoint.put(Arrays.asList("800", "1000"), Arrays.asList("node1", "node2", "node3")); rangeToEndpoint.put(Arrays.asList("1100", "1200"), Arrays.asList("node1", "node2", "node3")); rangeToEndpoint.put(Arrays.asList("1200", "1300"), Arrays.asList("node1", "node2", "node3")); rangeToEndpoint.put(Arrays.asList("1300", "1400"), Arrays.asList("node1", "node2", "node3")); rangeToEndpoint.put(Arrays.asList("1400", "1500"), Arrays.asList("node1", "node2", "node3")); Map<List<String>, List<RingRange>> replicasToRangeMap = RepairRunService.buildReplicasToRangeMap(rangeToEndpoint); List<Segment> segments = sg.coalesceTokenRanges(BigInteger.valueOf(200), replicasToRangeMap); assertEquals(7, segments.size()); }