@Override
  public synchronized boolean tryReturnRecordAt(
      boolean isAtSplitPoint, @Nullable ShufflePosition groupStart) {
    if (lastGroupStart == null && !isAtSplitPoint) {
      throw new IllegalStateException(
          String.format("The first group [at %s] must be at a split point", groupStart.toString()));
    }
    if (this.startPosition != null && groupStart.compareTo(this.startPosition) < 0) {
      throw new IllegalStateException(
          String.format(
              "Trying to return record at %s which is before the starting position at %s",
              groupStart, this.startPosition));
    }
    int comparedToLast = (lastGroupStart == null) ? 1 : groupStart.compareTo(this.lastGroupStart);
    if (comparedToLast < 0) {
      throw new IllegalStateException(
          String.format(
              "Trying to return group at %s which is before the last-returned group at %s",
              groupStart, this.lastGroupStart));
    }
    if (isAtSplitPoint) {
      splitPointsSeen++;

      if (comparedToLast == 0) {
        throw new IllegalStateException(
            String.format(
                "Trying to return a group at a split point with same position as the "
                    + "previous group: both at %s, last group was %s",
                groupStart,
                lastGroupWasAtSplitPoint ? "at a split point." : "not at a split point."));
      }
      if (stopPosition != null && groupStart.compareTo(stopPosition) >= 0) {
        return false;
      }
    } else {
      checkState(
          comparedToLast == 0,
          // This case is not a violation of general RangeTracker semantics, but it is
          // contrary to how GroupingShuffleReader in particular works. Hitting it would
          // mean it's behaving unexpectedly.
          "Trying to return a group not at a split point, but with a different position "
              + "than the previous group: last group was %s at %s, current at %s",
          lastGroupWasAtSplitPoint ? "a split point" : "a non-split point",
          lastGroupStart,
          groupStart);
    }
    this.lastGroupStart = groupStart;
    this.lastGroupWasAtSplitPoint = isAtSplitPoint;
    return true;
  }
@Test
  public void testTryReturnRecordBeforeStart() throws Exception {
    GroupingShuffleRangeTracker tracker =
        new GroupingShuffleRangeTracker(ofBytes(3, 0, 0), ofBytes(5, 0, 0));
    expected.expect(IllegalStateException.class);
    tracker.tryReturnRecordAt(true, ofBytes(1, 2, 3));
  }