public void add(TimestampedValue<T> value) {
    checkState(
        !isClosed,
        "OrderedList user state is no longer usable because it is closed for %s",
        requestTemplate.getStateKey());
    Instant timestamp = value.getTimestamp();
    pendingAdds.putIfAbsent(timestamp, new ArrayList<>());
    pendingAdds.get(timestamp).add(value.getValue());
  }
@Test
  public void testAdd() throws Exception {
    FakeBeamFnStateClient fakeClient =
        new FakeBeamFnStateClient(
            timestampedValueCoder,
            ImmutableMap.of(
                createOrderedListStateKey("A", 1),
                Collections.singletonList(A1),
                createOrderedListStateKey("A", 4),
                Collections.singletonList(A4),
                createOrderedListStateKey("A", 2),
                asList(A2, B2)));

    OrderedListUserState<String> userState =
        new OrderedListUserState<>(
            Caches.noop(),
            fakeClient,
            "instructionId",
            createOrderedListStateKey("A"),
            StringUtf8Coder.of());

    // add to an existing timestamp
    userState.add(B1);
    assertArrayEquals(
        asList(A1, B1, A2, B2, A4).toArray(),
        Iterables.toArray(userState.read(), TimestampedValue.class));

    // add to a nonexistent timestamp
    userState.add(A3);
    assertArrayEquals(
        asList(A1, B1, A2, B2, A3, A4).toArray(),
        Iterables.toArray(userState.read(), TimestampedValue.class));

    // add a duplicated value
    userState.add(B1);
    assertArrayEquals(
        asList(A1, B1, B1, A2, B2, A3, A4).toArray(),
        Iterables.toArray(userState.read(), TimestampedValue.class));

    userState.asyncClose();
    assertThrows(IllegalStateException.class, () -> userState.add(A1));
  }