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 getValueToConsumersWithDuplicateInputSucceeds() {
    PCollection<String> created = p.apply(Create.of("1", "2", "3"));

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

    p.traverseTopologically(visitor);

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

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