BagUserState { public void clear() { checkState( !isClosed, "Bag user state is no longer usable because it is closed for %s", request.getStateKey()); oldValues = null; newValues = new ArrayList<>(); } BagUserState(
      BeamFnStateClient beamFnStateClient,
      String instructionId,
      String ptransformId,
      String stateId,
      ByteString encodedWindow,
      ByteString encodedKey,
      Coder<T> valueCoder); Iterable<T> get(); void append(T t); void clear(); void asyncClose();  }
@Test public void testClear() throws Exception { FakeBeamFnStateClient fakeClient = new FakeBeamFnStateClient(ImmutableMap.of(key("A"), encode("A1", "A2", "A3"))); BagUserState<String> userState = new BagUserState<>( fakeClient, "instructionId", "ptransformId", "stateId", ByteString.copyFromUtf8("encodedWindow"), encode("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"))); thrown.expect(IllegalStateException.class); userState.clear(); }