public void clear() {
    checkState(
        !isClosed,
        "Bag user state is no longer usable because it is closed for %s",
        request.getStateKey());
    isCleared = true;
    newValues = new ArrayList<>();
  }
@Test
  public void testClear() throws Exception {
    FakeBeamFnStateClient fakeClient =
        new FakeBeamFnStateClient(
            StringUtf8Coder.of(), ImmutableMap.of(key("A"), asList("A1", "A2", "A3")));
    BagUserState<String> userState =
        new BagUserState<>(
            Caches.noop(), fakeClient, "instructionId", key("A"), StringUtf8Coder.of());
    assertArrayEquals(
        new String[] {"A1", "A2", "A3"}, Iterables.toArray(userState.get(), String.class));
    userState.clear();
    assertFalse(userState.get().iterator().hasNext());
    userState.append("A4");
    assertArrayEquals(new String[] {"A4"}, Iterables.toArray(userState.get(), String.class));
    userState.clear();
    assertFalse(userState.get().iterator().hasNext());
    userState.asyncClose();

    assertNull(fakeClient.getData().get(key("A")));
    assertThrows(IllegalStateException.class, () -> userState.clear());
  }