public PrefetchableIterable<T> get() {
    checkState(
        !isClosed,
        "Bag user state is no longer usable because it is closed for %s",
        request.getStateKey());
    if (isCleared) {
      // If we were cleared we should disregard old values.
      return PrefetchableIterables.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 PrefetchableIterables.concat(
        oldValues, Iterables.limit(Collections.unmodifiableList(newValues), newValues.size()));
  }
@Test
  public void testGet() 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.asyncClose();
    assertThrows(IllegalStateException.class, () -> userState.get());
  }