public void put(K key, V value) {
    checkState(
        !isClosed,
        "Multimap user state is no longer usable because it is closed for %s",
        keysStateRequest.getStateKey());
    Object keyStructuralValue = mapKeyCoder.structuralValue(key);
    pendingAdds.putIfAbsent(keyStructuralValue, KV.of(key, new ArrayList<>()));
    pendingAdds.get(keyStructuralValue).getValue().add(value);
  }
@Test
  public void testPut() 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.noop(),
            fakeClient,
            "instructionId",
            createMultimapKeyStateKey(),
            ByteArrayCoder.of(),
            StringUtf8Coder.of());

    Iterable<String> initValues = userState.get(A1);
    userState.put(A1, "V3");
    assertArrayEquals(new String[] {"V1", "V2"}, Iterables.toArray(initValues, String.class));
    assertArrayEquals(
        new String[] {"V1", "V2", "V3"}, Iterables.toArray(userState.get(A1), String.class));
    userState.asyncClose();
    assertThrows(IllegalStateException.class, () -> userState.put(A1, "V2"));
  }