GrowableOffsetRangeTracker extends OffsetRangeTracker { @Override public SplitResult<OffsetRange> trySplit(double fractionOfRemainder) { if (range.getTo() != Long.MAX_VALUE || range.getTo() == range.getFrom()) { return super.trySplit(fractionOfRemainder); } if (lastAttemptedOffset != null && lastAttemptedOffset == Long.MAX_VALUE) { return null; } BigDecimal cur = (lastAttemptedOffset == null) ? BigDecimal.valueOf(range.getFrom()).subtract(BigDecimal.ONE, MathContext.DECIMAL128) : BigDecimal.valueOf(lastAttemptedOffset); BigDecimal estimateRangeEnd = BigDecimal.valueOf(rangeEndEstimator.estimate()) .max(cur.add(BigDecimal.ONE, MathContext.DECIMAL128)); BigDecimal splitPos = cur.add( estimateRangeEnd .subtract(cur, MathContext.DECIMAL128) .multiply(BigDecimal.valueOf(fractionOfRemainder), MathContext.DECIMAL128) .max(BigDecimal.ONE), MathContext.DECIMAL128); long split = splitPos.longValue(); if (split > estimateRangeEnd.longValue()) { return null; } OffsetRange res = new OffsetRange(split, range.getTo()); this.range = new OffsetRange(range.getFrom(), split); return SplitResult.of(range, res); } GrowableOffsetRangeTracker(long start, RangeEndEstimator rangeEndEstimator); @Override SplitResult<OffsetRange> trySplit(double fractionOfRemainder); @Override Progress getProgress(); @Override IsBounded isBounded();  }
@Test public void testCheckpointBeforeStart() throws Exception { SimpleEstimator simpleEstimator = new SimpleEstimator(); GrowableOffsetRangeTracker tracker = new GrowableOffsetRangeTracker(0L, simpleEstimator); simpleEstimator.setEstimateRangeEnd(10); SplitResult res = tracker.trySplit(0); tracker.checkDone(); assertEquals(new OffsetRange(0, 0), res.getPrimary()); assertEquals(new OffsetRange(0, 0), tracker.currentRestriction()); assertEquals(new OffsetRange(0, Long.MAX_VALUE), res.getResidual()); }
@Test public void testCheckpointJustStarted() throws Exception { SimpleEstimator simpleEstimator = new SimpleEstimator(); GrowableOffsetRangeTracker tracker = new GrowableOffsetRangeTracker(0L, simpleEstimator); assertTrue(tracker.tryClaim(5L)); simpleEstimator.setEstimateRangeEnd(0L); SplitResult res = tracker.trySplit(0); tracker.checkDone(); assertEquals(new OffsetRange(0, 6), res.getPrimary()); assertEquals(new OffsetRange(0, 6), tracker.currentRestriction()); assertEquals(new OffsetRange(6, Long.MAX_VALUE), res.getResidual()); tracker = new GrowableOffsetRangeTracker(0L, simpleEstimator); assertTrue(tracker.tryClaim(5L)); simpleEstimator.setEstimateRangeEnd(20L); res = tracker.trySplit(0); tracker.checkDone(); assertEquals(new OffsetRange(0, 6), res.getPrimary()); assertEquals(new OffsetRange(6, Long.MAX_VALUE), res.getResidual()); }
@Test public void testCheckpointAfterAllProcessed() throws Exception { SimpleEstimator simpleEstimator = new SimpleEstimator(); GrowableOffsetRangeTracker tracker = new GrowableOffsetRangeTracker(0L, simpleEstimator); assertFalse(tracker.tryClaim(Long.MAX_VALUE)); tracker.checkDone(); assertNull(tracker.trySplit(0)); }
@Test public void testCheckpointAtEmptyRange() throws Exception { GrowableOffsetRangeTracker tracker = new GrowableOffsetRangeTracker(Long.MAX_VALUE, new SimpleEstimator()); tracker.checkDone(); assertNull(tracker.trySplit(0)); }
@Test public void testSplit() throws Exception { SimpleEstimator simpleEstimator = new SimpleEstimator(); GrowableOffsetRangeTracker tracker = new GrowableOffsetRangeTracker(0L, simpleEstimator); assertTrue(tracker.tryClaim(0L)); simpleEstimator.setEstimateRangeEnd(16L); SplitResult res = tracker.trySplit(0.5); assertEquals(new OffsetRange(0, 8), res.getPrimary()); assertEquals(new OffsetRange(0, 8), tracker.currentRestriction()); assertEquals(new OffsetRange(8, Long.MAX_VALUE), res.getResidual()); simpleEstimator.setEstimateRangeEnd(12L); res = tracker.trySplit(0.5); assertEquals(new OffsetRange(0, 4), res.getPrimary()); assertEquals(new OffsetRange(0, 4), tracker.currentRestriction()); assertEquals(new OffsetRange(4, 8), res.getResidual()); assertFalse(tracker.tryClaim(4L)); tracker.checkDone(); }
@Test public void testSplitWithMaxEstimateRangeEnd() throws Exception { SimpleEstimator simpleEstimator = new SimpleEstimator(); GrowableOffsetRangeTracker tracker = new GrowableOffsetRangeTracker(0L, simpleEstimator); assertTrue(tracker.tryClaim(1L)); simpleEstimator.setEstimateRangeEnd(Long.MAX_VALUE); SplitResult res = tracker.trySplit(0.5); long expectedEnd = 1L + (Long.MAX_VALUE - 1L) / 2; assertEquals(new OffsetRange(0L, expectedEnd), res.getPrimary()); assertEquals(new OffsetRange(expectedEnd, Long.MAX_VALUE), res.getResidual()); }
@Test public void testSmallRangeWithLargeValue() throws Exception { SimpleEstimator simpleEstimator = new SimpleEstimator(); GrowableOffsetRangeTracker tracker = new GrowableOffsetRangeTracker(123456789012345677L, simpleEstimator); assertTrue(tracker.tryClaim(123456789012345677L)); simpleEstimator.setEstimateRangeEnd(123456789012345679L); SplitResult res = tracker.trySplit(0.5); assertEquals(new OffsetRange(123456789012345677L, 123456789012345678L), res.getPrimary()); assertEquals(new OffsetRange(123456789012345678L, Long.MAX_VALUE), res.getResidual()); tracker = new GrowableOffsetRangeTracker(123456789012345681L, simpleEstimator); assertTrue(tracker.tryClaim(123456789012345681L)); simpleEstimator.setEstimateRangeEnd(123456789012345683L); res = tracker.trySplit(0.5); assertEquals(new OffsetRange(123456789012345681L, 123456789012345682L), res.getPrimary()); assertEquals(new OffsetRange(123456789012345682L, Long.MAX_VALUE), res.getResidual()); }