public static KeyedPValueTrackingVisitor create() {
    return new KeyedPValueTrackingVisitor();
  }
@Test
  public void traverseMultipleTimesThrows() {
    p.apply(
            Create.of(KV.of(1, (Void) null), KV.of(2, (Void) null), KV.of(3, (Void) null))
                .withCoder(KvCoder.of(VarIntCoder.of(), VoidCoder.of())))
        .apply(GroupByKey.create())
        .apply(Keys.create());

    p.traverseTopologically(visitor);
    thrown.expect(IllegalStateException.class);
    thrown.expectMessage("already been finalized");
    thrown.expectMessage(KeyedPValueTrackingVisitor.class.getSimpleName());
    p.traverseTopologically(visitor);
  }