public Iterable<TimestampedValue<T>> readRange(Instant minTimestamp, Instant limitTimestamp) {
    checkState(
        !isClosed,
        "OrderedList user state is no longer usable because it is closed for %s",
        requestTemplate.getStateKey());

    // Store pendingAdds whose sort key is in the query range and values are truncated by the
    // current size. The values (collections) of pendingAdds are kept, so that they will still be
    // accessible in pre-existing iterables even after:
    //   (1) a sort key is added to or removed from pendingAdds, or
    //   (2) a new value is added to an existing sort key
    ArrayList<PrefetchableIterable<TimestampedValue<T>>> pendingAddsInRange = new ArrayList<>();
    for (Entry<Instant, Collection<T>> kv :
        pendingAdds.subMap(minTimestamp, limitTimestamp).entrySet()) {
      pendingAddsInRange.add(
          PrefetchableIterables.limit(
              Iterables.transform(kv.getValue(), (v) -> TimestampedValue.of(v, kv.getKey())),
              kv.getValue().size()));
    }
    Iterable<TimestampedValue<T>> valuesInRange = Iterables.concat(pendingAddsInRange);

    if (!isCleared) {
      StateRequest.Builder getRequestBuilder = this.requestTemplate.toBuilder();
      getRequestBuilder
          .getStateKeyBuilder()
          .getOrderedListUserStateBuilder()
          .getRangeBuilder()
          .setStart(minTimestamp.getMillis())
          .setEnd(limitTimestamp.getMillis());

      // TODO: consider use cache here
      CachingStateIterable<TimestampedValue<T>> persistentValues =
          StateFetchingIterators.readAllAndDecodeStartingFrom(
              Caches.noop(),
              this.beamFnStateClient,
              getRequestBuilder.build(),
              this.timestampedValueCoder);

      // Make a snapshot of the current pendingRemoves and use them to filter persistent values.
      // The values of pendingRemoves are copied, so that they will still be accessible in
      // pre-existing iterables even after a sort key is removed.
      TreeRangeSet<Instant> pendingRemovesSnapshot = TreeRangeSet.create(pendingRemoves);
      Iterable<TimestampedValue<T>> persistentValuesAfterRemoval =
          Iterables.filter(
              persistentValues, v -> !pendingRemovesSnapshot.contains(v.getTimestamp()));

      return Iterables.mergeSorted(
          ImmutableList.of(persistentValuesAfterRemoval, valuesInRange),
          Comparator.comparing(TimestampedValue::getTimestamp));
    }

    return valuesInRange;
  }
@Test
  public void testReadRange() throws Exception {
    FakeBeamFnStateClient fakeClient =
        new FakeBeamFnStateClient(
            timestampedValueCoder,
            ImmutableMap.of(
                createOrderedListStateKey("A", 1), asList(A1, B1),
                createOrderedListStateKey("A", 4), Collections.singletonList(A4),
                createOrderedListStateKey("A", 2), Collections.singletonList(A2)));

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

    Iterable<TimestampedValue<String>> stateBeforeB2 =
        userState.readRange(Instant.ofEpochMilli(2), Instant.ofEpochMilli(4));
    assertArrayEquals(
        Collections.singletonList(A2).toArray(),
        Iterables.toArray(stateBeforeB2, TimestampedValue.class));

    // Add a new value to an existing sort key
    userState.add(B2);
    assertArrayEquals(
        Collections.singletonList(A2).toArray(),
        Iterables.toArray(stateBeforeB2, TimestampedValue.class));
    assertArrayEquals(
        asList(A2, B2).toArray(),
        Iterables.toArray(
            userState.readRange(Instant.ofEpochMilli(2), Instant.ofEpochMilli(4)),
            TimestampedValue.class));

    // Add a new value to a new sort key
    userState.add(A3);
    assertArrayEquals(
        Collections.singletonList(A2).toArray(),
        Iterables.toArray(stateBeforeB2, TimestampedValue.class));
    assertArrayEquals(
        asList(A2, B2, A3).toArray(),
        Iterables.toArray(
            userState.readRange(Instant.ofEpochMilli(2), Instant.ofEpochMilli(4)),
            TimestampedValue.class));

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