@Override
  public boolean tryClaim(StreamProgress streamProgress) {
    if (shouldStop) {
      return false;
    }
    // We perform copy instead of assignment because we want to ensure all references of
    // streamProgress gets updated.
    this.streamProgress = streamProgress;
    return true;
  }
@Test
  public void testTryClaim() {
    final StreamProgress streamProgress = new StreamProgress();
    final ReadChangeStreamPartitionProgressTracker tracker =
        new ReadChangeStreamPartitionProgressTracker(streamProgress);
    assertEquals(streamProgress, tracker.currentRestriction());

    ChangeStreamContinuationToken changeStreamContinuationToken =
        ChangeStreamContinuationToken.create(Range.ByteStringRange.create("a", "b"), "1234");
    final StreamProgress streamProgress2 =
        new StreamProgress(
            changeStreamContinuationToken, Instant.now(), BigDecimal.ONE, Instant.now(), false);
    assertTrue(tracker.tryClaim(streamProgress2));
    assertEquals(streamProgress2, tracker.currentRestriction());
    assertEquals(
        streamProgress2.getEstimatedLowWatermark(),
        tracker.currentRestriction().getEstimatedLowWatermark());

    assertNull(tracker.trySplit(0.5));
    assertEquals(streamProgress2, tracker.currentRestriction());
    assertEquals(
        streamProgress2.getEstimatedLowWatermark(),
        tracker.currentRestriction().getEstimatedLowWatermark());
    try {
      tracker.checkDone();
      fail("Should not reach here because checkDone should have thrown an exception");
    } catch (IllegalStateException e) {
      assertTrue("There's more work to be done. CheckDone threw an exception", true);
    }

    final SplitResult<StreamProgress> splitResult = SplitResult.of(null, streamProgress2);
    assertEquals(splitResult, tracker.trySplit(0));

    assertFalse(tracker.tryClaim(streamProgress2));
    // No exception thrown, it is done.
    tracker.checkDone();
  }