public static String toStepName(ExecutableStage executableStage) {
    /*
     * Look for the first/input ParDo/DoFn in this executable stage by
     * matching ParDo/DoFn's input PCollection with executable stage's
     * input PCollection
     */
    Set<PipelineNode.PTransformNode> inputs =
        executableStage.getTransforms().stream()
            .filter(
                transform ->
                    transform
                        .getTransform()
                        .getInputsMap()
                        .containsValue(executableStage.getInputPCollection().getId()))
            .collect(Collectors.toSet());

    Set<String> outputIds =
        executableStage.getOutputPCollections().stream()
            .map(PipelineNode.PCollectionNode::getId)
            .collect(Collectors.toSet());

    /*
     * Look for the last/output ParDo/DoFn in this executable stage by
     * matching ParDo/DoFn's output PCollection(s) with executable stage's
     * out PCollection(s)
     */
    Set<PipelineNode.PTransformNode> outputs =
        executableStage.getTransforms().stream()
            .filter(
                transform ->
                    CollectionUtils.containsAny(
                        transform.getTransform().getOutputsMap().values(), outputIds))
            .collect(Collectors.toSet());

    return String.format("[%s-%s]", toStepName(inputs), toStepName(outputs));
  }
@Test
  public void testExecutableStageWithOutput() {
    pipeline
        .apply("MyCreateOf", Create.of(KV.of(1L, "1")))
        .apply("MyFilterBy", Filter.by(Objects::nonNull))
        .apply(GroupByKey.create());

    assertEquals("[MyCreateOf-MyFilterBy]", DoFnUtils.toStepName(getOnlyExecutableStage(pipeline)));
  }