flink-streaming-java/src/test/java/org/apache/flink/streaming/api/DataStreamTest.java
/**
	 * Tests whether resources get set.
	 */
Tests whether resources get set.
Tests whether resources get set .
@Test
	public void testResources() throws Exception{
		StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
		ResourceSpec minResource1 = new ResourceSpec(1.0, 100);
		ResourceSpec preferredResource1 = new ResourceSpec(2.0, 200);
		ResourceSpec minResource2 = new ResourceSpec(1.0, 200);
		ResourceSpec preferredResource2 = new ResourceSpec(2.0, 300);
		ResourceSpec minResource3 = new ResourceSpec(1.0, 300);
		ResourceSpec preferredResource3 = new ResourceSpec(2.0, 400);
		ResourceSpec minResource4 = new ResourceSpec(1.0, 400);
		ResourceSpec preferredResource4 = new ResourceSpec(2.0, 500);
		ResourceSpec minResource5 = new ResourceSpec(1.0, 500);
		ResourceSpec preferredResource5 = new ResourceSpec(2.0, 600);
		ResourceSpec minResource6 = new ResourceSpec(1.0, 600);
		ResourceSpec preferredResource6 = new ResourceSpec(2.0, 700);
		ResourceSpec minResource7 = new ResourceSpec(1.0, 700);
		ResourceSpec preferredResource7 = new ResourceSpec(2.0, 800);
		Method opMethod = SingleOutputStreamOperator.class.getDeclaredMethod("setResources", ResourceSpec.class, ResourceSpec.class);
		opMethod.setAccessible(true);
		Method sinkMethod = DataStreamSink.class.getDeclaredMethod("setResources", ResourceSpec.class, ResourceSpec.class);
		sinkMethod.setAccessible(true);
		DataStream<Long> source1 = env.generateSequence(0, 0);
		opMethod.invoke(source1, minResource1, preferredResource1);
		DataStream<Long> map1 = source1.map(new MapFunction<Long, Long>() {
			@Override
			public Long map(Long value) throws Exception {
				return null;
			}
		});
		opMethod.invoke(map1, minResource2, preferredResource2);
		DataStream<Long> source2 = env.generateSequence(0, 0);
		opMethod.invoke(source2, minResource3, preferredResource3);
		DataStream<Long> map2 = source2.map(new MapFunction<Long, Long>() {
			@Override
			public Long map(Long value) throws Exception {
				return null;
			}
		});
		opMethod.invoke(map2, minResource4, preferredResource4);
		DataStream<Long> connected = map1.connect(map2)
				.flatMap(new CoFlatMapFunction<Long, Long, Long>() {
					@Override
					public void flatMap1(Long value, Collector<Long> out) throws Exception {
					}
					@Override
					public void flatMap2(Long value, Collector<Long> out) throws Exception {
					}
				});
		opMethod.invoke(connected, minResource5, preferredResource5);
		DataStream<Long> windowed = connected
				.windowAll(GlobalWindows.create())
				.trigger(PurgingTrigger.of(CountTrigger.of(10)))
				.fold(0L, new FoldFunction<Long, Long>() {
					private static final long serialVersionUID = 1L;
					@Override
					public Long fold(Long accumulator, Long value) throws Exception {
						return null;
					}
				});
		opMethod.invoke(windowed, minResource6, preferredResource6);
		DataStreamSink<Long> sink = windowed.print();
		sinkMethod.invoke(sink, minResource7, preferredResource7);
		assertEquals(minResource1, env.getStreamGraph().getStreamNode(source1.getId()).getMinResources());
		assertEquals(preferredResource1, env.getStreamGraph().getStreamNode(source1.getId()).getPreferredResources());
		assertEquals(minResource2, env.getStreamGraph().getStreamNode(map1.getId()).getMinResources());
		assertEquals(preferredResource2, env.getStreamGraph().getStreamNode(map1.getId()).getPreferredResources());
		assertEquals(minResource3, env.getStreamGraph().getStreamNode(source2.getId()).getMinResources());
		assertEquals(preferredResource3, env.getStreamGraph().getStreamNode(source2.getId()).getPreferredResources());
		assertEquals(minResource4, env.getStreamGraph().getStreamNode(map2.getId()).getMinResources());
		assertEquals(preferredResource4, env.getStreamGraph().getStreamNode(map2.getId()).getPreferredResources());
		assertEquals(minResource5, env.getStreamGraph().getStreamNode(connected.getId()).getMinResources());
		assertEquals(preferredResource5, env.getStreamGraph().getStreamNode(connected.getId()).getPreferredResources());
		assertEquals(minResource6, env.getStreamGraph().getStreamNode(windowed.getId()).getMinResources());
		assertEquals(preferredResource6, env.getStreamGraph().getStreamNode(windowed.getId()).getPreferredResources());
		assertEquals(minResource7, env.getStreamGraph().getStreamNode(sink.getTransformation().getId()).getMinResources());
		assertEquals(preferredResource7, env.getStreamGraph().getStreamNode(sink.getTransformation().getId()).getPreferredResources());
	}
