@Override
  public SplitResult<OffsetRange> trySplit(double fractionOfRemainder) {
    // If current tracking range is no longer growable, split it as a normal range.
    if (range.getTo() != Long.MAX_VALUE || range.getTo() == range.getFrom()) {
      return super.trySplit(fractionOfRemainder);
    }
    // If current range has been done, there is no more space to split.
    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);

    // Fetch the estimated end offset. If the estimated end is smaller than the next offset, use
    // the next offset as end.
    BigDecimal estimateRangeEnd =
        BigDecimal.valueOf(rangeEndEstimator.estimate())
            .max(cur.add(BigDecimal.ONE, MathContext.DECIMAL128));

    // Convert to BigDecimal in computation to prevent overflow, which may result in loss of
    // precision.
    // split = cur + max(1, (estimateRangeEnd - cur) * fractionOfRemainder)
    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);
  }
@Test
  public void testSplit() throws Exception {
    SimpleEstimator simpleEstimator = new SimpleEstimator();
    GrowableOffsetRangeTracker tracker = new GrowableOffsetRangeTracker(0L, simpleEstimator);
    assertTrue(tracker.tryClaim(0L));

    simpleEstimator.setEstimateRangeEnd(16L);
    // The split of infinite range results in one finite range and one infinite range.
    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());

    // After the first split, the tracker should track a finite range. Estimate offset should not
    // impact split.
    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();
  }