@Override
  public synchronized boolean tryReturnRecordAt(boolean isAtSplitPoint, ByteKey recordStart) {
    if (done) {
      return false;
    }

    checkState(!(position == null && !isAtSplitPoint), "The first record must be at a split point");
    checkState(
        !(recordStart.compareTo(range.getStartKey()) < 0),
        "Trying to return record which is before the start key");
    checkState(
        !(position != null && recordStart.compareTo(position) < 0),
        "Trying to return record which is before the last-returned record");

    if (position == null) {
      LOG.info(
          "Adjusting range start from {} to {} as position of first returned record",
          range.getStartKey(),
          recordStart);
      range = range.withStartKey(recordStart);
    }
    position = recordStart;

    if (isAtSplitPoint) {
      if (!range.containsKey(recordStart)) {
        done = true;
        return false;
      }
      ++splitPointsSeen;
    }
    return true;
  }
@Test
  public void testTryReturnRecordAt() {
    ByteKeyRangeTracker tracker = ByteKeyRangeTracker.of(INITIAL_RANGE);

    // Should be able to emit at the same key twice, should that happen.
    // Should be able to emit within range (in order, but system guarantees won't try out of order).
    // Should not be able to emit past end of range.

    assertTrue(tracker.tryReturnRecordAt(true, INITIAL_START_KEY));
    assertTrue(tracker.tryReturnRecordAt(true, INITIAL_START_KEY));

    assertTrue(tracker.tryReturnRecordAt(true, INITIAL_MIDDLE_KEY));
    assertTrue(tracker.tryReturnRecordAt(true, INITIAL_MIDDLE_KEY));

    assertTrue(tracker.tryReturnRecordAt(true, BEFORE_END_KEY));

    assertFalse(tracker.tryReturnRecordAt(true, END_KEY)); // after end

    assertFalse(tracker.tryReturnRecordAt(true, BEFORE_END_KEY)); // false because done
  }