CloningBundleFactory implements BundleFactory { @Override public <K, T> UncommittedBundle<T> createKeyedBundle( StructuralKey<K> key, PCollection<T> output) { return new CloningBundle<>(underlying.createKeyedBundle(key, output)); } private  CloningBundleFactory(); static CloningBundleFactory create(); @Override UncommittedBundle<T> createRootBundle(); @Override UncommittedBundle<T> createBundle(PCollection<T> output); @Override UncommittedBundle<T> createKeyedBundle(
      StructuralKey<K> key, PCollection<T> output);  }
@Test public void keyedBundleEncodeFailsAddFails() { PCollection<Record> pc = p.apply(Create.empty(new RecordNoEncodeCoder())); UncommittedBundle<Record> bundle = factory.createKeyedBundle(StructuralKey.of("foo", StringUtf8Coder.of()), pc); thrown.expect(UserCodeException.class); thrown.expectCause(isA(CoderException.class)); thrown.expectMessage("Encode not allowed"); bundle.add(WindowedValue.valueInGlobalWindow(new Record())); }
@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())); }