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