ImmutableListBundleFactory implements BundleFactory { @Override public <T> UncommittedBundle<T> createRootBundle() { return UncommittedImmutableListBundle.create(null, StructuralKey.empty()); } private  ImmutableListBundleFactory(); static ImmutableListBundleFactory create(); @Override UncommittedBundle<T> createRootBundle(); @Override UncommittedBundle<T> createBundle(PCollection<T> output); @Override UncommittedBundle<T> createKeyedBundle(
      StructuralKey<K> key, PCollection<T> output);  }
@Test public void addElementsAtEndOfTimeThrows() { Instant timestamp = BoundedWindow.TIMESTAMP_MAX_VALUE; WindowedValue<Integer> value = WindowedValue.timestampedValueInGlobalWindow(1, timestamp); UncommittedBundle<Integer> bundle = bundleFactory.createRootBundle(); thrown.expect(IllegalArgumentException.class); thrown.expectMessage(timestamp.toString()); bundle.add(value); }
@Test public void addElementsPastEndOfTimeThrows() { Instant timestamp = BoundedWindow.TIMESTAMP_MAX_VALUE.plus(Duration.standardMinutes(2)); WindowedValue<Integer> value = WindowedValue.timestampedValueInGlobalWindow(1, timestamp); UncommittedBundle<Integer> bundle = bundleFactory.createRootBundle(); thrown.expect(IllegalArgumentException.class); thrown.expectMessage(timestamp.toString()); bundle.add(value); }
@Test public void addAfterCommitShouldThrowException() { UncommittedBundle<Integer> bundle = bundleFactory.createRootBundle(); bundle.add(WindowedValue.valueInGlobalWindow(1)); CommittedBundle<Integer> firstCommit = bundle.commit(Instant.now()); assertThat(firstCommit.getElements(), containsInAnyOrder(WindowedValue.valueInGlobalWindow(1))); thrown.expect(IllegalStateException.class); thrown.expectMessage("3"); thrown.expectMessage("committed"); bundle.add(WindowedValue.valueInGlobalWindow(3)); }
@Test public void commitAfterCommitShouldThrowException() { UncommittedBundle<Integer> bundle = bundleFactory.createRootBundle(); bundle.add(WindowedValue.valueInGlobalWindow(1)); CommittedBundle<Integer> firstCommit = bundle.commit(Instant.now()); assertThat(firstCommit.getElements(), containsInAnyOrder(WindowedValue.valueInGlobalWindow(1))); thrown.expect(IllegalStateException.class); thrown.expectMessage("committed"); bundle.commit(Instant.now()); }