flink-streaming-java/src/test/java/org/apache/flink/streaming/api/DataStreamTest.java
/**
	 * Tests union functionality. This ensures that self-unions and unions of streams
	 * with differing parallelism work.
	 *
	 * @throws Exception
	 */
Tests union functionality. This ensures that self-unions and unions of streams
with differing parallelism work.

@throws Exception
Tests union functionality . This ensures that self - unions and unions of streams with differing parallelism work . @throws Exception
@Test
	public void testUnion() throws Exception {
		StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
		env.setParallelism(4);
		DataStream<Long> input1 = env.generateSequence(0, 0)
			.map(new MapFunction<Long, Long>() {
				@Override
				public Long map(Long value) throws Exception {
					return null;
				}
			});
		DataStream<Long> selfUnion = input1.union(input1).map(new MapFunction<Long, Long>() {
			@Override
			public Long map(Long value) throws Exception {
				return null;
			}
		});
		DataStream<Long> input6 = env.generateSequence(0, 0)
			.map(new MapFunction<Long, Long>() {
				@Override
				public Long map(Long value) throws Exception {
					return null;
				}
			});
		DataStream<Long> selfUnionDifferentPartition = input6.broadcast().union(input6).map(new MapFunction<Long, Long>() {
			@Override
			public Long map(Long value) throws Exception {
				return null;
			}
		});
		DataStream<Long> input2 = env.generateSequence(0, 0)
			.map(new MapFunction<Long, Long>() {
				@Override
				public Long map(Long value) throws Exception {
					return null;
				}
			}).setParallelism(4);
		DataStream<Long> input3 = env.generateSequence(0, 0)
			.map(new MapFunction<Long, Long>() {
				@Override
				public Long map(Long value) throws Exception {
					return null;
				}
			}).setParallelism(2);
		DataStream<Long> unionDifferingParallelism = input2.union(input3).map(new MapFunction<Long, Long>() {
			@Override
			public Long map(Long value) throws Exception {
				return null;
			}
		}).setParallelism(4);
		DataStream<Long> input4 = env.generateSequence(0, 0)
			.map(new MapFunction<Long, Long>() {
				@Override
				public Long map(Long value) throws Exception {
					return null;
				}
			}).setParallelism(2);
		DataStream<Long> input5 = env.generateSequence(0, 0)
			.map(new MapFunction<Long, Long>() {
				@Override
				public Long map(Long value) throws Exception {
					return null;
				}
			}).setParallelism(4);
		DataStream<Long> unionDifferingPartitioning = input4.broadcast().union(input5).map(new MapFunction<Long, Long>() {
			@Override
			public Long map(Long value) throws Exception {
				return null;
			}
		}).setParallelism(4);
		StreamGraph streamGraph = env.getStreamGraph();
		assertTrue(streamGraph.getStreamNode(selfUnion.getId()).getInEdges().size() == 2);
		for (StreamEdge edge: streamGraph.getStreamNode(selfUnion.getId()).getInEdges()) {
			assertTrue(edge.getPartitioner() instanceof ForwardPartitioner);
		}
		 with differnt partitioners
		assertTrue(streamGraph.getStreamNode(selfUnionDifferentPartition.getId()).getInEdges().size() == 2);
		boolean hasForward = false;
		boolean hasBroadcast = false;
		for (StreamEdge edge: streamGraph.getStreamNode(selfUnionDifferentPartition.getId()).getInEdges()) {
			if (edge.getPartitioner() instanceof ForwardPartitioner) {
				hasForward = true;
			}
			if (edge.getPartitioner() instanceof BroadcastPartitioner) {
				hasBroadcast = true;
			}
		}
		assertTrue(hasForward && hasBroadcast);
		assertTrue(streamGraph.getStreamNode(unionDifferingParallelism.getId()).getInEdges().size() == 2);
		for (StreamEdge edge: streamGraph.getStreamNode(unionDifferingParallelism.getId()).getInEdges()) {
			if (edge.getSourceId() == input2.getId()) {
				assertTrue(edge.getPartitioner() instanceof ForwardPartitioner);
			} else if (edge.getSourceId() == input3.getId()) {
				assertTrue(edge.getPartitioner() instanceof RebalancePartitioner);
			} else {
				fail("Wrong input edge.");
			}
		}
		assertTrue(streamGraph.getStreamNode(unionDifferingPartitioning.getId()).getInEdges().size() == 2);
		for (StreamEdge edge: streamGraph.getStreamNode(unionDifferingPartitioning.getId()).getInEdges()) {
			if (edge.getSourceId() == input4.getId()) {
				assertTrue(edge.getPartitioner() instanceof BroadcastPartitioner);
			} else if (edge.getSourceId() == input5.getId()) {
				assertTrue(edge.getPartitioner() instanceof ForwardPartitioner);
			} else {
				fail("Wrong input edge.");
			}
		}
	}
