@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) {
        // Let coder inference happen later.
      }
    }
    return res;
  }
@Test
  public void getDefaultOutputCoderDelegates() throws Exception {
    @SuppressWarnings("unchecked")
    PCollection<Integer> input =
        PCollection.createPrimitiveOutputInternal(
            null /* pipeline */,
            WindowingStrategy.globalDefault(),
            PCollection.IsBounded.BOUNDED,
            null /* coder */);
    @SuppressWarnings("unchecked")
    PCollection<String> output =
        PCollection.createPrimitiveOutputInternal(
            null /* pipeline */,
            WindowingStrategy.globalDefault(),
            PCollection.IsBounded.BOUNDED,
            null /* coder */);
    @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));
  }