ImmutabilityCheckingBundleFactory implements BundleFactory { @Override public <T> UncommittedBundle<T> createBundle(PCollection<T> output) { if (Enforcement.IMMUTABILITY.appliesTo(output, graph)) { return new ImmutabilityEnforcingBundle<>(underlying.createBundle(output)); } return underlying.createBundle(output); } private  ImmutabilityCheckingBundleFactory(BundleFactory underlying, DirectGraph graph); static ImmutabilityCheckingBundleFactory create(
      BundleFactory underlying, DirectGraph graph); @Override UncommittedBundle<T> createRootBundle(); @Override UncommittedBundle<T> createBundle(PCollection<T> output); @Override UncommittedBundle<T> createKeyedBundle(
      StructuralKey<K> key, PCollection<T> output);  }
@Test public void noMutationCreateBundleSucceeds() { UncommittedBundle<byte[]> intermediate = factory.createBundle(transformed); WindowedValue<byte[]> windowedArray = WindowedValue.of( new byte[] {4, 8, 12}, new Instant(891L), new IntervalWindow(new Instant(0), new Instant(1000)), PaneInfo.ON_TIME_AND_ONLY_FIRING); intermediate.add(windowedArray); CommittedBundle<byte[]> committed = intermediate.commit(Instant.now()); assertThat(committed.getElements(), containsInAnyOrder(windowedArray)); }
@Test public void mutationBeforeAddCreateBundleSucceeds() { UncommittedBundle<byte[]> intermediate = factory.createBundle(transformed); byte[] array = new byte[] {4, 8, 12}; WindowedValue<byte[]> windowedArray = WindowedValue.of( array, new Instant(891L), new IntervalWindow(new Instant(0), new Instant(1000)), PaneInfo.ON_TIME_AND_ONLY_FIRING); array[2] = -3; intermediate.add(windowedArray); CommittedBundle<byte[]> committed = intermediate.commit(Instant.now()); assertThat(committed.getElements(), containsInAnyOrder(windowedArray)); }
@Test public void mutationAfterAddCreateBundleThrows() { UncommittedBundle<byte[]> intermediate = factory.createBundle(transformed); byte[] array = new byte[] {4, 8, 12}; WindowedValue<byte[]> windowedArray = WindowedValue.of( array, new Instant(891L), new IntervalWindow(new Instant(0), new Instant(1000)), PaneInfo.ON_TIME_AND_ONLY_FIRING); intermediate.add(windowedArray); array[2] = -3; thrown.expect(IllegalMutationException.class); thrown.expectMessage("Values must not be mutated in any way after being output"); intermediate.commit(Instant.now()); }