flink-instrumentation/workspace/flink-1.4.1-instrumented/flink-1.4.1/flink-streaming-java/src/test/java/org/apache/flink/streaming/api/graph/StreamingJobGraphGeneratorTest.java
/**
	 * Verifies that the resources are merged correctly for chained operators (covers source and sink cases)
	 * when generating job graph.
	 */
Verifies that the resources are merged correctly for chained operators (covers source and sink cases)
when generating job graph.
Verifies that the resources are merged correctly for chained operators ( covers source and sink cases ) when generating job graph .
@Test
	public void testResourcesForChainedSourceSink() throws Exception {
		ResourceSpec resource1 = new ResourceSpec(0.1, 100);
		ResourceSpec resource2 = new ResourceSpec(0.2, 200);
		ResourceSpec resource3 = new ResourceSpec(0.3, 300);
		ResourceSpec resource4 = new ResourceSpec(0.4, 400);
		ResourceSpec resource5 = new ResourceSpec(0.5, 500);
		Method opMethod = SingleOutputStreamOperator.class.getDeclaredMethod("setResources", ResourceSpec.class);
		opMethod.setAccessible(true);
		Method sinkMethod = DataStreamSink.class.getDeclaredMethod("setResources", ResourceSpec.class);
		sinkMethod.setAccessible(true);
		StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
		DataStream<Tuple2<Integer, Integer>> source = env.addSource(new ParallelSourceFunction<Tuple2<Integer, Integer>>() {
			@Override
			public void run(SourceContext<Tuple2<Integer, Integer>> ctx) throws Exception {
			}
			@Override
			public void cancel() {
			}
		});
		opMethod.invoke(source, resource1);
		DataStream<Tuple2<Integer, Integer>> map = source.map(new MapFunction<Tuple2<Integer, Integer>, Tuple2<Integer, Integer>>() {
			@Override
			public Tuple2<Integer, Integer> map(Tuple2<Integer, Integer> value) throws Exception {
				return value;
			}
		});
		opMethod.invoke(map, resource2);
		DataStream<Tuple2<Integer, Integer>> filter = map.filter(new FilterFunction<Tuple2<Integer, Integer>>() {
			@Override
			public boolean filter(Tuple2<Integer, Integer> value) throws Exception {
				return false;
			}
		});
		opMethod.invoke(filter, resource3);
		DataStream<Tuple2<Integer, Integer>> reduce = filter.keyBy(0).reduce(new ReduceFunction<Tuple2<Integer, Integer>>() {
			@Override
			public Tuple2<Integer, Integer> reduce(Tuple2<Integer, Integer> value1, Tuple2<Integer, Integer> value2) throws Exception {
				return new Tuple2<>(value1.f0, value1.f1 + value2.f1);
			}
		});
		opMethod.invoke(reduce, resource4);
		DataStreamSink<Tuple2<Integer, Integer>> sink = reduce.addSink(new SinkFunction<Tuple2<Integer, Integer>>() {
			@Override
			public void invoke(Tuple2<Integer, Integer> value) throws Exception {
			}
		});
		sinkMethod.invoke(sink, resource5);
		JobGraph jobGraph = StreamingJobGraphGenerator.createJobGraph(env.getStreamGraph());
		JobVertex sourceMapFilterVertex = jobGraph.getVerticesSortedTopologicallyFromSources().get(0);
		JobVertex reduceSinkVertex = jobGraph.getVerticesSortedTopologicallyFromSources().get(1);
		assertTrue(sourceMapFilterVertex.getMinResources().equals(resource1.merge(resource2).merge(resource3)));
		assertTrue(reduceSinkVertex.getPreferredResources().equals(resource4.merge(resource5)));
	}
@ Test public void testResourcesForChainedSourceSink ( ) throws Exception { ResourceSpec resource1 = new ResourceSpec ( 0.1 , 100 ) ; ResourceSpec resource2 = new ResourceSpec ( 0.2 , 200 ) ; ResourceSpec resource3 = new ResourceSpec ( 0.3 , 300 ) ; ResourceSpec resource4 = new ResourceSpec ( 0.4 , 400 ) ; ResourceSpec resource5 = new ResourceSpec ( 0.5 , 500 ) ; Method opMethod = SingleOutputStreamOperator . class . getDeclaredMethod ( "setResources" , ResourceSpec . class ) ; opMethod . setAccessible ( true ) ; Method sinkMethod = DataStreamSink . class . getDeclaredMethod ( "setResources" , ResourceSpec . class ) ; sinkMethod . setAccessible ( true ) ; StreamExecutionEnvironment env = StreamExecutionEnvironment . getExecutionEnvironment ( ) ; DataStream < Tuple2 < Integer , Integer > > source = env . addSource ( new ParallelSourceFunction < Tuple2 < Integer , Integer > > ( ) { @ Override public void run ( SourceContext < Tuple2 < Integer , Integer > > ctx ) throws Exception { } @ Override public void cancel ( ) { } } ) ; opMethod . invoke ( source , resource1 ) ; DataStream < Tuple2 < Integer , Integer > > map = source . map ( new MapFunction < Tuple2 < Integer , Integer > , Tuple2 < Integer , Integer > > ( ) { @ Override public Tuple2 < Integer , Integer > map ( Tuple2 < Integer , Integer > value ) throws Exception { return value ; } } ) ; opMethod . invoke ( map , resource2 ) ; DataStream < Tuple2 < Integer , Integer > > filter = map . filter ( new FilterFunction < Tuple2 < Integer , Integer > > ( ) { @ Override public boolean filter ( Tuple2 < Integer , Integer > value ) throws Exception { return false ; } } ) ; opMethod . invoke ( filter , resource3 ) ; DataStream < Tuple2 < Integer , Integer > > reduce = filter . keyBy ( 0 ) . reduce ( new ReduceFunction < Tuple2 < Integer , Integer > > ( ) { @ Override public Tuple2 < Integer , Integer > reduce ( Tuple2 < Integer , Integer > value1 , Tuple2 < Integer , Integer > value2 ) throws Exception { return new Tuple2 < > ( value1 . f0 , value1 . f1 + value2 . f1 ) ; } } ) ; opMethod . invoke ( reduce , resource4 ) ; DataStreamSink < Tuple2 < Integer , Integer > > sink = reduce . addSink ( new SinkFunction < Tuple2 < Integer , Integer > > ( ) { @ Override public void invoke ( Tuple2 < Integer , Integer > value ) throws Exception { } } ) ; sinkMethod . invoke ( sink , resource5 ) ; JobGraph jobGraph = StreamingJobGraphGenerator . createJobGraph ( env . getStreamGraph ( ) ) ; JobVertex sourceMapFilterVertex = jobGraph . getVerticesSortedTopologicallyFromSources ( ) . get ( 0 ) ; JobVertex reduceSinkVertex = jobGraph . getVerticesSortedTopologicallyFromSources ( ) . get ( 1 ) ; assertTrue ( sourceMapFilterVertex . getMinResources ( ) . equals ( resource1 . merge ( resource2 ) . merge ( resource3 ) ) ) ; assertTrue ( reduceSinkVertex . getPreferredResources ( ) . equals ( resource4 . merge ( resource5 ) ) ) ; }
Verifies that the resources are merged correctly for chained operators (covers source and sink cases)
 when generating job graph.
Verifies that the resources are merged correctly for chained operators ( covers source and sink cases ) when generating job graph .
// CHAIN(Source -> Map -> Filter)