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 set onto each job vertex correctly when generating job graph
	 * which covers the delta iteration case
	 */
Verifies that the resources are set onto each job vertex correctly when generating job graph
which covers the delta iteration case
Verifies that the resources are set onto each job vertex correctly when generating job graph which covers the delta iteration case
@Test
	public void testResourcesForDeltaIteration() 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);
		Method opMethod = Operator.class.getDeclaredMethod("setResources", ResourceSpec.class);
		opMethod.setAccessible(true);
		Method deltaMethod = DeltaIteration.class.getDeclaredMethod("setResources", ResourceSpec.class);
		deltaMethod.setAccessible(true);
		Method sinkMethod = DataSink.class.getDeclaredMethod("setResources", ResourceSpec.class);
		sinkMethod.setAccessible(true);
		MapFunction<Tuple2<Long, Long>, Tuple2<Long, Long>> mapFunction = new MapFunction<Tuple2<Long, Long>, Tuple2<Long, Long>>() {
			@Override
			public Tuple2<Long, Long> map(Tuple2<Long, Long> value) throws Exception {
				return value;
			}
		};
		FilterFunction<Tuple2<Long, Long>> filterFunction = new FilterFunction<Tuple2<Long, Long>>() {
			@Override
			public boolean filter(Tuple2<Long, Long> value) throws Exception {
				return false;
			}
		};
		ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment();
		DataSet<Tuple2<Long, Long>> input = env.fromElements(new Tuple2<>(1L, 2L));
		opMethod.invoke(input, resource1);
		DataSet<Tuple2<Long, Long>> map = input.map(mapFunction);
		opMethod.invoke(map, resource2);
		DeltaIteration<Tuple2<Long, Long>, Tuple2<Long, Long>> iteration = map.iterateDelta(map, 100, 0).
				registerAggregator("test", new LongSumAggregator());
		deltaMethod.invoke(iteration, resource3);
		DataSet<Tuple2<Long, Long>> delta = iteration.getWorkset().map(mapFunction);
		opMethod.invoke(delta, resource4);
		DataSet<Tuple2<Long, Long>> feedback = delta.filter(filterFunction);
		opMethod.invoke(feedback, resource5);
		DataSink<Tuple2<Long, Long>> sink = iteration.closeWith(delta, feedback).
				output(new DiscardingOutputFormat<Tuple2<Long, Long>>());
		sinkMethod.invoke(sink, resource6);
		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 sourceMapVertex = jobGraph.getVerticesSortedTopologicallyFromSources().get(0);
		JobVertex iterationHeadVertex = jobGraph.getVerticesSortedTopologicallyFromSources().get(1);
		JobVertex deltaVertex = jobGraph.getVerticesSortedTopologicallyFromSources().get(2);
		JobVertex iterationTailVertex = jobGraph.getVerticesSortedTopologicallyFromSources().get(3);
		JobVertex feedbackVertex = jobGraph.getVerticesSortedTopologicallyFromSources().get(4);
		JobVertex sinkVertex = jobGraph.getVerticesSortedTopologicallyFromSources().get(5);
		JobVertex iterationSyncVertex = jobGraph.getVerticesSortedTopologicallyFromSources().get(6);
		assertTrue(sourceMapVertex.getMinResources().equals(resource1.merge(resource2)));
		assertTrue(iterationHeadVertex.getPreferredResources().equals(resource3));
		assertTrue(deltaVertex.getMinResources().equals(resource4));
		assertTrue(iterationTailVertex.getPreferredResources().equals(ResourceSpec.DEFAULT));
		assertTrue(feedbackVertex.getMinResources().equals(resource5));
		assertTrue(sinkVertex.getPreferredResources().equals(resource6));
		assertTrue(iterationSyncVertex.getMinResources().equals(resource3));
	}
