@Override
  public Progress getProgress() {
    BigDecimal end;
    if (range.getTo().compareTo(Timestamp.MAX_VALUE) == 0) {
      // When the given end timestamp equals to Timestamp.MAX_VALUE, this means that
      // the end timestamp is not specified which should be a streaming job. So we
      // use now() as the end timestamp.
      end = BigDecimal.valueOf(timeSupplier.get().getSeconds());
    } else {
      end = BigDecimal.valueOf(range.getTo().getSeconds());
    }
    BigDecimal current;
    if (lastClaimedPosition == null) {
      current = BigDecimal.valueOf(range.getFrom().getSeconds());
    } else {
      current = BigDecimal.valueOf(lastClaimedPosition.getSeconds());
    }
    // The remaining work must be greater than 0. Otherwise, it will cause an issue
    // that the watermark does not advance.
    final BigDecimal workRemaining = end.subtract(current).max(BigDecimal.ONE);

    LOG.debug(
        "Reported progress current: {}, end: {}, workRemaining: {}",
        current.doubleValue(),
        end.doubleValue(),
        workRemaining.doubleValue());

    return Progress.from(current.doubleValue(), workRemaining.doubleValue());
  }
@Test
  public void testGetProgressReturnsWorkRemainingAsWholeRangeWhenNoClaimWasAttempted() {
    final Timestamp from = Timestamp.ofTimeSecondsAndNanos(0, 0);
    final Timestamp to = Timestamp.now();
    final TimestampRange range = TimestampRange.of(from, to);
    final TimestampRangeTracker tracker = new TimestampRangeTracker(range);

    final Progress progress = tracker.getProgress();
    assertEquals(0D, progress.getWorkCompleted(), DELTA);
    assertEquals(to.getSeconds(), progress.getWorkRemaining(), DELTA);
  }