flink-instrumentation/workspace/flink-1.4.1-instrumented/flink-1.4.1/flink-streaming-java/src/test/java/org/apache/flink/streaming/runtime/tasks/SourceStreamTaskTest.java
/**
	 * This test ensures that the SourceStreamTask properly serializes checkpointing
	 * and element emission. This also verifies that there are no concurrent invocations
	 * of the checkpoint method on the source operator.
	 *
	 * <p>The source emits elements and performs checkpoints. We have several checkpointer threads
	 * that fire checkpoint requests at the source task.
	 *
	 * <p>If element emission and checkpointing are not in series the count of elements at the
	 * beginning of a checkpoint and at the end of a checkpoint are not the same because the
	 * source kept emitting elements while the checkpoint was ongoing.
	 */
This test ensures that the SourceStreamTask properly serializes checkpointing
and element emission. This also verifies that there are no concurrent invocations
of the checkpoint method on the source operator.

The source emits elements and performs checkpoints. We have several checkpointer threads
that fire checkpoint requests at the source task.

If element emission and checkpointing are not in series the count of elements at the
beginning of a checkpoint and at the end of a checkpoint are not the same because the
source kept emitting elements while the checkpoint was ongoing.
This test ensures that the SourceStreamTask properly serializes checkpointing and element emission . This also verifies that there are no concurrent invocations of the checkpoint method on the source operator . The source emits elements and performs checkpoints . We have several checkpointer threads that fire checkpoint requests at the source task . If element emission and checkpointing are not in series the count of elements at the beginning of a checkpoint and at the end of a checkpoint are not the same because the source kept emitting elements while the checkpoint was ongoing .
@Test
	@SuppressWarnings("unchecked")
	public void testCheckpointing() throws Exception {
		final int numElements = 100;
		final int numCheckpoints = 100;
		final int numCheckpointers = 1;
		final int checkpointInterval = 5; 
		final int sourceCheckpointDelay = 1000; 
		final int sourceReadDelay = 1; 
		ExecutorService executor = Executors.newFixedThreadPool(10);
		try {
			final TupleTypeInfo<Tuple2<Long, Integer>> typeInfo = new TupleTypeInfo<Tuple2<Long, Integer>>(BasicTypeInfo.LONG_TYPE_INFO, BasicTypeInfo.INT_TYPE_INFO);
			final SourceStreamTask<Tuple2<Long, Integer>, SourceFunction<Tuple2<Long, Integer>>,
				StreamSource<Tuple2<Long, Integer>, SourceFunction<Tuple2<Long, Integer>>>> sourceTask = new SourceStreamTask<>();
			final StreamTaskTestHarness<Tuple2<Long, Integer>> testHarness = new StreamTaskTestHarness<Tuple2<Long, Integer>>(sourceTask, typeInfo);
			testHarness.setupOutputForSingletonOperatorChain();
			StreamConfig streamConfig = testHarness.getStreamConfig();
			StreamSource<Tuple2<Long, Integer>, ?> sourceOperator = new StreamSource<>(new MockSource(numElements, sourceCheckpointDelay, sourceReadDelay));
			streamConfig.setStreamOperator(sourceOperator);
			streamConfig.setOperatorID(new OperatorID());
			Future<Boolean>[] checkpointerResults = new Future[numCheckpointers];
			testHarness.invoke();
			for (int i = 0; i < numCheckpointers; i++) {
				checkpointerResults[i] = executor.submit(new Checkpointer(numCheckpoints, checkpointInterval, sourceTask));
			}
			testHarness.waitForTaskCompletion();
			for (int i = 0; i < numCheckpointers; i++) {
				if (!checkpointerResults[i].isDone()) {
					checkpointerResults[i].cancel(true);
				}
				if (!checkpointerResults[i].isCancelled()) {
					checkpointerResults[i].get();
				}
			}
			List<Tuple2<Long, Integer>> resultElements = TestHarnessUtil.getRawElementsFromOutput(testHarness.getOutput());
			Assert.assertEquals(numElements, resultElements.size());
		}
		finally {
			executor.shutdown();
		}
	}
@ Test @ SuppressWarnings ( "unchecked" ) public void testCheckpointing ( ) throws Exception { final int numElements = 100 ; final int numCheckpoints = 100 ; final int numCheckpointers = 1 ; final int checkpointInterval = 5 ; final int sourceCheckpointDelay = 1000 ; final int sourceReadDelay = 1 ; ExecutorService executor = Executors . newFixedThreadPool ( 10 ) ; try { final TupleTypeInfo < Tuple2 < Long , Integer > > typeInfo = new TupleTypeInfo < Tuple2 < Long , Integer > > ( BasicTypeInfo . LONG_TYPE_INFO , BasicTypeInfo . INT_TYPE_INFO ) ; final SourceStreamTask < Tuple2 < Long , Integer > , SourceFunction < Tuple2 < Long , Integer > > , StreamSource < Tuple2 < Long , Integer > , SourceFunction < Tuple2 < Long , Integer > > > > sourceTask = new SourceStreamTask < > ( ) ; final StreamTaskTestHarness < Tuple2 < Long , Integer > > testHarness = new StreamTaskTestHarness < Tuple2 < Long , Integer > > ( sourceTask , typeInfo ) ; testHarness . setupOutputForSingletonOperatorChain ( ) ; StreamConfig streamConfig = testHarness . getStreamConfig ( ) ; StreamSource < Tuple2 < Long , Integer > , ? > sourceOperator = new StreamSource < > ( new MockSource ( numElements , sourceCheckpointDelay , sourceReadDelay ) ) ; streamConfig . setStreamOperator ( sourceOperator ) ; streamConfig . setOperatorID ( new OperatorID ( ) ) ; Future < Boolean > [ ] checkpointerResults = new Future [ numCheckpointers ] ; testHarness . invoke ( ) ; for ( int i = 0 ; i < numCheckpointers ; i ++ ) { checkpointerResults [ i ] = executor . submit ( new Checkpointer ( numCheckpoints , checkpointInterval , sourceTask ) ) ; } testHarness . waitForTaskCompletion ( ) ; for ( int i = 0 ; i < numCheckpointers ; i ++ ) { if ( ! checkpointerResults [ i ] . isDone ( ) ) { checkpointerResults [ i ] . cancel ( true ) ; } if ( ! checkpointerResults [ i ] . isCancelled ( ) ) { checkpointerResults [ i ] . get ( ) ; } } List < Tuple2 < Long , Integer > > resultElements = TestHarnessUtil . getRawElementsFromOutput ( testHarness . getOutput ( ) ) ; Assert . assertEquals ( numElements , resultElements . size ( ) ) ; } finally { executor . shutdown ( ) ; } }
This test ensures that the SourceStreamTask properly serializes checkpointing
 and element emission.
This test ensures that the SourceStreamTask properly serializes checkpointing and element emission .
// in ms // how many random values we sum up in storeCheckpoint // in ms // prepare the // invoke this first, so the tasks are actually running when the checkpoints are scheduled // Get the result from the checkpointers, if these threw an exception it // will be rethrown here