@SuppressWarnings({
    "FutureReturnValueIgnored",
    "nullness" // TODO(https://github.com/apache/beam/issues/21068)
  })
  // Update data in persistent store
  public void asyncClose() throws Exception {
    checkState(
        !isClosed,
        "Multimap user state is no longer usable because it is closed for %s",
        keysStateRequest.getStateKey());
    isClosed = true;
    // No mutations necessary
    if (!isCleared && pendingRemoves.isEmpty() && pendingAdds.isEmpty()) {
      return;
    }

    startStateApiWrites();
    updateCache();
  }
@Test
  public void testAsyncClose() throws Exception {
    FakeBeamFnStateClient fakeClient =
        new FakeBeamFnStateClient(
            ImmutableMap.of(
                createMultimapKeyStateKey(),
                KV.of(ByteArrayCoder.of(), asList(A0, A1)),
                createMultimapValueStateKey(A0),
                KV.of(StringUtf8Coder.of(), asList("V1")),
                createMultimapValueStateKey(A1),
                KV.of(StringUtf8Coder.of(), asList("V1", "V2"))));
    MultimapUserState<byte[], String> userState =
        new MultimapUserState<>(
            Caches.noop(),
            fakeClient,
            "instructionId",
            createMultimapKeyStateKey(),
            ByteArrayCoder.of(),
            StringUtf8Coder.of());
    userState.remove(A0);
    userState.put(A1, "V3");
    userState.put(A2, "V1");
    userState.put(A3, "V1");
    userState.remove(A3);
    userState.asyncClose();
    Map<StateKey, ByteString> data = fakeClient.getData();
    assertNull(data.get(createMultimapValueStateKey(A0)));
    assertEquals(encode("V1", "V2", "V3"), data.get(createMultimapValueStateKey(A1)));
    assertEquals(encode("V1"), data.get(createMultimapValueStateKey(A2)));
  }