public PrefetchableIterable<V> get(K key) {
    checkState(
        !isClosed,
        "Multimap user state is no longer usable because it is closed for %s",
        keysStateRequest.getStateKey());

    Object structuralKey = mapKeyCoder.structuralValue(key);
    KV<K, List<V>> pendingAddValues = pendingAdds.get(structuralKey);

    PrefetchableIterable<V> pendingValues =
        pendingAddValues == null
            ? PrefetchableIterables.fromArray()
            : PrefetchableIterables.limit(
                pendingAddValues.getValue(), pendingAddValues.getValue().size());
    if (isCleared || pendingRemoves.containsKey(structuralKey)) {
      return pendingValues;
    }

    return PrefetchableIterables.concat(getPersistedValues(structuralKey, key), pendingValues);
  }
@Test
  public void testGetValuesPrefetch() throws Exception {
    FakeBeamFnStateClient fakeClient =
        new FakeBeamFnStateClient(
            ImmutableMap.of(
                createMultimapKeyStateKey(),
                KV.of(ByteArrayCoder.of(), singletonList(A1)),
                createMultimapValueStateKey(A1),
                KV.of(StringUtf8Coder.of(), asList("V1", "V2"))));
    MultimapUserState<byte[], String> userState =
        new MultimapUserState<>(
            Caches.eternal(),
            fakeClient,
            "instructionId",
            createMultimapKeyStateKey(),
            ByteArrayCoder.of(),
            StringUtf8Coder.of());

    PrefetchableIterable<String> values = userState.get(A1);
    assertEquals(0, fakeClient.getCallCount());
    values.prefetch();
    assertEquals(1, fakeClient.getCallCount());
    assertArrayEquals(new String[] {"V1", "V2"}, Iterables.toArray(values, String.class));
    assertEquals(1, fakeClient.getCallCount());
  }