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 are correctly forwarded.
	 */
This test verifies that checkpoint barriers are correctly forwarded.
This test verifies that checkpoint barriers are correctly forwarded .
@Test
	public void testCheckpointBarriers() 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(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();
		expectedOutput.add(new CheckpointBarrier(0, 0, CheckpointOptions.forCheckpoint()));
		expectedOutput.add(new StreamRecord<String>("Hello-0-0", initialTime));
		expectedOutput.add(new StreamRecord<String>("Ciao-0-0", initialTime));
		testHarness.endInput();
		testHarness.waitForTaskCompletion();
		TestHarnessUtil.assertOutputEquals("Output was not correct.", expectedOutput, testHarness.getOutput());
	}
@ Test public void testCheckpointBarriers ( ) 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 ( 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 ( ) ; expectedOutput . add ( new CheckpointBarrier ( 0 , 0 , CheckpointOptions . forCheckpoint ( ) ) ) ; expectedOutput . add ( new StreamRecord < String > ( "Hello-0-0" , initialTime ) ) ; expectedOutput . add ( new StreamRecord < String > ( "Ciao-0-0" , initialTime ) ) ; testHarness . endInput ( ) ; testHarness . waitForTaskCompletion ( ) ; TestHarnessUtil . assertOutputEquals ( "Output was not correct." , expectedOutput , testHarness . getOutput ( ) ) ; }
This test verifies that checkpoint barriers are correctly forwarded.
This test verifies that checkpoint barriers are correctly forwarded .
// 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 we should see the barrier and after that the buffered elements