PipelineTranslationModeOptimizer extends FlinkPipelineTranslator { static boolean hasUnboundedOutput(Pipeline p) { PipelineTranslationModeOptimizer optimizer = new PipelineTranslationModeOptimizer(); optimizer.translate(p); return optimizer.hasUnboundedCollections; } private  PipelineTranslationModeOptimizer(); @Override void translate(Pipeline pipeline); @Override CompositeBehavior enterCompositeTransform(TransformHierarchy.Node node); @Override void leaveCompositeTransform(TransformHierarchy.Node node); @Override void visitPrimitiveTransform(TransformHierarchy.Node node); @Override void visitValue(PValue value, TransformHierarchy.Node producer);  }
@Test public void testUnboundedCollectionProducingTransform() { PipelineOptions options = PipelineOptionsFactory.create(); options.setRunner(FlinkRunner.class); Pipeline pipeline = Pipeline.create(options); pipeline.apply(GenerateSequence.from(0)); assertThat(PipelineTranslationModeOptimizer.hasUnboundedOutput(pipeline), is(true)); }
@Test public void testBoundedCollectionProducingTransform() { PipelineOptions options = PipelineOptionsFactory.create(); options.setRunner(FlinkRunner.class); Pipeline pipeline = Pipeline.create(options); pipeline.apply(GenerateSequence.from(0).to(10)); assertThat(PipelineTranslationModeOptimizer.hasUnboundedOutput(pipeline), is(false)); }