BagUserState { public Iterable<T> get() { checkState( !isClosed, "Bag user state is no longer usable because it is closed for %s", request.getStateKey()); if (oldValues == null) { return Iterables.limit(Collections.unmodifiableList(newValues), newValues.size()); } else if (newValues.isEmpty()) { return oldValues; } return Iterables.concat( oldValues, Iterables.limit(Collections.unmodifiableList(newValues), newValues.size())); } 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 testGet() 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.asyncClose(); thrown.expect(IllegalStateException.class); userState.get(); }