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