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 testReadMultimapMultipleEntriesWithPagination() throws Exception {
    Future<Iterable<Integer>> future1 =
        underTest.multimapFetchSingleEntryFuture(
            STATE_MULTIMAP_KEY_1, STATE_KEY_1, STATE_FAMILY, INT_CODER);
    Future<Iterable<Integer>> future2 =
        underTest.multimapFetchSingleEntryFuture(
            STATE_MULTIMAP_KEY_2, 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())
                    .addEntriesToFetch(
                        Windmill.TagMultimapEntry.newBuilder()
                            .setEntryName(STATE_MULTIMAP_KEY_2)
                            .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(800))
                    .addEntries(
                        Windmill.TagMultimapEntry.newBuilder()
                            .setEntryName(STATE_MULTIMAP_KEY_2)
                            .addAllValues(Arrays.asList(intData(15), intData(16)))));
    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(800)
                            .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)))
                            .setRequestPosition(800)));
    when(mockWindmill.getStateData(ArgumentMatchers.eq(COMPUTATION), ArgumentMatchers.any()))
        .thenReturn(response1.build())
        .thenReturn(response2.build());

    Iterable<Integer> results1 = future1.get();
    Iterable<Integer> results2 = future2.get();

    assertThat(results1, Matchers.containsInAnyOrder(5, 6, 7, 8));
    assertThat(results2, Matchers.containsInAnyOrder(15, 16));

    final ArgumentCaptor<Windmill.KeyedGetDataRequest> requestCaptor =
        ArgumentCaptor.forClass(Windmill.KeyedGetDataRequest.class);
    Mockito.verify(mockWindmill, times(2))
        .getStateData(ArgumentMatchers.eq(COMPUTATION), requestCaptor.capture());
    assertMultimapFetchRequestEqual(
        expectedRequest1.build().getMultimapsToFetch(0),
        requestCaptor.getAllValues().get(0).getMultimapsToFetch(0));
    assertMultimapFetchRequestEqual(
        expectedRequest2.build().getMultimapsToFetch(0),
        requestCaptor.getAllValues().get(1).getMultimapsToFetch(0));
    Mockito.verifyNoMoreInteractions(mockWindmill);
    // NOTE: The future will still contain a reference to the underlying reader, thus not calling
    // assertNoReader(future).
  }