flink-instrumentation/workspace/flink-1.4.1-instrumented/flink-1.4.1/flink-optimizer/src/test/java/org/apache/flink/optimizer/plantranslate/JobGraphGeneratorTest.java
/**
	 * Verifies that the resources are merged correctly for chained operators when
	 * generating job graph
	 */
Verifies that the resources are merged correctly for chained operators when
generating job graph
Verifies that the resources are merged correctly for chained operators when generating job graph
@Test
	public void testResourcesForChainedOperators() 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);
		ResourceSpec resource6 = new ResourceSpec(0.6, 600);
		ResourceSpec resource7 = new ResourceSpec(0.7, 700);
		Method opMethod = Operator.class.getDeclaredMethod("setResources", ResourceSpec.class);
		opMethod.setAccessible(true);
		Method sinkMethod = DataSink.class.getDeclaredMethod("setResources", ResourceSpec.class);
		sinkMethod.setAccessible(true);
		MapFunction<Long, Long> mapFunction = new MapFunction<Long, Long>() {
			@Override
			public Long map(Long value) throws Exception {
				return value;
			}
		};
		FilterFunction<Long> filterFunction = new FilterFunction<Long>() {
			@Override
			public boolean filter(Long value) throws Exception {
				return false;
			}
		};
		ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment();
		DataSet<Long> input = env.fromElements(1L, 2L, 3L);
		opMethod.invoke(input, resource1);
		DataSet<Long> map1 = input.map(mapFunction);
		opMethod.invoke(map1, resource2);
		DataSet<Long> filter1 = map1.filter(filterFunction);
		opMethod.invoke(filter1, resource3);
		IterativeDataSet<Long> startOfIteration = filter1.iterate(10);
		opMethod.invoke(startOfIteration, resource4);
		DataSet<Long> map2 = startOfIteration.map(mapFunction);
		opMethod.invoke(map2, resource5);
		DataSet<Long> feedback = map2.filter(filterFunction);
		opMethod.invoke(feedback, resource6);
		DataSink<Long> sink = startOfIteration.closeWith(feedback).output(new DiscardingOutputFormat<Long>());
		sinkMethod.invoke(sink, resource7);
		Plan plan = env.createProgramPlan();
		Optimizer pc = new Optimizer(new Configuration());
		OptimizedPlan op = pc.compile(plan);
		JobGraphGenerator jgg = new JobGraphGenerator();
		JobGraph jobGraph = jgg.compileJobGraph(op);
		JobVertex sourceMapFilterVertex = jobGraph.getVerticesSortedTopologicallyFromSources().get(0);
		JobVertex iterationHeadVertex = jobGraph.getVerticesSortedTopologicallyFromSources().get(1);
		JobVertex feedbackVertex = jobGraph.getVerticesSortedTopologicallyFromSources().get(2);
		JobVertex sinkVertex = jobGraph.getVerticesSortedTopologicallyFromSources().get(3);
		JobVertex iterationSyncVertex = jobGraph.getVerticesSortedTopologicallyFromSources().get(4);
		assertTrue(sourceMapFilterVertex.getMinResources().equals(resource1.merge(resource2).merge(resource3)));
		assertTrue(iterationHeadVertex.getPreferredResources().equals(resource4));
		assertTrue(feedbackVertex.getMinResources().equals(resource5.merge(resource6)));
		assertTrue(sinkVertex.getPreferredResources().equals(resource7));
		assertTrue(iterationSyncVertex.getMinResources().equals(resource4));
	}
@ Test public void testResourcesForChainedOperators ( ) 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 ) ; ResourceSpec resource6 = new ResourceSpec ( 0.6 , 600 ) ; ResourceSpec resource7 = new ResourceSpec ( 0.7 , 700 ) ; Method opMethod = Operator . class . getDeclaredMethod ( "setResources" , ResourceSpec . class ) ; opMethod . setAccessible ( true ) ; Method sinkMethod = DataSink . class . getDeclaredMethod ( "setResources" , ResourceSpec . class ) ; sinkMethod . setAccessible ( true ) ; MapFunction < Long , Long > mapFunction = new MapFunction < Long , Long > ( ) { @ Override public Long map ( Long value ) throws Exception { return value ; } } ; FilterFunction < Long > filterFunction = new FilterFunction < Long > ( ) { @ Override public boolean filter ( Long value ) throws Exception { return false ; } } ; ExecutionEnvironment env = ExecutionEnvironment . getExecutionEnvironment ( ) ; DataSet < Long > input = env . fromElements ( 1L , 2L , 3L ) ; opMethod . invoke ( input , resource1 ) ; DataSet < Long > map1 = input . map ( mapFunction ) ; opMethod . invoke ( map1 , resource2 ) ; DataSet < Long > filter1 = map1 . filter ( filterFunction ) ; opMethod . invoke ( filter1 , resource3 ) ; IterativeDataSet < Long > startOfIteration = filter1 . iterate ( 10 ) ; opMethod . invoke ( startOfIteration , resource4 ) ; DataSet < Long > map2 = startOfIteration . map ( mapFunction ) ; opMethod . invoke ( map2 , resource5 ) ; DataSet < Long > feedback = map2 . filter ( filterFunction ) ; opMethod . invoke ( feedback , resource6 ) ; DataSink < Long > sink = startOfIteration . closeWith ( feedback ) . output ( new DiscardingOutputFormat < Long > ( ) ) ; sinkMethod . invoke ( sink , resource7 ) ; Plan plan = env . createProgramPlan ( ) ; Optimizer pc = new Optimizer ( new Configuration ( ) ) ; OptimizedPlan op = pc . compile ( plan ) ; JobGraphGenerator jgg = new JobGraphGenerator ( ) ; JobGraph jobGraph = jgg . compileJobGraph ( op ) ; JobVertex sourceMapFilterVertex = jobGraph . getVerticesSortedTopologicallyFromSources ( ) . get ( 0 ) ; JobVertex iterationHeadVertex = jobGraph . getVerticesSortedTopologicallyFromSources ( ) . get ( 1 ) ; JobVertex feedbackVertex = jobGraph . getVerticesSortedTopologicallyFromSources ( ) . get ( 2 ) ; JobVertex sinkVertex = jobGraph . getVerticesSortedTopologicallyFromSources ( ) . get ( 3 ) ; JobVertex iterationSyncVertex = jobGraph . getVerticesSortedTopologicallyFromSources ( ) . get ( 4 ) ; assertTrue ( sourceMapFilterVertex . getMinResources ( ) . equals ( resource1 . merge ( resource2 ) . merge ( resource3 ) ) ) ; assertTrue ( iterationHeadVertex . getPreferredResources ( ) . equals ( resource4 ) ) ; assertTrue ( feedbackVertex . getMinResources ( ) . equals ( resource5 . merge ( resource6 ) ) ) ; assertTrue ( sinkVertex . getPreferredResources ( ) . equals ( resource7 ) ) ; assertTrue ( iterationSyncVertex . getMinResources ( ) . equals ( resource4 ) ) ; }
Verifies that the resources are merged correctly for chained operators when
 generating job graph
Verifies that the resources are merged correctly for chained operators when generating job graph
// CHAIN(Source -> Map -> Filter) // CHAIN(Map -> Filter)