flink-instrumentation/workspace/flink-1.4.1-instrumented/flink-1.4.1/flink-streaming-java/src/test/java/org/apache/flink/streaming/runtime/tasks/OneInputStreamTaskTest.java
/**
	 * This test verifies that checkpoint barriers and barrier buffers work correctly with
	 * concurrent checkpoint barriers where one checkpoint is "overtaking" another checkpoint, i.e.
	 * some inputs receive barriers from an earlier checkpoint, thereby blocking,
	 * then all inputs receive barriers from a later checkpoint.
	 */
This test verifies that checkpoint barriers and barrier buffers work correctly with
concurrent checkpoint barriers where one checkpoint is "overtaking" another checkpoint, i.e.
some inputs receive barriers from an earlier checkpoint, thereby blocking,
then all inputs receive barriers from a later checkpoint.
This test verifies that checkpoint barriers and barrier buffers work correctly with concurrent checkpoint barriers where one checkpoint is " overtaking " another checkpoint i . e . some inputs receive barriers from an earlier checkpoint thereby blocking then all inputs receive barriers from a later checkpoint .
@Test
	public void testOvertakingCheckpointBarriers() throws Exception {
		final OneInputStreamTask<String, String> mapTask = new OneInputStreamTask<String, String>();
		final OneInputStreamTaskTestHarness<String, String> testHarness = new OneInputStreamTaskTestHarness<String, String>(mapTask, 2, 2, BasicTypeInfo.STRING_TYPE_INFO, BasicTypeInfo.STRING_TYPE_INFO);
		testHarness.setupOutputForSingletonOperatorChain();
		StreamConfig streamConfig = testHarness.getStreamConfig();
		StreamMap<String, String> mapOperator = new StreamMap<String, String>(new IdentityMap());
		streamConfig.setStreamOperator(mapOperator);
		streamConfig.setOperatorID(new OperatorID());
		ConcurrentLinkedQueue<Object> expectedOutput = new ConcurrentLinkedQueue<Object>();
		long initialTime = 0L;
		testHarness.invoke();
		testHarness.waitForTaskRunning();
		testHarness.processEvent(new CheckpointBarrier(0, 0, CheckpointOptions.forCheckpoint()), 0, 0);
		testHarness.processElement(new StreamRecord<String>("Hello-0-0", initialTime), 0, 0);
		testHarness.processElement(new StreamRecord<String>("Ciao-0-0", initialTime), 0, 0);
		testHarness.processElement(new StreamRecord<String>("Hello-1-1", initialTime), 1, 1);
		testHarness.processElement(new StreamRecord<String>("Ciao-1-1", initialTime), 1, 1);
		expectedOutput.add(new StreamRecord<String>("Hello-1-1", initialTime));
		expectedOutput.add(new StreamRecord<String>("Ciao-1-1", initialTime));
		testHarness.waitForInputProcessing();
		TestHarnessUtil.assertOutputEquals("Output was not correct.", expectedOutput, testHarness.getOutput());
		testHarness.processEvent(new CheckpointBarrier(1, 1, CheckpointOptions.forCheckpoint()), 0, 0);
		testHarness.processEvent(new CheckpointBarrier(1, 1, CheckpointOptions.forCheckpoint()), 0, 1);
		testHarness.processEvent(new CheckpointBarrier(1, 1, CheckpointOptions.forCheckpoint()), 1, 0);
		testHarness.processEvent(new CheckpointBarrier(1, 1, CheckpointOptions.forCheckpoint()), 1, 1);
		expectedOutput.add(new CancelCheckpointMarker(0));
		expectedOutput.add(new StreamRecord<String>("Hello-0-0", initialTime));
		expectedOutput.add(new StreamRecord<String>("Ciao-0-0", initialTime));
		expectedOutput.add(new CheckpointBarrier(1, 1, CheckpointOptions.forCheckpoint()));
		testHarness.waitForInputProcessing();
		TestHarnessUtil.assertOutputEquals("Output was not correct.", expectedOutput, testHarness.getOutput());
		testHarness.processEvent(new CheckpointBarrier(0, 0, CheckpointOptions.forCheckpoint()), 0, 1);
		testHarness.processEvent(new CheckpointBarrier(0, 0, CheckpointOptions.forCheckpoint()), 1, 0);
		testHarness.processEvent(new CheckpointBarrier(0, 0, CheckpointOptions.forCheckpoint()), 1, 1);
		testHarness.waitForInputProcessing();
		testHarness.endInput();
		testHarness.waitForTaskCompletion();
		TestHarnessUtil.assertOutputEquals("Output was not correct.", expectedOutput, testHarness.getOutput());
	}
