@Override
  public <T extends State> T state(StateNamespace namespace, StateTag<T> address) {
    return workItemState.get(namespace, address, StateContexts.nullContext());
  }
@Test
  public void testOrderedListAddBeforeRangeRead() throws Exception {
    StateTag<OrderedListState<String>> addr =
        StateTags.orderedList("orderedList", StringUtf8Coder.of());
    OrderedListState<String> orderedList = underTest.state(NAMESPACE, addr);

    SettableFuture<Iterable<TimestampedValue<String>>> future = SettableFuture.create();
    Range<Long> readSubrange = Range.closedOpen(70 * 1000L, 100 * 1000L);
    when(mockReader.orderedListFuture(
            readSubrange, key(NAMESPACE, "orderedList"), STATE_FAMILY, StringUtf8Coder.of()))
        .thenReturn(future);

    orderedList.readRangeLater(Instant.ofEpochMilli(70), Instant.ofEpochMilli(100));

    final TimestampedValue<String> helloValue =
        TimestampedValue.of("hello", Instant.ofEpochMilli(100));
    final TimestampedValue<String> worldValue =
        TimestampedValue.of("world", Instant.ofEpochMilli(75));
    final TimestampedValue<String> goodbyeValue =
        TimestampedValue.of("goodbye", Instant.ofEpochMilli(50));

    orderedList.add(helloValue);
    waitAndSet(future, Collections.singletonList(worldValue), 200);
    orderedList.add(goodbyeValue);

    assertThat(
        orderedList.readRange(Instant.ofEpochMilli(70), Instant.ofEpochMilli(100)),
        Matchers.contains(worldValue));
  }