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 watermarks and stream statuses are correctly forwarded. This also checks whether
	 * watermarks are forwarded only when we have received watermarks from all inputs. The
	 * forwarded watermark must be the minimum of the watermarks of all active inputs.
	 */
This test verifies that watermarks and stream statuses are correctly forwarded. This also checks whether
watermarks are forwarded only when we have received watermarks from all inputs. The
forwarded watermark must be the minimum of the watermarks of all active inputs.
This test verifies that watermarks and stream statuses are correctly forwarded . This also checks whether watermarks are forwarded only when we have received watermarks from all inputs . The forwarded watermark must be the minimum of the watermarks of all active inputs .
@Test
	@SuppressWarnings("unchecked")
	public void testWatermarkAndStreamStatusForwarding() 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.processElement(new Watermark(initialTime), 0, 0);
		testHarness.processElement(new Watermark(initialTime), 0, 1);
		testHarness.processElement(new Watermark(initialTime), 1, 0);
		testHarness.waitForInputProcessing();
		TestHarnessUtil.assertOutputEquals("Output was not correct.", expectedOutput, testHarness.getOutput());
		testHarness.processElement(new Watermark(initialTime), 1, 1);
		testHarness.waitForInputProcessing();
		expectedOutput.add(new Watermark(initialTime));
		TestHarnessUtil.assertOutputEquals("Output was not correct.",
			expectedOutput,
			testHarness.getOutput());
		testHarness.processElement(new StreamRecord<String>("Hello", initialTime));
		testHarness.processElement(new StreamRecord<String>("Ciao", initialTime));
		expectedOutput.add(new StreamRecord<String>("Hello", initialTime));
		expectedOutput.add(new StreamRecord<String>("Ciao", initialTime));
		testHarness.processElement(new Watermark(initialTime + 4), 0, 0);
		testHarness.processElement(new Watermark(initialTime + 3), 0, 1);
		testHarness.processElement(new Watermark(initialTime + 3), 1, 0);
		testHarness.processElement(new Watermark(initialTime + 2), 1, 1);
		testHarness.waitForInputProcessing();
		expectedOutput.add(new Watermark(initialTime + 2));
		TestHarnessUtil.assertOutputEquals("Output was not correct.", expectedOutput, testHarness.getOutput());
		testHarness.processElement(new Watermark(initialTime + 4), 1, 1);
		testHarness.waitForInputProcessing();
		expectedOutput.add(new Watermark(initialTime + 3));
		TestHarnessUtil.assertOutputEquals("Output was not correct.", expectedOutput, testHarness.getOutput());
		 again
		testHarness.processElement(new Watermark(initialTime + 4), 0, 1);
		testHarness.processElement(new Watermark(initialTime + 4), 1, 0);
		testHarness.waitForInputProcessing();
		expectedOutput.add(new Watermark(initialTime + 4));
		TestHarnessUtil.assertOutputEquals("Output was not correct.", expectedOutput, testHarness.getOutput());
		testHarness.processElement(StreamStatus.IDLE, 0, 1);
		testHarness.processElement(StreamStatus.IDLE, 1, 0);
		testHarness.processElement(new Watermark(initialTime + 6), 0, 0);
		testHarness.processElement(new Watermark(initialTime + 5), 1, 1); 
		testHarness.processElement(StreamStatus.IDLE, 1, 1); 
		testHarness.waitForInputProcessing();
		expectedOutput.add(new Watermark(initialTime + 5));
		expectedOutput.add(new Watermark(initialTime + 6));
		TestHarnessUtil.assertOutputEquals("Output was not correct.", expectedOutput, testHarness.getOutput());
		testHarness.processElement(StreamStatus.IDLE, 0, 0);
		testHarness.waitForInputProcessing();
		expectedOutput.add(StreamStatus.IDLE);
		TestHarnessUtil.assertOutputEquals("Output was not correct.", expectedOutput, testHarness.getOutput());
		testHarness.processElement(StreamStatus.ACTIVE, 1, 0);
		testHarness.processElement(StreamStatus.ACTIVE, 0, 1);
		testHarness.waitForInputProcessing();
		expectedOutput.add(StreamStatus.ACTIVE);
		TestHarnessUtil.assertOutputEquals("Output was not correct.", expectedOutput, testHarness.getOutput());
		testHarness.endInput();
		testHarness.waitForTaskCompletion();
		List<String> resultElements = TestHarnessUtil.getRawElementsFromOutput(testHarness.getOutput());
		assertEquals(2, resultElements.size());
	}
