CloningBundleFactory implements BundleFactory { @Override public <T> UncommittedBundle<T> createBundle(PCollection<T> output) { return new CloningBundle<>(underlying.createBundle(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 bundleWorkingCoderSucceedsClonesOutput() { PCollection<Integer> created = p.apply(Create.of(1, 3).withCoder(VarIntCoder.of())); PCollection<KV<String, Integer>> kvs = created .apply(WithKeys.of("foo")) .setCoder(KvCoder.of(StringUtf8Coder.of(), VarIntCoder.of())); WindowedValue<KV<String, Integer>> fooOne = WindowedValue.valueInGlobalWindow(KV.of("foo", 1)); WindowedValue<KV<String, Integer>> fooThree = WindowedValue.valueInGlobalWindow(KV.of("foo", 3)); CommittedBundle<KV<String, Integer>> bundle = factory.createBundle(kvs).add(fooOne).add(fooThree).commit(Instant.now()); assertThat(bundle.getElements(), containsInAnyOrder(fooOne, fooThree)); assertThat( bundle.getElements(), not(containsInAnyOrder(theInstance(fooOne), theInstance(fooThree)))); for (WindowedValue<KV<String, Integer>> foo : bundle.getElements()) { assertThat( foo.getValue(), not(anyOf(theInstance(fooOne.getValue()), theInstance(fooThree.getValue())))); } assertThat(bundle.getPCollection(), equalTo(kvs)); }
@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())); }
@Test public void bundleDecodeFailsAddFails() { PCollection<Record> pc = p.apply(Create.empty(new RecordNoDecodeCoder())); UncommittedBundle<Record> bundle = factory.createBundle(pc); thrown.expect(UserCodeException.class); thrown.expectCause(isA(CoderException.class)); thrown.expectMessage("Decode not allowed"); bundle.add(WindowedValue.valueInGlobalWindow(new Record())); }