@Override
  public PTransformOverrideFactory.PTransformReplacement<
          PCollection<? extends InputT>, PCollection<OutputT>>
      getReplacementTransform(
          AppliedPTransform<
                  PCollection<? extends InputT>,
                  PCollection<OutputT>,
                  SingleOutput<InputT, OutputT>>
              transform) {
    return PTransformOverrideFactory.PTransformReplacement.of(
        PTransformReplacements.getSingletonMainInput(transform),
        new ParDoSingle<>(
            transform.getTransform(),
            Iterables.getOnlyElement(transform.getOutputs().keySet()),
            PTransformReplacements.getSingletonMainOutput(transform).getCoder()));
  }
@Test
  public void getReplacementTransformGetFn() {
    DoFn<Integer, Long> originalFn = new ToLongFn();
    ParDo.SingleOutput<Integer, Long> originalTransform = ParDo.of(originalFn);
    PCollection<? extends Integer> input = pipeline.apply(Create.of(1, 2, 3));
    AppliedPTransform<
            PCollection<? extends Integer>, PCollection<Long>, ParDo.SingleOutput<Integer, Long>>
        application =
            AppliedPTransform.of(
                "original",
                PValues.expandInput(input),
                PValues.expandOutput(input.apply(originalTransform)),
                originalTransform,
                ResourceHints.create(),
                pipeline);

    PTransformReplacement<PCollection<? extends Integer>, PCollection<Long>> replacementTransform =
        factory.getReplacementTransform(application);
    ParDoSingle<Integer, Long> parDoSingle =
        (ParDoSingle<Integer, Long>) replacementTransform.getTransform();

    assertThat(parDoSingle.getFn(), equalTo(originalTransform.getFn()));
    assertThat(parDoSingle.getFn(), equalTo(originalFn));
  }