@ Test public void testUnion ( ) throws Exception { StreamExecutionEnvironment env = StreamExecutionEnvironment . getExecutionEnvironment ( ) ; env . setParallelism ( 4 ) ; DataStream < Long > input1 = env . generateSequence ( 0 , 0 ) . map ( new MapFunction < Long , Long > ( ) { @ Override public Long map ( Long value ) throws Exception { return null ; } } ) ; DataStream < Long > selfUnion = input1 . union ( input1 ) . map ( new MapFunction < Long , Long > ( ) { @ Override public Long map ( Long value ) throws Exception { return null ; } } ) ; DataStream < Long > input6 = env . generateSequence ( 0 , 0 ) . map ( new MapFunction < Long , Long > ( ) { @ Override public Long map ( Long value ) throws Exception { return null ; } } ) ; DataStream < Long > selfUnionDifferentPartition = input6 . broadcast ( ) . union ( input6 ) . map ( new MapFunction < Long , Long > ( ) { @ Override public Long map ( Long value ) throws Exception { return null ; } } ) ; DataStream < Long > input2 = env . generateSequence ( 0 , 0 ) . map ( new MapFunction < Long , Long > ( ) { @ Override public Long map ( Long value ) throws Exception { return null ; } } ) . setParallelism ( 4 ) ; DataStream < Long > input3 = env . generateSequence ( 0 , 0 ) . map ( new MapFunction < Long , Long > ( ) { @ Override public Long map ( Long value ) throws Exception { return null ; } } ) . setParallelism ( 2 ) ; DataStream < Long > unionDifferingParallelism = input2 . union ( input3 ) . map ( new MapFunction < Long , Long > ( ) { @ Override public Long map ( Long value ) throws Exception { return null ; } } ) . setParallelism ( 4 ) ; DataStream < Long > input4 = env . generateSequence ( 0 , 0 ) . map ( new MapFunction < Long , Long > ( ) { @ Override public Long map ( Long value ) throws Exception { return null ; } } ) . setParallelism ( 2 ) ; DataStream < Long > input5 = env . generateSequence ( 0 , 0 ) . map ( new MapFunction < Long , Long > ( ) { @ Override public Long map ( Long value ) throws Exception { return null ; } } ) . setParallelism ( 4 ) ; DataStream < Long > unionDifferingPartitioning = input4 . broadcast ( ) . union ( input5 ) . map ( new MapFunction < Long , Long > ( ) { @ Override public Long map ( Long value ) throws Exception { return null ; } } ) . setParallelism ( 4 ) ; StreamGraph streamGraph = env . getStreamGraph ( ) ; assertTrue ( streamGraph . getStreamNode ( selfUnion . getId ( ) ) . getInEdges ( ) . size ( ) == 2 ) ; for ( StreamEdge edge : streamGraph . getStreamNode ( selfUnion . getId ( ) ) . getInEdges ( ) ) { assertTrue ( edge . getPartitioner ( ) instanceof ForwardPartitioner ) ; } assertTrue ( streamGraph . getStreamNode ( selfUnionDifferentPartition . getId ( ) ) . getInEdges ( ) . size ( ) == 2 ) ; boolean hasForward = false ; boolean hasBroadcast = false ; for ( StreamEdge edge : streamGraph . getStreamNode ( selfUnionDifferentPartition . getId ( ) ) . getInEdges ( ) ) { if ( edge . getPartitioner ( ) instanceof ForwardPartitioner ) { hasForward = true ; } if ( edge . getPartitioner ( ) instanceof BroadcastPartitioner ) { hasBroadcast = true ; } } assertTrue ( hasForward && hasBroadcast ) ; assertTrue ( streamGraph . getStreamNode ( unionDifferingParallelism . getId ( ) ) . getInEdges ( ) . size ( ) == 2 ) ; for ( StreamEdge edge : streamGraph . getStreamNode ( unionDifferingParallelism . getId ( ) ) . getInEdges ( ) ) { if ( edge . getSourceId ( ) == input2 . getId ( ) ) { assertTrue ( edge . getPartitioner ( ) instanceof ForwardPartitioner ) ; } else if ( edge . getSourceId ( ) == input3 . getId ( ) ) { assertTrue ( edge . getPartitioner ( ) instanceof RebalancePartitioner ) ; } else { fail ( "Wrong input edge." ) ; } } assertTrue ( streamGraph . getStreamNode ( unionDifferingPartitioning . getId ( ) ) . getInEdges ( ) . size ( ) == 2 ) ; for ( StreamEdge edge : streamGraph . getStreamNode ( unionDifferingPartitioning . getId ( ) ) . getInEdges ( ) ) { if ( edge . getSourceId ( ) == input4 . getId ( ) ) { assertTrue ( edge . getPartitioner ( ) instanceof BroadcastPartitioner ) ; } else if ( edge . getSourceId ( ) == input5 . getId ( ) ) { assertTrue ( edge . getPartitioner ( ) instanceof ForwardPartitioner ) ; } else { fail ( "Wrong input edge." ) ; } } }
Tests union functionality.
Tests union functionality .
// verify self union // verify self union with differnt partitioners // verify union of streams with differing parallelism // verify union of streams with differing partitionings