@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 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));
  }