@ Test public void testOvertakingCheckpointBarriers ( ) throws Exception { final OneInputStreamTask < String , String > mapTask = new OneInputStreamTask < String , String > ( ) ; final OneInputStreamTaskTestHarness < String , String > testHarness = new OneInputStreamTaskTestHarness < String , String > ( mapTask , 2 , 2 , BasicTypeInfo . STRING_TYPE_INFO , BasicTypeInfo . STRING_TYPE_INFO ) ; testHarness . setupOutputForSingletonOperatorChain ( ) ; StreamConfig streamConfig = testHarness . getStreamConfig ( ) ; StreamMap < String , String > mapOperator = new StreamMap < String , String > ( new IdentityMap ( ) ) ; streamConfig . setStreamOperator ( mapOperator ) ; streamConfig . setOperatorID ( new OperatorID ( ) ) ; ConcurrentLinkedQueue < Object > expectedOutput = new ConcurrentLinkedQueue < Object > ( ) ; long initialTime = 0L ; testHarness . invoke ( ) ; testHarness . waitForTaskRunning ( ) ; testHarness . processEvent ( new CheckpointBarrier ( 0 , 0 , CheckpointOptions . forCheckpoint ( ) ) , 0 , 0 ) ; testHarness . processElement ( new StreamRecord < String > ( "Hello-0-0" , initialTime ) , 0 , 0 ) ; testHarness . processElement ( new StreamRecord < String > ( "Ciao-0-0" , initialTime ) , 0 , 0 ) ; testHarness . processElement ( new StreamRecord < String > ( "Hello-1-1" , initialTime ) , 1 , 1 ) ; testHarness . processElement ( new StreamRecord < String > ( "Ciao-1-1" , initialTime ) , 1 , 1 ) ; expectedOutput . add ( new StreamRecord < String > ( "Hello-1-1" , initialTime ) ) ; expectedOutput . add ( new StreamRecord < String > ( "Ciao-1-1" , initialTime ) ) ; testHarness . waitForInputProcessing ( ) ; TestHarnessUtil . assertOutputEquals ( "Output was not correct." , expectedOutput , testHarness . getOutput ( ) ) ; testHarness . processEvent ( new CheckpointBarrier ( 1 , 1 , CheckpointOptions . forCheckpoint ( ) ) , 0 , 0 ) ; testHarness . processEvent ( new CheckpointBarrier ( 1 , 1 , CheckpointOptions . forCheckpoint ( ) ) , 0 , 1 ) ; testHarness . processEvent ( new CheckpointBarrier ( 1 , 1 , CheckpointOptions . forCheckpoint ( ) ) , 1 , 0 ) ; testHarness . processEvent ( new CheckpointBarrier ( 1 , 1 , CheckpointOptions . forCheckpoint ( ) ) , 1 , 1 ) ; expectedOutput . add ( new CancelCheckpointMarker ( 0 ) ) ; expectedOutput . add ( new StreamRecord < String > ( "Hello-0-0" , initialTime ) ) ; expectedOutput . add ( new StreamRecord < String > ( "Ciao-0-0" , initialTime ) ) ; expectedOutput . add ( new CheckpointBarrier ( 1 , 1 , CheckpointOptions . forCheckpoint ( ) ) ) ; testHarness . waitForInputProcessing ( ) ; TestHarnessUtil . assertOutputEquals ( "Output was not correct." , expectedOutput , testHarness . getOutput ( ) ) ; testHarness . processEvent ( new CheckpointBarrier ( 0 , 0 , CheckpointOptions . forCheckpoint ( ) ) , 0 , 1 ) ; testHarness . processEvent ( new CheckpointBarrier ( 0 , 0 , CheckpointOptions . forCheckpoint ( ) ) , 1 , 0 ) ; testHarness . processEvent ( new CheckpointBarrier ( 0 , 0 , CheckpointOptions . forCheckpoint ( ) ) , 1 , 1 ) ; testHarness . waitForInputProcessing ( ) ; testHarness . endInput ( ) ; testHarness . waitForTaskCompletion ( ) ; TestHarnessUtil . assertOutputEquals ( "Output was not correct." , expectedOutput , testHarness . getOutput ( ) ) ; }
This test verifies that checkpoint barriers and barrier buffers work correctly with
 concurrent checkpoint barriers where one checkpoint is "overtaking" another checkpoint, i.e.
This test verifies that checkpoint barriers and barrier buffers work correctly with concurrent checkpoint barriers where one checkpoint is " overtaking " another checkpoint i . e .
// These elements should be buffered until we receive barriers from // all inputs // These elements should be forwarded, since we did not yet receive a checkpoint barrier // on that input, only add to same input, otherwise we would not know the ordering // of the output since the Task might read the inputs in any order // we should not yet see the barrier, only the two elements from non-blocked input // Now give a later barrier to all inputs, this should unblock the first channel, // thereby allowing the two blocked elements through // Then give the earlier barrier, these should be ignored