public <T> Future<Iterable<T>> multimapFetchSingleEntryFuture(
      ByteString encodedKey, ByteString encodedTag, String stateFamily, Coder<T> elemCoder) {
    StateTag<ByteString> stateTag =
        StateTag.<ByteString>of(Kind.MULTIMAP_SINGLE_ENTRY, encodedTag, stateFamily)
            .toBuilder()
            .setMultimapKey(encodedKey)
            .build();
    return valuesToPagingIterableFuture(stateTag, elemCoder, this.stateFuture(stateTag, elemCoder));
  }
@Test
  public void testReadMultimapSingleEntryPaginated() throws Exception {
    Future<Iterable<Integer>> future =
        underTest.multimapFetchSingleEntryFuture(
            STATE_MULTIMAP_KEY_1, STATE_KEY_1, STATE_FAMILY, INT_CODER);
    Mockito.verifyNoMoreInteractions(mockWindmill);

    Windmill.KeyedGetDataRequest.Builder expectedRequest1 =
        Windmill.KeyedGetDataRequest.newBuilder()
            .setKey(DATA_KEY)
            .setShardingKey(SHARDING_KEY)
            .setWorkToken(WORK_TOKEN)
            .setMaxBytes(WindmillStateReader.MAX_KEY_BYTES)
            .addMultimapsToFetch(
                Windmill.TagMultimapFetchRequest.newBuilder()
                    .setTag(STATE_KEY_1)
                    .setStateFamily(STATE_FAMILY)
                    .setFetchEntryNamesOnly(false)
                    .addEntriesToFetch(
                        Windmill.TagMultimapEntry.newBuilder()
                            .setEntryName(STATE_MULTIMAP_KEY_1)
                            .setFetchMaxBytes(WindmillStateReader.INITIAL_MAX_MULTIMAP_BYTES)
                            .build()));

    Windmill.KeyedGetDataResponse.Builder response1 =
        Windmill.KeyedGetDataResponse.newBuilder()
            .setKey(DATA_KEY)
            .addTagMultimaps(
                Windmill.TagMultimapFetchResponse.newBuilder()
                    .setTag(STATE_KEY_1)
                    .setStateFamily(STATE_FAMILY)
                    .addEntries(
                        Windmill.TagMultimapEntry.newBuilder()
                            .setEntryName(STATE_MULTIMAP_KEY_1)
                            .addAllValues(Arrays.asList(intData(5), intData(6)))
                            .setContinuationPosition(500)));
    Windmill.KeyedGetDataRequest.Builder expectedRequest2 =
        Windmill.KeyedGetDataRequest.newBuilder()
            .setKey(DATA_KEY)
            .setShardingKey(SHARDING_KEY)
            .setWorkToken(WORK_TOKEN)
            .setMaxBytes(WindmillStateReader.MAX_CONTINUATION_KEY_BYTES)
            .addMultimapsToFetch(
                Windmill.TagMultimapFetchRequest.newBuilder()
                    .setTag(STATE_KEY_1)
                    .setStateFamily(STATE_FAMILY)
                    .setFetchEntryNamesOnly(false)
                    .addEntriesToFetch(
                        Windmill.TagMultimapEntry.newBuilder()
                            .setEntryName(STATE_MULTIMAP_KEY_1)
                            .setFetchMaxBytes(WindmillStateReader.CONTINUATION_MAX_MULTIMAP_BYTES)
                            .setRequestPosition(500)
                            .build()));

    Windmill.KeyedGetDataResponse.Builder response2 =
        Windmill.KeyedGetDataResponse.newBuilder()
            .setKey(DATA_KEY)
            .addTagMultimaps(
                Windmill.TagMultimapFetchResponse.newBuilder()
                    .setTag(STATE_KEY_1)
                    .setStateFamily(STATE_FAMILY)
                    .addEntries(
                        Windmill.TagMultimapEntry.newBuilder()
                            .setEntryName(STATE_MULTIMAP_KEY_1)
                            .addAllValues(Arrays.asList(intData(7), intData(8)))
                            .setContinuationPosition(800)
                            .setRequestPosition(500)));
    Windmill.KeyedGetDataRequest.Builder expectedRequest3 =
        Windmill.KeyedGetDataRequest.newBuilder()
            .setKey(DATA_KEY)
            .setShardingKey(SHARDING_KEY)
            .setWorkToken(WORK_TOKEN)
            .setMaxBytes(WindmillStateReader.MAX_CONTINUATION_KEY_BYTES)
            .addMultimapsToFetch(
                Windmill.TagMultimapFetchRequest.newBuilder()
                    .setTag(STATE_KEY_1)
                    .setStateFamily(STATE_FAMILY)
                    .setFetchEntryNamesOnly(false)
                    .addEntriesToFetch(
                        Windmill.TagMultimapEntry.newBuilder()
                            .setEntryName(STATE_MULTIMAP_KEY_1)
                            .setFetchMaxBytes(WindmillStateReader.CONTINUATION_MAX_MULTIMAP_BYTES)
                            .setRequestPosition(800)
                            .build()));

    Windmill.KeyedGetDataResponse.Builder response3 =
        Windmill.KeyedGetDataResponse.newBuilder()
            .setKey(DATA_KEY)
            .addTagMultimaps(
                Windmill.TagMultimapFetchResponse.newBuilder()
                    .setTag(STATE_KEY_1)
                    .setStateFamily(STATE_FAMILY)
                    .addEntries(
                        Windmill.TagMultimapEntry.newBuilder()
                            .setEntryName(STATE_MULTIMAP_KEY_1)
                            .addAllValues(Arrays.asList(intData(9), intData(10)))
                            .setRequestPosition(800)));
    Mockito.when(mockWindmill.getStateData(COMPUTATION, expectedRequest1.build()))
        .thenReturn(response1.build());
    Mockito.when(mockWindmill.getStateData(COMPUTATION, expectedRequest2.build()))
        .thenReturn(response2.build());
    Mockito.when(mockWindmill.getStateData(COMPUTATION, expectedRequest3.build()))
        .thenReturn(response3.build());

    Iterable<Integer> results = future.get();

    assertThat(results, Matchers.contains(5, 6, 7, 8, 9, 10));

    Mockito.verify(mockWindmill).getStateData(COMPUTATION, expectedRequest1.build());
    Mockito.verify(mockWindmill).getStateData(COMPUTATION, expectedRequest2.build());
    Mockito.verify(mockWindmill).getStateData(COMPUTATION, expectedRequest3.build());
    Mockito.verifyNoMoreInteractions(mockWindmill);
    // NOTE: The future will still contain a reference to the underlying reader, thus not calling
    // assertNoReader(future).
  }