@ Test public void testResourcesForDeltaIteration ( ) 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 ) ; Method opMethod = Operator . class . getDeclaredMethod ( "setResources" , ResourceSpec . class ) ; opMethod . setAccessible ( true ) ; Method deltaMethod = DeltaIteration . class . getDeclaredMethod ( "setResources" , ResourceSpec . class ) ; deltaMethod . setAccessible ( true ) ; Method sinkMethod = DataSink . class . getDeclaredMethod ( "setResources" , ResourceSpec . class ) ; sinkMethod . setAccessible ( true ) ; MapFunction < Tuple2 < Long , Long > , Tuple2 < Long , Long > > mapFunction = new MapFunction < Tuple2 < Long , Long > , Tuple2 < Long , Long > > ( ) { @ Override public Tuple2 < Long , Long > map ( Tuple2 < Long , Long > value ) throws Exception { return value ; } } ; FilterFunction < Tuple2 < Long , Long > > filterFunction = new FilterFunction < Tuple2 < Long , Long > > ( ) { @ Override public boolean filter ( Tuple2 < Long , Long > value ) throws Exception { return false ; } } ; ExecutionEnvironment env = ExecutionEnvironment . getExecutionEnvironment ( ) ; DataSet < Tuple2 < Long , Long > > input = env . fromElements ( new Tuple2 < > ( 1L , 2L ) ) ; opMethod . invoke ( input , resource1 ) ; DataSet < Tuple2 < Long , Long > > map = input . map ( mapFunction ) ; opMethod . invoke ( map , resource2 ) ; DeltaIteration < Tuple2 < Long , Long > , Tuple2 < Long , Long > > iteration = map . iterateDelta ( map , 100 , 0 ) . registerAggregator ( "test" , new LongSumAggregator ( ) ) ; deltaMethod . invoke ( iteration , resource3 ) ; DataSet < Tuple2 < Long , Long > > delta = iteration . getWorkset ( ) . map ( mapFunction ) ; opMethod . invoke ( delta , resource4 ) ; DataSet < Tuple2 < Long , Long > > feedback = delta . filter ( filterFunction ) ; opMethod . invoke ( feedback , resource5 ) ; DataSink < Tuple2 < Long , Long > > sink = iteration . closeWith ( delta , feedback ) . output ( new DiscardingOutputFormat < Tuple2 < Long , Long > > ( ) ) ; sinkMethod . invoke ( sink , resource6 ) ; 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 sourceMapVertex = jobGraph . getVerticesSortedTopologicallyFromSources ( ) . get ( 0 ) ; JobVertex iterationHeadVertex = jobGraph . getVerticesSortedTopologicallyFromSources ( ) . get ( 1 ) ; JobVertex deltaVertex = jobGraph . getVerticesSortedTopologicallyFromSources ( ) . get ( 2 ) ; JobVertex iterationTailVertex = jobGraph . getVerticesSortedTopologicallyFromSources ( ) . get ( 3 ) ; JobVertex feedbackVertex = jobGraph . getVerticesSortedTopologicallyFromSources ( ) . get ( 4 ) ; JobVertex sinkVertex = jobGraph . getVerticesSortedTopologicallyFromSources ( ) . get ( 5 ) ; JobVertex iterationSyncVertex = jobGraph . getVerticesSortedTopologicallyFromSources ( ) . get ( 6 ) ; assertTrue ( sourceMapVertex . getMinResources ( ) . equals ( resource1 . merge ( resource2 ) ) ) ; assertTrue ( iterationHeadVertex . getPreferredResources ( ) . equals ( resource3 ) ) ; assertTrue ( deltaVertex . getMinResources ( ) . equals ( resource4 ) ) ; assertTrue ( iterationTailVertex . getPreferredResources ( ) . equals ( ResourceSpec . DEFAULT ) ) ; assertTrue ( feedbackVertex . getMinResources ( ) . equals ( resource5 ) ) ; assertTrue ( sinkVertex . getPreferredResources ( ) . equals ( resource6 ) ) ; assertTrue ( iterationSyncVertex . getMinResources ( ) . equals ( resource3 ) ) ; }
Verifies that the resources are set onto each job vertex correctly when generating job graph
 which covers the delta iteration case
Verifies that the resources are set onto each job vertex correctly when generating job graph which covers the delta iteration case
// CHAIN(Map -> Filter) // the iteration tail task will be scheduled in the same instance with iteration head, and currently not set resources.