@ Test public void testResources ( ) throws Exception { StreamExecutionEnvironment env = StreamExecutionEnvironment . getExecutionEnvironment ( ) ; ResourceSpec minResource1 = new ResourceSpec ( 1.0 , 100 ) ; ResourceSpec preferredResource1 = new ResourceSpec ( 2.0 , 200 ) ; ResourceSpec minResource2 = new ResourceSpec ( 1.0 , 200 ) ; ResourceSpec preferredResource2 = new ResourceSpec ( 2.0 , 300 ) ; ResourceSpec minResource3 = new ResourceSpec ( 1.0 , 300 ) ; ResourceSpec preferredResource3 = new ResourceSpec ( 2.0 , 400 ) ; ResourceSpec minResource4 = new ResourceSpec ( 1.0 , 400 ) ; ResourceSpec preferredResource4 = new ResourceSpec ( 2.0 , 500 ) ; ResourceSpec minResource5 = new ResourceSpec ( 1.0 , 500 ) ; ResourceSpec preferredResource5 = new ResourceSpec ( 2.0 , 600 ) ; ResourceSpec minResource6 = new ResourceSpec ( 1.0 , 600 ) ; ResourceSpec preferredResource6 = new ResourceSpec ( 2.0 , 700 ) ; ResourceSpec minResource7 = new ResourceSpec ( 1.0 , 700 ) ; ResourceSpec preferredResource7 = new ResourceSpec ( 2.0 , 800 ) ; Method opMethod = SingleOutputStreamOperator . class . getDeclaredMethod ( "setResources" , ResourceSpec . class , ResourceSpec . class ) ; opMethod . setAccessible ( true ) ; Method sinkMethod = DataStreamSink . class . getDeclaredMethod ( "setResources" , ResourceSpec . class , ResourceSpec . class ) ; sinkMethod . setAccessible ( true ) ; DataStream < Long > source1 = env . generateSequence ( 0 , 0 ) ; opMethod . invoke ( source1 , minResource1 , preferredResource1 ) ; DataStream < Long > map1 = source1 . map ( new MapFunction < Long , Long > ( ) { @ Override public Long map ( Long value ) throws Exception { return null ; } } ) ; opMethod . invoke ( map1 , minResource2 , preferredResource2 ) ; DataStream < Long > source2 = env . generateSequence ( 0 , 0 ) ; opMethod . invoke ( source2 , minResource3 , preferredResource3 ) ; DataStream < Long > map2 = source2 . map ( new MapFunction < Long , Long > ( ) { @ Override public Long map ( Long value ) throws Exception { return null ; } } ) ; opMethod . invoke ( map2 , minResource4 , preferredResource4 ) ; DataStream < Long > connected = map1 . connect ( map2 ) . flatMap ( new CoFlatMapFunction < Long , Long , Long > ( ) { @ Override public void flatMap1 ( Long value , Collector < Long > out ) throws Exception { } @ Override public void flatMap2 ( Long value , Collector < Long > out ) throws Exception { } } ) ; opMethod . invoke ( connected , minResource5 , preferredResource5 ) ; DataStream < Long > windowed = connected . windowAll ( GlobalWindows . create ( ) ) . trigger ( PurgingTrigger . of ( CountTrigger . of ( 10 ) ) ) . fold ( 0L , new FoldFunction < Long , Long > ( ) { private static final long serialVersionUID = 1L ; @ Override public Long fold ( Long accumulator , Long value ) throws Exception { return null ; } } ) ; opMethod . invoke ( windowed , minResource6 , preferredResource6 ) ; DataStreamSink < Long > sink = windowed . print ( ) ; sinkMethod . invoke ( sink , minResource7 , preferredResource7 ) ; assertEquals ( minResource1 , env . getStreamGraph ( ) . getStreamNode ( source1 . getId ( ) ) . getMinResources ( ) ) ; assertEquals ( preferredResource1 , env . getStreamGraph ( ) . getStreamNode ( source1 . getId ( ) ) . getPreferredResources ( ) ) ; assertEquals ( minResource2 , env . getStreamGraph ( ) . getStreamNode ( map1 . getId ( ) ) . getMinResources ( ) ) ; assertEquals ( preferredResource2 , env . getStreamGraph ( ) . getStreamNode ( map1 . getId ( ) ) . getPreferredResources ( ) ) ; assertEquals ( minResource3 , env . getStreamGraph ( ) . getStreamNode ( source2 . getId ( ) ) . getMinResources ( ) ) ; assertEquals ( preferredResource3 , env . getStreamGraph ( ) . getStreamNode ( source2 . getId ( ) ) . getPreferredResources ( ) ) ; assertEquals ( minResource4 , env . getStreamGraph ( ) . getStreamNode ( map2 . getId ( ) ) . getMinResources ( ) ) ; assertEquals ( preferredResource4 , env . getStreamGraph ( ) . getStreamNode ( map2 . getId ( ) ) . getPreferredResources ( ) ) ; assertEquals ( minResource5 , env . getStreamGraph ( ) . getStreamNode ( connected . getId ( ) ) . getMinResources ( ) ) ; assertEquals ( preferredResource5 , env . getStreamGraph ( ) . getStreamNode ( connected . getId ( ) ) . getPreferredResources ( ) ) ; assertEquals ( minResource6 , env . getStreamGraph ( ) . getStreamNode ( windowed . getId ( ) ) . getMinResources ( ) ) ; assertEquals ( preferredResource6 , env . getStreamGraph ( ) . getStreamNode ( windowed . getId ( ) ) . getPreferredResources ( ) ) ; assertEquals ( minResource7 , env . getStreamGraph ( ) . getStreamNode ( sink . getTransformation ( ) . getId ( ) ) . getMinResources ( ) ) ; assertEquals ( preferredResource7 , env . getStreamGraph ( ) . getStreamNode ( sink . getTransformation ( ) . getId ( ) ) . getPreferredResources ( ) ) ; }
Tests whether resources get set.
Tests whether resources get set .