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