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 getRootTransformsContainsRootTransforms() {
    PCollection<byte[]> impulse = p.apply(Impulse.create());
    impulse.apply(WithKeys.of("abc"));
    p.traverseTopologically(visitor);
    DirectGraph graph = visitor.getGraph();
    assertThat(graph.getRootTransforms(), hasSize(1));
    assertThat(
        graph.getRootTransforms(),
        Matchers.containsInAnyOrder(new Object[] {graph.getProducer(impulse)}));
    for (AppliedPTransform<?, ?, ?> root : graph.getRootTransforms()) {
      // Root transforms will have no inputs
      assertThat(root.getInputs().entrySet(), emptyIterable());
      assertThat(
          Iterables.getOnlyElement(root.getOutputs().values()), Matchers.<POutput>isOneOf(impulse));
    }
  }