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 getRootTransformsContainsEmptyFlatten() {
    PCollections<String> flatten = Flatten.pCollections();
    PCollectionList<String> emptyList = PCollectionList.empty(p);
    PCollection<String> empty = emptyList.apply(flatten);
    empty.setCoder(StringUtf8Coder.of());
    p.traverseTopologically(visitor);
    DirectGraph graph = visitor.getGraph();
    assertThat(
        graph.getRootTransforms(),
        Matchers.containsInAnyOrder(new Object[] {graph.getProducer(empty)}));
    AppliedPTransform<?, ?, ?> onlyRoot = Iterables.getOnlyElement(graph.getRootTransforms());
    assertThat((Object) onlyRoot.getTransform(), equalTo(flatten));
    assertThat(onlyRoot.getInputs().entrySet(), emptyIterable());
    assertThat(onlyRoot.getOutputs(), equalTo(empty.expand()));
  }