public DirectGraph getGraph() {
    checkState(finalized, "Can't get a graph before the Pipeline has been completely traversed");
    return DirectGraph.create(
        producers, viewWriters, perElementConsumers, rootTransforms, stepNames);
  }
@Test
  public void getValueToConsumersSucceeds() {
    PCollection<String> created = p.apply(Create.of("1", "2", "3"));
    PCollection<String> transformed =
        created.apply(
            ParDo.of(
                new DoFn<String, String>() {
                  @ProcessElement
                  public void processElement(DoFn<String, String>.ProcessContext c)
                      throws Exception {
                    c.output(Integer.toString(c.element().length()));
                  }
                }));

    PCollection<String> flattened =
        PCollectionList.of(created).and(transformed).apply(Flatten.pCollections());

    p.traverseTopologically(visitor);

    DirectGraph graph = visitor.getGraph();
    AppliedPTransform<?, ?, ?> transformedProducer = graph.getProducer(transformed);
    AppliedPTransform<?, ?, ?> flattenedProducer = graph.getProducer(flattened);

    assertThat(
        graph.getPerElementConsumers(created),
        Matchers.containsInAnyOrder(new Object[] {transformedProducer, flattenedProducer}));
    assertThat(
        graph.getPerElementConsumers(transformed),
        Matchers.containsInAnyOrder(new Object[] {flattenedProducer}));
    assertThat(graph.getPerElementConsumers(flattened), emptyIterable());
  }