@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);
  }
@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());
  }