public static FusedPipeline fuse(Pipeline p) {
    return new GreedyPipelineFuser(p).fusedPipeline;
  }
@Test
  public void singleEnvironmentAcrossGroupByKeyMultipleStages() {
    Components components =
        partialComponents
            .toBuilder()
            .putTransforms(
                "read",
                PTransform.newBuilder()
                    .setUniqueName("Read")
                    .putInputs("input", "impulse.out")
                    .putOutputs("output", "read.out")
                    .setSpec(
                        FunctionSpec.newBuilder()
                            .setUrn(PTransformTranslation.PAR_DO_TRANSFORM_URN)
                            .setPayload(
                                ParDoPayload.newBuilder()
                                    .setDoFn(FunctionSpec.newBuilder())
                                    .build()
                                    .toByteString()))
                    .setEnvironmentId("py")
                    .build())
            .putPcollections("read.out", pc("read.out"))
            .putTransforms(
                "groupByKey",
                PTransform.newBuilder()
                    .setUniqueName("GroupByKey")
                    .putInputs("input", "read.out")
                    .putOutputs("output", "groupByKey.out")
                    .setSpec(
                        FunctionSpec.newBuilder()
                            .setUrn(PTransformTranslation.GROUP_BY_KEY_TRANSFORM_URN))
                    .build())
            .putPcollections("groupByKey.out", pc("groupByKey.out"))
            .putTransforms(
                "parDo",
                PTransform.newBuilder()
                    .setUniqueName("ParDo")
                    .putInputs("input", "groupByKey.out")
                    .putOutputs("output", "parDo.out")
                    .setSpec(
                        FunctionSpec.newBuilder()
                            .setUrn(PTransformTranslation.PAR_DO_TRANSFORM_URN)
                            .setPayload(
                                ParDoPayload.newBuilder()
                                    .setDoFn(FunctionSpec.newBuilder())
                                    .build()
                                    .toByteString()))
                    .setEnvironmentId("py")
                    .build())
            .putPcollections("parDo.out", pc("parDo.out"))
            .build();
    FusedPipeline fused =
        GreedyPipelineFuser.fuse(Pipeline.newBuilder().setComponents(components).build());

    assertThat(
        fused.getRunnerExecutedTransforms(),
        containsInAnyOrder(
            PipelineNode.pTransform("impulse", components.getTransformsOrThrow("impulse")),
            PipelineNode.pTransform("groupByKey", components.getTransformsOrThrow("groupByKey"))));
    assertThat(
        fused.getFusedStages(),
        containsInAnyOrder(
            ExecutableStageMatcher.withInput("impulse.out")
                .withOutputs("read.out")
                .withTransforms("read"),
            ExecutableStageMatcher.withInput("groupByKey.out")
                .withNoOutputs()
                .withTransforms("parDo")));
  }