public static FusedPipeline fuse(Pipeline p) {
    return new GreedyPipelineFuser(p).fusedPipeline;
  }
@Test
  public void statefulParDoRootsStage() {
    // (impulse.out) -> parDo -> (parDo.out)
    // (parDo.out) -> stateful -> stateful.out
    // stateful has a state spec which prevents it from fusing with an upstream ParDo
    PTransform parDoTransform =
        PTransform.newBuilder()
            .setUniqueName("ParDo")
            .putInputs("input", "impulse.out")
            .putOutputs("output", "parDo.out")
            .setSpec(
                FunctionSpec.newBuilder()
                    .setUrn(PTransformTranslation.PAR_DO_TRANSFORM_URN)
                    .setPayload(
                        ParDoPayload.newBuilder()
                            .setDoFn(FunctionSpec.newBuilder())
                            .build()
                            .toByteString()))
            .setEnvironmentId("common")
            .build();
    PTransform statefulTransform =
        PTransform.newBuilder()
            .setUniqueName("StatefulParDo")
            .putInputs("input", "parDo.out")
            .putOutputs("output", "stateful.out")
            .setSpec(
                FunctionSpec.newBuilder()
                    .setUrn(PTransformTranslation.PAR_DO_TRANSFORM_URN)
                    .setPayload(
                        ParDoPayload.newBuilder()
                            .setDoFn(FunctionSpec.newBuilder())
                            .putStateSpecs("state", StateSpec.getDefaultInstance())
                            .build()
                            .toByteString()))
            .setEnvironmentId("common")
            .build();

    Components components =
        partialComponents
            .toBuilder()
            .putTransforms("parDo", parDoTransform)
            .putPcollections("parDo.out", pc("parDo.out"))
            .putTransforms("stateful", statefulTransform)
            .putPcollections("stateful.out", pc("stateful.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(),
        containsInAnyOrder(
            ExecutableStageMatcher.withInput("impulse.out")
                .withOutputs("parDo.out")
                .withTransforms("parDo"),
            ExecutableStageMatcher.withInput("parDo.out")
                .withNoOutputs()
                .withTransforms("stateful")));
  }