@Override
  public <T> ModelEnforcement<T> forBundle(
      CommittedBundle<T> input, AppliedPTransform<?, ?, ?> consumer) {
    if (isReadTransform(consumer)) {
      return NoopReadEnforcement.INSTANCE;
    }
    return new ImmutabilityCheckingEnforcement<>(input, consumer);
  }
@Test
  public void mutatedDuringProcessElementThrows() {
    WindowedValue<byte[]> element = WindowedValue.valueInGlobalWindow("bar".getBytes(UTF_8));
    CommittedBundle<byte[]> elements =
        bundleFactory.createBundle(pcollection).add(element).commit(Instant.now());

    ModelEnforcement<byte[]> enforcement = factory.forBundle(elements, consumer);
    enforcement.beforeElement(element);
    element.getValue()[0] = 'f';
    thrown.expect(IllegalMutationException.class);
    thrown.expectMessage(consumer.getFullName());
    thrown.expectMessage("illegaly mutated");
    thrown.expectMessage("Input values must not be mutated");
    enforcement.afterElement(element);
    enforcement.afterFinish(
        elements,
        StepTransformResult.<byte[]>withoutHold(consumer).build(),
        Collections.emptyList());
  }