class 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) {
      // If we were cleared we should disregard old values.
      return Iterables.limit(Collections.unmodifiableList(newValues), newValues.size());
    } else if (newValues.isEmpty()) {
      // If we have no new values then just return the old values.
      return oldValues;
    }
    return Iterables.concat(
        oldValues, Iterables.limit(Collections.unmodifiableList(newValues), newValues.size()));
  }

  public  BagUserState(
      BeamFnStateClient beamFnStateClient,
      String instructionId,
      String ptransformId,
      String stateId,
      ByteString encodedWindow,
      ByteString encodedKey,
      Coder<T> valueCoder);

}