public <T> Future<Iterable<Map.Entry<ByteString, Iterable<T>>>> multimapFetchAllFuture(
      boolean omitValues, ByteString encodedTag, String stateFamily, Coder<T> elemCoder) {
    StateTag<ByteString> stateTag =
        StateTag.<ByteString>of(Kind.MULTIMAP_ALL, encodedTag, stateFamily)
            .toBuilder()
            .setOmitValues(omitValues)
            .build();
    return valuesToPagingIterableFuture(stateTag, elemCoder, this.stateFuture(stateTag, elemCoder));
  }
@Test
  public void testReadMultimapAllEntries() throws Exception {
    Future<Iterable<Map.Entry<ByteString, Iterable<Integer>>>> future =
        underTest.multimapFetchAllFuture(false, 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)
                    .setFetchMaxBytes(WindmillStateReader.INITIAL_MAX_MULTIMAP_BYTES));

    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)
                            .addValues(intData(1))
                            .addValues(intData(2)))
                    .addEntries(
                        Windmill.TagMultimapEntry.newBuilder()
                            .setEntryName(STATE_MULTIMAP_KEY_2)
                            .addValues(intData(10))
                            .addValues(intData(20))));
    Mockito.when(mockWindmill.getStateData(COMPUTATION, expectedRequest.build()))
        .thenReturn(response.build());

    Iterable<Map.Entry<ByteString, Iterable<Integer>>> results = future.get();
    Mockito.verify(mockWindmill).getStateData(COMPUTATION, expectedRequest.build());
    int foundEntries = 0;
    for (Map.Entry<ByteString, Iterable<Integer>> entry : results) {
      if (entry.getKey().equals(STATE_MULTIMAP_KEY_1)) {
        foundEntries++;
        assertThat(entry.getValue(), Matchers.containsInAnyOrder(1, 2));
      } else {
        foundEntries++;
        assertEquals(STATE_MULTIMAP_KEY_2, entry.getKey());
        assertThat(entry.getValue(), Matchers.containsInAnyOrder(10, 20));
      }
    }
    assertEquals(2, foundEntries);
    Mockito.verifyNoMoreInteractions(mockWindmill);
    assertNoReader(future);
  }