@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);
	// CHAIN(Source -> Map -> Filter)
	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);
	// CHAIN(Map -> Filter)
	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));
}