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 testNegativeCache() throws Exception {
    FakeBeamFnStateClient fakeClient = new FakeBeamFnStateClient(Collections.emptyMap());
    MultimapUserState<byte[], String> userState =
        new MultimapUserState<>(
            Caches.eternal(),
            fakeClient,
            "instructionId",
            createMultimapKeyStateKey(),
            ByteArrayCoder.of(),
            StringUtf8Coder.of());
    assertArrayEquals(new String[] {}, Iterables.toArray(userState.get(A1), String.class));
    assertArrayEquals(new String[] {}, Iterables.toArray(userState.get(A1), String.class));
    assertThat(fakeClient.getCallCount(), is(1));
  }