flink-instrumentation/workspace/flink-1.4.1-instrumented/flink-1.4.1/flink-contrib/flink-streaming-contrib/src/test/java/org/apache/flink/contrib/streaming/ReinterpretAsKeyedStreamITCase.java
/**
	 * This test checks that reinterpreting a data stream to a keyed stream works as expected. This test consists of
	 * two jobs. The first job materializes a keyBy into files, one files per partition. The second job opens the
	 * files created by the first jobs as sources (doing the correct assignment of files to partitions) and
	 * reinterprets the sources as keyed, because we know they have been partitioned in a keyBy from the first job.
	 */
This test checks that reinterpreting a data stream to a keyed stream works as expected. This test consists of
two jobs. The first job materializes a keyBy into files, one files per partition. The second job opens the
files created by the first jobs as sources (doing the correct assignment of files to partitions) and
reinterprets the sources as keyed, because we know they have been partitioned in a keyBy from the first job.
This test checks that reinterpreting a data stream to a keyed stream works as expected . This test consists of two jobs . The first job materializes a keyBy into files one files per partition . The second job opens the files created by the first jobs as sources ( doing the correct assignment of files to partitions ) and reinterprets the sources as keyed because we know they have been partitioned in a keyBy from the first job .
@Test
	public void testReinterpretAsKeyedStream() throws Exception {
		final int numEventsPerInstance = 100;
		final int maxParallelism = 8;
		final int parallelism = 3;
		final int numUniqueKeys = 12;
		final StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
		env.setStreamTimeCharacteristic(TimeCharacteristic.IngestionTime);
		env.setMaxParallelism(maxParallelism);
		env.setParallelism(parallelism);
		final List<File> partitionFiles = new ArrayList<>(parallelism);
		for (int i = 0; i < parallelism; ++i) {
			File partitionFile = temporaryFolder.newFile();
			partitionFiles.add(i, partitionFile);
		}
		env.addSource(new RandomTupleSource(numEventsPerInstance, numUniqueKeys))
			.keyBy(0)
			.addSink(new ToPartitionFileSink(partitionFiles));
		env.execute();
		DataStreamUtils.reinterpretAsKeyedStream(
			env.addSource(new FromPartitionFileSource(partitionFiles)),
			(KeySelector<Tuple2<Integer, Integer>, Integer>) value -> value.f0,
			TypeInformation.of(Integer.class))
			.timeWindow(Time.seconds(1)) 
			.reduce((ReduceFunction<Tuple2<Integer, Integer>>) (value1, value2) ->
				new Tuple2<>(value1.f0, value1.f1 + value2.f1))
			.addSink(new ValidatingSink(numEventsPerInstance * parallelism)).setParallelism(1);
		env.execute();
	}
@ Test public void testReinterpretAsKeyedStream ( ) throws Exception { final int numEventsPerInstance = 100 ; final int maxParallelism = 8 ; final int parallelism = 3 ; final int numUniqueKeys = 12 ; final StreamExecutionEnvironment env = StreamExecutionEnvironment . getExecutionEnvironment ( ) ; env . setStreamTimeCharacteristic ( TimeCharacteristic . IngestionTime ) ; env . setMaxParallelism ( maxParallelism ) ; env . setParallelism ( parallelism ) ; final List < File > partitionFiles = new ArrayList < > ( parallelism ) ; for ( int i = 0 ; i < parallelism ; ++ i ) { File partitionFile = temporaryFolder . newFile ( ) ; partitionFiles . add ( i , partitionFile ) ; } env . addSource ( new RandomTupleSource ( numEventsPerInstance , numUniqueKeys ) ) . keyBy ( 0 ) . addSink ( new ToPartitionFileSink ( partitionFiles ) ) ; env . execute ( ) ; DataStreamUtils . reinterpretAsKeyedStream ( env . addSource ( new FromPartitionFileSource ( partitionFiles ) ) , ( KeySelector < Tuple2 < Integer , Integer > , Integer > ) value -> value . f0 , TypeInformation . of ( Integer . class ) ) . timeWindow ( Time . seconds ( 1 ) ) . reduce ( ( ReduceFunction < Tuple2 < Integer , Integer > > ) ( value1 , value2 ) -> new Tuple2 < > ( value1 . f0 , value1 . f1 + value2 . f1 ) ) . addSink ( new ValidatingSink ( numEventsPerInstance * parallelism ) ) . setParallelism ( 1 ) ; env . execute ( ) ; }
This test checks that reinterpreting a data stream to a keyed stream works as expected.
This test checks that reinterpreting a data stream to a keyed stream works as expected .
// test that also timers and aggregated state work as expected