flink-instrumentation/workspace/flink-1.4.1-instrumented/flink-1.4.1/flink-streaming-java/src/test/java/org/apache/flink/streaming/api/graph/StreamingJobGraphGeneratorTest.java
/**
	 * Verifies that the chain start/end is correctly set.
	 */
Verifies that the chain start/end is correctly set.
Verifies that the chain start / end is correctly set .
@Test
	public void testChainStartEndSetting() throws Exception {
		StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
		env.fromElements(1, 2, 3)
			.map(new MapFunction<Integer, Integer>() {
				@Override
				public Integer map(Integer value) throws Exception {
					return value;
				}
			})
			.print();
		JobGraph jobGraph = StreamingJobGraphGenerator.createJobGraph(env.getStreamGraph());
		List<JobVertex> verticesSorted = jobGraph.getVerticesSortedTopologicallyFromSources();
		JobVertex sourceVertex = verticesSorted.get(0);
		JobVertex mapPrintVertex = verticesSorted.get(1);
		assertEquals(ResultPartitionType.PIPELINED_BOUNDED, sourceVertex.getProducedDataSets().get(0).getResultType());
		assertEquals(ResultPartitionType.PIPELINED_BOUNDED, mapPrintVertex.getInputs().get(0).getSource().getResultType());
		StreamConfig sourceConfig = new StreamConfig(sourceVertex.getConfiguration());
		StreamConfig mapConfig = new StreamConfig(mapPrintVertex.getConfiguration());
		Map<Integer, StreamConfig> chainedConfigs = mapConfig.getTransitiveChainedTaskConfigs(getClass().getClassLoader());
		StreamConfig printConfig = chainedConfigs.values().iterator().next();
		assertTrue(sourceConfig.isChainStart());
		assertTrue(sourceConfig.isChainEnd());
		assertTrue(mapConfig.isChainStart());
		assertFalse(mapConfig.isChainEnd());
		assertFalse(printConfig.isChainStart());
		assertTrue(printConfig.isChainEnd());
	}
@ Test public void testChainStartEndSetting ( ) throws Exception { StreamExecutionEnvironment env = StreamExecutionEnvironment . getExecutionEnvironment ( ) ; env . fromElements ( 1 , 2 , 3 ) . map ( new MapFunction < Integer , Integer > ( ) { @ Override public Integer map ( Integer value ) throws Exception { return value ; } } ) . print ( ) ; JobGraph jobGraph = StreamingJobGraphGenerator . createJobGraph ( env . getStreamGraph ( ) ) ; List < JobVertex > verticesSorted = jobGraph . getVerticesSortedTopologicallyFromSources ( ) ; JobVertex sourceVertex = verticesSorted . get ( 0 ) ; JobVertex mapPrintVertex = verticesSorted . get ( 1 ) ; assertEquals ( ResultPartitionType . PIPELINED_BOUNDED , sourceVertex . getProducedDataSets ( ) . get ( 0 ) . getResultType ( ) ) ; assertEquals ( ResultPartitionType . PIPELINED_BOUNDED , mapPrintVertex . getInputs ( ) . get ( 0 ) . getSource ( ) . getResultType ( ) ) ; StreamConfig sourceConfig = new StreamConfig ( sourceVertex . getConfiguration ( ) ) ; StreamConfig mapConfig = new StreamConfig ( mapPrintVertex . getConfiguration ( ) ) ; Map < Integer , StreamConfig > chainedConfigs = mapConfig . getTransitiveChainedTaskConfigs ( getClass ( ) . getClassLoader ( ) ) ; StreamConfig printConfig = chainedConfigs . values ( ) . iterator ( ) . next ( ) ; assertTrue ( sourceConfig . isChainStart ( ) ) ; assertTrue ( sourceConfig . isChainEnd ( ) ) ; assertTrue ( mapConfig . isChainStart ( ) ) ; assertFalse ( mapConfig . isChainEnd ( ) ) ; assertFalse ( printConfig . isChainStart ( ) ) ; assertTrue ( printConfig . isChainEnd ( ) ) ; }
Verifies that the chain start/end is correctly set.
Verifies that the chain start / end is correctly set .
// fromElements -> CHAIN(Map -> Print)