ForwardingPTransform extends PTransform<InputT, OutputT> { @Override public OutputT expand(InputT input) { OutputT res = delegate().expand(input); if (res instanceof PCollection) { PCollection pc = (PCollection) res; try { pc.setCoder(delegate().getDefaultOutputCoder(input, pc)); } catch (CannotProvideCoderException e) { } } return res; }  @Override OutputT expand(InputT input); @Override Map<TupleTag<?>, PValue> getAdditionalInputs(); @Override void validate(PipelineOptions options); @Override String getName(); @Override void populateDisplayData(DisplayData.Builder builder);  }
@Test public void applyDelegates() { @SuppressWarnings("unchecked") PCollection<Integer> collection = mock(PCollection.class); @SuppressWarnings("unchecked") PCollection<String> output = mock(PCollection.class); when(delegate.expand(collection)).thenReturn(output); PCollection<String> result = forwarding.expand(collection); assertThat(result, equalTo(output)); }
@Test public void getDefaultOutputCoderDelegates() throws Exception { @SuppressWarnings("unchecked") PCollection<Integer> input = PCollection.createPrimitiveOutputInternal( null , WindowingStrategy.globalDefault(), PCollection.IsBounded.BOUNDED, null ); @SuppressWarnings("unchecked") PCollection<String> output = PCollection.createPrimitiveOutputInternal( null , WindowingStrategy.globalDefault(), PCollection.IsBounded.BOUNDED, null ); @SuppressWarnings("unchecked") Coder<String> outputCoder = mock(Coder.class); when(delegate.expand(input)).thenReturn(output); when(delegate.getDefaultOutputCoder(input, output)).thenReturn(outputCoder); assertThat(forwarding.expand(input).getCoder(), equalTo(outputCoder)); }