@Override
  public <T extends State> T state(StateNamespace namespace, StateTag<T> address) {
    return workItemState.get(namespace, address, StateContexts.nullContext());
  }
@Test
  public void testOrderedListInterleavedLocalAddClearReadRange() {
    Future<Map<Range<Instant>, RangeSet<Long>>> orderedListFuture = Futures.immediateFuture(null);
    Future<Map<Range<Instant>, RangeSet<Instant>>> deletionsFuture = Futures.immediateFuture(null);
    when(mockReader.valueFuture(
            systemKey(NAMESPACE, "orderedList" + IdTracker.IDS_AVAILABLE_STR),
            STATE_FAMILY,
            IdTracker.IDS_AVAILABLE_CODER))
        .thenReturn(orderedListFuture);
    when(mockReader.valueFuture(
            systemKey(NAMESPACE, "orderedList" + IdTracker.DELETIONS_STR),
            STATE_FAMILY,
            IdTracker.SUBRANGE_DELETIONS_CODER))
        .thenReturn(deletionsFuture);

    SettableFuture<Iterable<TimestampedValue<String>>> fromStorage = SettableFuture.create();

    Range<Long> readSubrange = Range.closedOpen(1 * 1000000L, 8 * 1000000L);
    when(mockReader.orderedListFuture(
            readSubrange, key(NAMESPACE, "orderedList"), STATE_FAMILY, StringUtf8Coder.of()))
        .thenReturn(fromStorage);

    StateTag<OrderedListState<String>> addr =
        StateTags.orderedList("orderedList", StringUtf8Coder.of());
    OrderedListState<String> orderedListState = underTest.state(NAMESPACE, addr);

    orderedListState.add(TimestampedValue.of("1", Instant.ofEpochSecond(1)));
    orderedListState.add(TimestampedValue.of("2", Instant.ofEpochSecond(2)));
    orderedListState.add(TimestampedValue.of("3", Instant.ofEpochSecond(3)));
    orderedListState.add(TimestampedValue.of("4", Instant.ofEpochSecond(4)));

    orderedListState.clearRange(Instant.ofEpochSecond(1), Instant.ofEpochSecond(4));

    orderedListState.add(TimestampedValue.of("5", Instant.ofEpochSecond(5)));
    orderedListState.add(TimestampedValue.of("6", Instant.ofEpochSecond(6)));

    orderedListState.add(TimestampedValue.of("3_again", Instant.ofEpochSecond(3)));

    orderedListState.add(TimestampedValue.of("7", Instant.ofEpochSecond(7)));
    orderedListState.add(TimestampedValue.of("8", Instant.ofEpochSecond(8)));

    fromStorage.set(ImmutableList.<TimestampedValue<String>>of());

    TimestampedValue[] expected =
        Iterables.toArray(
            ImmutableList.of(
                TimestampedValue.of("3_again", Instant.ofEpochSecond(3)),
                TimestampedValue.of("4", Instant.ofEpochSecond(4)),
                TimestampedValue.of("5", Instant.ofEpochSecond(5)),
                TimestampedValue.of("6", Instant.ofEpochSecond(6)),
                TimestampedValue.of("7", Instant.ofEpochSecond(7))),
            TimestampedValue.class);

    TimestampedValue[] read =
        Iterables.toArray(
            orderedListState.readRange(Instant.ofEpochSecond(1), Instant.ofEpochSecond(8)),
            TimestampedValue.class);
    assertArrayEquals(expected, read);
  }