public static <InputT, OutputT> MapElements<InputT, OutputT> via(
      final InferableFunction<InputT, OutputT> fn) {
    return new MapElements<>(fn, fn.getInputTypeDescriptor(), fn.getOutputTypeDescriptor());
  }
@Test
  public void testNestedPolymorphicSimpleFunction() throws Exception {
    pipeline.enableAbandonedNodeEnforcement(false);

    pipeline
        .apply(Create.of(1, 2, 3))

        // This is the function that needs to propagate the input T to output T
        .apply("Polymorphic Identity", MapElements.via(new NestedPolymorphicSimpleFunction<>()))

        // This is a consumer to ensure that all coder inference logic is executed.
        .apply(
            "Test Consumer",
            MapElements.via(
                new SimpleFunction<KV<Integer, String>, Integer>() {
                  @Override
                  public Integer apply(KV<Integer, String> input) {
                    return 42;
                  }
                }));
  }