public static String toJsonString(Pipeline pipeline, ConfigContext ctx) {
    final PipelineJsonRenderer visitor = new PipelineJsonRenderer(ctx);
    pipeline.traverseTopologically(visitor);
    return visitor.jsonBuilder.toString();
  }
@Test
  public void testEmptyPipeline() {
    SamzaPipelineOptions options = PipelineOptionsFactory.create().as(SamzaPipelineOptions.class);
    options.setRunner(SamzaRunner.class);

    Pipeline p = Pipeline.create(options);
    final Map<PValue, String> idMap = PViewToIdMapper.buildIdMap(p);
    final Set<String> nonUniqueStateIds = StateIdParser.scan(p);
    final ConfigContext ctx = new ConfigContext(idMap, nonUniqueStateIds, options);

    String jsonDag =
        "{  \"RootNode\": ["
            + "    { \"fullName\":\"OuterMostNode\","
            + "      \"ChildNodes\":[    ]}],\"graphLinks\": [],\"transformIOInfo\": []"
            + "}";

    assertEquals(
        JsonParser.parseString(jsonDag),
        JsonParser.parseString(
            PipelineJsonRenderer.toJsonString(p, ctx).replaceAll(System.lineSeparator(), "")));
  }