public static FusedPipeline fuse(Pipeline p) {
    return new GreedyPipelineFuser(p).fusedPipeline;
  }
@Test
  public void parDoWithStateAndTimerRootsStage() {
    PTransform timerTransform =
        PTransform.newBuilder()
            .setUniqueName("TimerParDo")
            .putInputs("input", "impulse.out")
            .putInputs("timer", "timer.out")
            .putOutputs("timer", "timer.out")
            .putOutputs("output", "output.out")
            .setSpec(
                FunctionSpec.newBuilder()
                    .setUrn(PTransformTranslation.PAR_DO_TRANSFORM_URN)
                    .setPayload(
                        ParDoPayload.newBuilder()
                            .setDoFn(FunctionSpec.newBuilder())
                            .putStateSpecs("state", StateSpec.getDefaultInstance())
                            .putTimerFamilySpecs("timer", TimerFamilySpec.getDefaultInstance())
                            .build()
                            .toByteString()))
            .setEnvironmentId("common")
            .build();

    Components components =
        partialComponents
            .toBuilder()
            .putTransforms("timer", timerTransform)
            .putPcollections("timer.out", pc("timer.out"))
            .putPcollections("output.out", pc("output.out"))
            .putEnvironments("common", Environments.createDockerEnvironment("common"))
            .build();

    FusedPipeline fused =
        GreedyPipelineFuser.fuse(
            Pipeline.newBuilder()
                .setComponents(components)
                .addRequirements(ParDoTranslation.REQUIRES_STATEFUL_PROCESSING_URN)
                .build());

    assertThat(
        fused.getRunnerExecutedTransforms(),
        containsInAnyOrder(
            PipelineNode.pTransform("impulse", components.getTransformsOrThrow("impulse"))));
    assertThat(
        fused.getFusedStages(),
        contains(
            ExecutableStageMatcher.withInput("impulse.out")
                .withNoOutputs()
                .withTransforms("timer")));
  }