public void clearRange(Instant minTimestamp, Instant limitTimestamp) {
    checkState(
        !isClosed,
        "OrderedList user state is no longer usable because it is closed for %s",
        requestTemplate.getStateKey());

    // Remove items (in a collection) in the specific range from pendingAdds.
    // The old values of the removed sub map are kept, so that they will still be accessible in
    // pre-existing iterables even after the sort key is cleared.
    pendingAdds.subMap(minTimestamp, limitTimestamp).clear();
    if (!isCleared) {
      pendingRemoves.add(
          Range.range(minTimestamp, BoundType.CLOSED, limitTimestamp, BoundType.OPEN));
    }
  }
@Test
  public void testClearRange() throws Exception {
    FakeBeamFnStateClient fakeClient =
        new FakeBeamFnStateClient(
            timestampedValueCoder,
            ImmutableMap.of(
                createOrderedListStateKey("A", 1),
                asList(A1, B1),
                createOrderedListStateKey("A", 4),
                Collections.singletonList(A4),
                createOrderedListStateKey("A", 2),
                asList(A2, B2),
                createOrderedListStateKey("A", 3),
                Collections.singletonList(A3)));

    OrderedListUserState<String> userState =
        new OrderedListUserState<>(
            Caches.noop(),
            fakeClient,
            "instructionId",
            createOrderedListStateKey("A"),
            StringUtf8Coder.of());

    Iterable<TimestampedValue<String>> initStateFrom2To3 =
        userState.readRange(Instant.ofEpochMilli(2), Instant.ofEpochMilli(4));

    // clear range below the current timestamp range
    userState.clearRange(Instant.ofEpochMilli(-1), Instant.ofEpochMilli(0));
    assertArrayEquals(
        asList(A2, B2, A3).toArray(), Iterables.toArray(initStateFrom2To3, TimestampedValue.class));
    assertArrayEquals(
        asList(A1, B1, A2, B2, A3, A4).toArray(),
        Iterables.toArray(userState.read(), TimestampedValue.class));

    // clear range above the current timestamp range
    userState.clearRange(Instant.ofEpochMilli(5), Instant.ofEpochMilli(10));
    assertArrayEquals(
        asList(A2, B2, A3).toArray(), Iterables.toArray(initStateFrom2To3, TimestampedValue.class));
    assertArrayEquals(
        asList(A1, B1, A2, B2, A3, A4).toArray(),
        Iterables.toArray(userState.read(), TimestampedValue.class));

    // clear range that falls inside the current timestamp range
    userState.clearRange(Instant.ofEpochMilli(2), Instant.ofEpochMilli(4));
    assertArrayEquals(
        asList(A2, B2, A3).toArray(), Iterables.toArray(initStateFrom2To3, TimestampedValue.class));
    assertArrayEquals(
        asList(A1, B1, A4).toArray(), Iterables.toArray(userState.read(), TimestampedValue.class));

    // clear range that partially covers the current timestamp range
    userState.clearRange(Instant.ofEpochMilli(3), Instant.ofEpochMilli(5));
    assertArrayEquals(
        asList(A2, B2, A3).toArray(), Iterables.toArray(initStateFrom2To3, TimestampedValue.class));
    assertArrayEquals(
        asList(A1, B1).toArray(), Iterables.toArray(userState.read(), TimestampedValue.class));

    // clear range that fully covers the current timestamp range
    userState.clearRange(Instant.ofEpochMilli(-1), Instant.ofEpochMilli(10));
    assertArrayEquals(
        asList(A2, B2, A3).toArray(), Iterables.toArray(initStateFrom2To3, TimestampedValue.class));
    assertThat(userState.read(), is(emptyIterable()));

    userState.asyncClose();
    assertThrows(
        IllegalStateException.class,
        () -> userState.clearRange(Instant.ofEpochMilli(1), Instant.ofEpochMilli(2)));
  }