BagUserState { public void append(T t) { checkState( !isClosed, "Bag user state is no longer usable because it is closed for %s", request.getStateKey()); newValues.add(t); } 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 testAppend() throws Exception { FakeBeamFnStateClient fakeClient = new FakeBeamFnStateClient(ImmutableMap.of(key("A"), encode("A1"))); BagUserState<String> userState = new BagUserState<>( fakeClient, "instructionId", "ptransformId", "stateId", ByteString.copyFromUtf8("encodedWindow"), encode("A"), StringUtf8Coder.of()); userState.append("A2"); Iterable<String> stateBeforeA3 = userState.get(); assertArrayEquals(new String[] {"A1", "A2"}, Iterables.toArray(stateBeforeA3, String.class)); userState.append("A3"); assertArrayEquals(new String[] {"A1", "A2"}, Iterables.toArray(stateBeforeA3, String.class)); assertArrayEquals( new String[] {"A1", "A2", "A3"}, Iterables.toArray(userState.get(), String.class)); userState.asyncClose(); assertEquals(encode("A1", "A2", "A3"), fakeClient.getData().get(key("A"))); thrown.expect(IllegalStateException.class); userState.append("A4"); }