public <V> Future<Iterable<Map.Entry<ByteString, V>>> valuePrefixFuture(
      ByteString prefix, String stateFamily, Coder<V> valueCoder) {
    // First request has no continuation position.
    StateTag<ByteString> stateTag =
        StateTag.<ByteString>of(Kind.VALUE_PREFIX, prefix, stateFamily).toBuilder().build();
    return valuesToPagingIterableFuture(
        stateTag, valueCoder, this.stateFuture(stateTag, valueCoder));
  }
@Test
  public void testReadTagValuePrefix() throws Exception {
    Future<Iterable<Map.Entry<ByteString, Integer>>> future =
        underTest.valuePrefixFuture(STATE_KEY_PREFIX, STATE_FAMILY, INT_CODER);
    Mockito.verifyNoMoreInteractions(mockWindmill);

    Windmill.KeyedGetDataRequest.Builder expectedRequest =
        Windmill.KeyedGetDataRequest.newBuilder()
            .setKey(DATA_KEY)
            .setShardingKey(SHARDING_KEY)
            .setWorkToken(WORK_TOKEN)
            .setMaxBytes(WindmillStateReader.MAX_KEY_BYTES)
            .addTagValuePrefixesToFetch(
                Windmill.TagValuePrefixRequest.newBuilder()
                    .setTagPrefix(STATE_KEY_PREFIX)
                    .setStateFamily(STATE_FAMILY)
                    .setFetchMaxBytes(WindmillStateReader.MAX_TAG_VALUE_PREFIX_BYTES));

    Windmill.KeyedGetDataResponse.Builder response =
        Windmill.KeyedGetDataResponse.newBuilder()
            .setKey(DATA_KEY)
            .addTagValuePrefixes(
                Windmill.TagValuePrefixResponse.newBuilder()
                    .setTagPrefix(STATE_KEY_PREFIX)
                    .setStateFamily(STATE_FAMILY)
                    .addTagValues(
                        Windmill.TagValue.newBuilder()
                            .setTag(STATE_KEY_1)
                            .setStateFamily(STATE_FAMILY)
                            .setValue(intValue(8)))
                    .addTagValues(
                        Windmill.TagValue.newBuilder()
                            .setTag(STATE_KEY_2)
                            .setStateFamily(STATE_FAMILY)
                            .setValue(intValue(9))));

    Mockito.when(mockWindmill.getStateData(COMPUTATION, expectedRequest.build()))
        .thenReturn(response.build());

    Iterable<Map.Entry<ByteString, Integer>> result = future.get();
    Mockito.verify(mockWindmill).getStateData(COMPUTATION, expectedRequest.build());
    Mockito.verifyNoMoreInteractions(mockWindmill);

    assertThat(
        result,
        Matchers.containsInAnyOrder(
            new AbstractMap.SimpleEntry<>(STATE_KEY_1, 8),
            new AbstractMap.SimpleEntry<>(STATE_KEY_2, 9)));

    assertNoReader(future);
  }