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 testReadMultimapSingleEntry() 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 expectedRequest =
        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 response =
        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)))));
    Mockito.when(mockWindmill.getStateData(COMPUTATION, expectedRequest.build()))
        .thenReturn(response.build());

    Iterable<Integer> results = future.get();
    Mockito.verify(mockWindmill).getStateData(COMPUTATION, expectedRequest.build());

    assertThat(results, Matchers.containsInAnyOrder(5, 6));
    Mockito.verifyNoMoreInteractions(mockWindmill);
    assertNoReader(future);
  }