@Override
  public <K, T> UncommittedBundle<T> createKeyedBundle(
      StructuralKey<K> key, PCollection<T> output) {
    return new CloningBundle<>(underlying.createKeyedBundle(key, output));
  }
@Test
  public void keyedBundleDecodeFailsAddFails() {
    PCollection<Record> pc = p.apply(Create.empty(new RecordNoDecodeCoder()));
    UncommittedBundle<Record> bundle =
        factory.createKeyedBundle(StructuralKey.of("foo", StringUtf8Coder.of()), pc);

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