KeyedPValueTrackingVisitor extends PipelineVisitor.Defaults { public static KeyedPValueTrackingVisitor create() { return new KeyedPValueTrackingVisitor(); } private  KeyedPValueTrackingVisitor(); static KeyedPValueTrackingVisitor create(); @Override CompositeBehavior enterCompositeTransform(TransformHierarchy.Node node); @Override void leaveCompositeTransform(TransformHierarchy.Node node); @Override void visitValue(PValue value, TransformHierarchy.Node producer); Set<PValue> getKeyedPValues();  }
@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); }