public static String toJsonString(Pipeline pipeline, ConfigContext ctx) {
    final PipelineJsonRenderer visitor = new PipelineJsonRenderer(ctx);
    pipeline.traverseTopologically(visitor);
    return visitor.jsonBuilder.toString();
  }
@Test
  public void testCompositePipeline() throws IOException {
    SamzaPipelineOptions options = PipelineOptionsFactory.create().as(SamzaPipelineOptions.class);
    options.setRunner(SamzaRunner.class);
    options.setJobName("TestEnvConfig");
    options.setSamzaExecutionEnvironment(SamzaExecutionEnvironment.LOCAL);

    Pipeline p = Pipeline.create(options);

    p.apply(
            Create.timestamped(
                TimestampedValue.of(KV.of(1, 1), new Instant(1)),
                TimestampedValue.of(KV.of(2, 2), new Instant(2))))
        .apply(Window.into(FixedWindows.of(Duration.millis(10))))
        .apply(Sum.integersPerKey());

    p.replaceAll(SamzaTransformOverrides.getDefaultOverrides());

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

    String jsonDagFileName = "src/test/resources/ExpectedDag.json";
    String jsonDag =
        new String(Files.readAllBytes(Paths.get(jsonDagFileName)), StandardCharsets.UTF_8);
    String renderedDag = PipelineJsonRenderer.toJsonString(p, ctx);

    assertEquals(
        JsonParser.parseString(jsonDag),
        JsonParser.parseString(renderedDag.replaceAll(System.lineSeparator(), "")));
  }