@ Test @ SuppressWarnings ( "unchecked" ) public void testWatermarkAndStreamStatusForwarding ( ) 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 . processElement ( new Watermark ( initialTime ) , 0 , 0 ) ; testHarness . processElement ( new Watermark ( initialTime ) , 0 , 1 ) ; testHarness . processElement ( new Watermark ( initialTime ) , 1 , 0 ) ; testHarness . waitForInputProcessing ( ) ; TestHarnessUtil . assertOutputEquals ( "Output was not correct." , expectedOutput , testHarness . getOutput ( ) ) ; testHarness . processElement ( new Watermark ( initialTime ) , 1 , 1 ) ; testHarness . waitForInputProcessing ( ) ; expectedOutput . add ( new Watermark ( initialTime ) ) ; TestHarnessUtil . assertOutputEquals ( "Output was not correct." , expectedOutput , testHarness . getOutput ( ) ) ; testHarness . processElement ( new StreamRecord < String > ( "Hello" , initialTime ) ) ; testHarness . processElement ( new StreamRecord < String > ( "Ciao" , initialTime ) ) ; expectedOutput . add ( new StreamRecord < String > ( "Hello" , initialTime ) ) ; expectedOutput . add ( new StreamRecord < String > ( "Ciao" , initialTime ) ) ; testHarness . processElement ( new Watermark ( initialTime + 4 ) , 0 , 0 ) ; testHarness . processElement ( new Watermark ( initialTime + 3 ) , 0 , 1 ) ; testHarness . processElement ( new Watermark ( initialTime + 3 ) , 1 , 0 ) ; testHarness . processElement ( new Watermark ( initialTime + 2 ) , 1 , 1 ) ; testHarness . waitForInputProcessing ( ) ; expectedOutput . add ( new Watermark ( initialTime + 2 ) ) ; TestHarnessUtil . assertOutputEquals ( "Output was not correct." , expectedOutput , testHarness . getOutput ( ) ) ; testHarness . processElement ( new Watermark ( initialTime + 4 ) , 1 , 1 ) ; testHarness . waitForInputProcessing ( ) ; expectedOutput . add ( new Watermark ( initialTime + 3 ) ) ; TestHarnessUtil . assertOutputEquals ( "Output was not correct." , expectedOutput , testHarness . getOutput ( ) ) ; testHarness . processElement ( new Watermark ( initialTime + 4 ) , 0 , 1 ) ; testHarness . processElement ( new Watermark ( initialTime + 4 ) , 1 , 0 ) ; testHarness . waitForInputProcessing ( ) ; expectedOutput . add ( new Watermark ( initialTime + 4 ) ) ; TestHarnessUtil . assertOutputEquals ( "Output was not correct." , expectedOutput , testHarness . getOutput ( ) ) ; testHarness . processElement ( StreamStatus . IDLE , 0 , 1 ) ; testHarness . processElement ( StreamStatus . IDLE , 1 , 0 ) ; testHarness . processElement ( new Watermark ( initialTime + 6 ) , 0 , 0 ) ; testHarness . processElement ( new Watermark ( initialTime + 5 ) , 1 , 1 ) ; testHarness . processElement ( StreamStatus . IDLE , 1 , 1 ) ; testHarness . waitForInputProcessing ( ) ; expectedOutput . add ( new Watermark ( initialTime + 5 ) ) ; expectedOutput . add ( new Watermark ( initialTime + 6 ) ) ; TestHarnessUtil . assertOutputEquals ( "Output was not correct." , expectedOutput , testHarness . getOutput ( ) ) ; testHarness . processElement ( StreamStatus . IDLE , 0 , 0 ) ; testHarness . waitForInputProcessing ( ) ; expectedOutput . add ( StreamStatus . IDLE ) ; TestHarnessUtil . assertOutputEquals ( "Output was not correct." , expectedOutput , testHarness . getOutput ( ) ) ; testHarness . processElement ( StreamStatus . ACTIVE , 1 , 0 ) ; testHarness . processElement ( StreamStatus . ACTIVE , 0 , 1 ) ; testHarness . waitForInputProcessing ( ) ; expectedOutput . add ( StreamStatus . ACTIVE ) ; TestHarnessUtil . assertOutputEquals ( "Output was not correct." , expectedOutput , testHarness . getOutput ( ) ) ; testHarness . endInput ( ) ; testHarness . waitForTaskCompletion ( ) ; List < String > resultElements = TestHarnessUtil . getRawElementsFromOutput ( testHarness . getOutput ( ) ) ; assertEquals ( 2 , resultElements . size ( ) ) ; }
This test verifies that watermarks and stream statuses are correctly forwarded.
This test verifies that watermarks and stream statuses are correctly forwarded .
// now the output should still be empty // now the watermark should have propagated, Map simply forward Watermarks // contrary to checkpoint barriers these elements are not blocked by watermarks // check whether we get the minimum of all the watermarks, this must also only occur in // the output after the two StreamRecords // advance watermark from one of the inputs, now we should get a new one since the // minimum increases // advance the other two inputs, now we should get a new one since the // minimum increases again // test whether idle input channels are acknowledged correctly when forwarding watermarks // this watermark should be advanced first // once this is acknowledged, // watermark (initial + 6) should be forwarded // make all input channels idle and check that the operator's idle status is forwarded // make some input channels active again and check that the operator's active status is forwarded only once