ImmutabilityCheckingBundleFactory implements BundleFactory { @Override public <K, T> UncommittedBundle<T> createKeyedBundle( StructuralKey<K> key, PCollection<T> output) { if (Enforcement.IMMUTABILITY.appliesTo(output, graph)) { return new ImmutabilityEnforcingBundle<>(underlying.createKeyedBundle(key, output)); } return underlying.createKeyedBundle(key, 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 noMutationKeyedBundleSucceeds() { UncommittedBundle<byte[]> keyed = factory.createKeyedBundle(StructuralKey.of("mykey", StringUtf8Coder.of()), 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); keyed.add(windowedArray); CommittedBundle<byte[]> committed = keyed.commit(Instant.now()); assertThat(committed.getElements(), containsInAnyOrder(windowedArray)); }
@Test public void mutationBeforeAddKeyedBundleSucceeds() { UncommittedBundle<byte[]> keyed = factory.createKeyedBundle(StructuralKey.of("mykey", StringUtf8Coder.of()), transformed); byte[] array = new byte[] {4, 8, 12}; array[0] = Byte.MAX_VALUE; WindowedValue<byte[]> windowedArray = WindowedValue.of( array, new Instant(891L), new IntervalWindow(new Instant(0), new Instant(1000)), PaneInfo.ON_TIME_AND_ONLY_FIRING); keyed.add(windowedArray); CommittedBundle<byte[]> committed = keyed.commit(Instant.now()); assertThat(committed.getElements(), containsInAnyOrder(windowedArray)); }
@Test public void mutationAfterAddKeyedBundleThrows() { UncommittedBundle<byte[]> keyed = factory.createKeyedBundle(StructuralKey.of("mykey", StringUtf8Coder.of()), 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); keyed.add(windowedArray); array[0] = Byte.MAX_VALUE; thrown.expect(IllegalMutationException.class); thrown.expectMessage("Values must not be mutated in any way after being output"); keyed.commit(Instant.now()); }