public static <InputT extends PInput, OutputT extends POutput>
      PTransform<InputT, OutputT> compose(SerializableFunction<InputT, OutputT> fn) {
    return new PTransform<InputT, OutputT>() {
      @Override
      public OutputT expand(InputT input) {
        return fn.apply(input);
      }
    };
  }
@Test
  @Category(NeedsRunner.class)
  public void testComposeBasicSerializableFunction() throws Exception {
    PCollection<Integer> output =
        pipeline
            .apply(Create.of(1, 2, 3))
            .apply(
                PTransform.compose(
                    (PCollection<Integer> numbers) -> {
                      PCollection<Integer> inverted =
                          numbers.apply(MapElements.into(integers()).via(input -> -input));
                      return PCollectionList.of(numbers)
                          .and(inverted)
                          .apply(Flatten.pCollections());
                    }));

    PAssert.that(output).containsInAnyOrder(-2, -1, -3, 2, 1, 3);
    pipeline.run();
  }