@Override
  public <T> UncommittedBundle<T> createBundle(PCollection<T> output) {
    return new CloningBundle<>(underlying.createBundle(output));
  }
@Test
  public void bundleEncodeFailsAddFails() {
    PCollection<Record> pc = p.apply(Create.empty(new RecordNoEncodeCoder()));
    UncommittedBundle<Record> bundle = factory.createBundle(pc);

    thrown.expect(UserCodeException.class);
    thrown.expectCause(isA(CoderException.class));
    thrown.expectMessage("Encode not allowed");
    bundle.add(WindowedValue.valueInGlobalWindow(new Record()));
  }