@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);
	// CHAIN(Map -> Filter)
	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));
	// the iteration tail task will be scheduled in the same instance with iteration head, and currently not set resources.
	assertTrue(iterationTailVertex.getPreferredResources().equals(ResourceSpec.DEFAULT));
	assertTrue(feedbackVertex.getMinResources().equals(resource5));
	assertTrue(sinkVertex.getPreferredResources().equals(resource6));
	assertTrue(iterationSyncVertex.getMinResources().equals(resource3));
}