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 are not forwarded when the task is idle.
	 * It also verifies that when task is idle, watermarks generated in the middle of chains are also blocked and
	 * never forwarded.
	 *
	 * <p>The tested chain will be: (HEAD: normal operator) --> (watermark generating operator) --> (normal operator).
	 * The operators will throw an exception and fail the test if either of them were forwarded watermarks when
	 * the task is idle.
	 */
This test verifies that watermarks are not forwarded when the task is idle.
It also verifies that when task is idle, watermarks generated in the middle of chains are also blocked and
never forwarded.

The tested chain will be: (HEAD: normal operator) --> (watermark generating operator) --> (normal operator).
The operators will throw an exception and fail the test if either of them were forwarded watermarks when
the task is idle.
This test verifies that watermarks are not forwarded when the task is idle . It also verifies that when task is idle watermarks generated in the middle of chains are also blocked and never forwarded . The tested chain will be : ( HEAD : normal operator ) -- > ( watermark generating operator ) -- > ( normal operator ) . The operators will throw an exception and fail the test if either of them were forwarded watermarks when the task is idle .
@Test
	public void testWatermarksNotForwardedWithinChainWhenIdle() throws Exception {
		final OneInputStreamTask<String, String> testTask = new OneInputStreamTask<>();
		final OneInputStreamTaskTestHarness<String, String> testHarness =
			new OneInputStreamTaskTestHarness<String, String>(
				testTask, 1, 1,
				BasicTypeInfo.STRING_TYPE_INFO,
				BasicTypeInfo.STRING_TYPE_INFO);
		TriggerableFailOnWatermarkTestOperator headOperator = new TriggerableFailOnWatermarkTestOperator();
		WatermarkGeneratingTestOperator watermarkOperator = new WatermarkGeneratingTestOperator();
		TriggerableFailOnWatermarkTestOperator tailOperator = new TriggerableFailOnWatermarkTestOperator();
		testHarness.setupOperatorChain(new OperatorID(42L, 42L), headOperator)
			.chain(new OperatorID(4711L, 42L), watermarkOperator, StringSerializer.INSTANCE)
			.chain(new OperatorID(123L, 123L), tailOperator, StringSerializer.INSTANCE)
			.finish();
		ConcurrentLinkedQueue<Object> expectedOutput = new ConcurrentLinkedQueue<Object>();
		testHarness.invoke();
		testHarness.waitForTaskRunning();
		testHarness.processElement(new StreamRecord<>(TriggerableFailOnWatermarkTestOperator.EXPECT_FORWARDED_WATERMARKS_MARKER));
		testHarness.processElement(new StreamRecord<>("10"), 0, 0);
		testHarness.processElement(new Watermark(15));
		testHarness.processElement(new StreamRecord<>("20"), 0, 0);
		testHarness.processElement(new StreamRecord<>("30"), 0, 0);
		testHarness.waitForInputProcessing();
		expectedOutput.add(new StreamRecord<>(TriggerableFailOnWatermarkTestOperator.EXPECT_FORWARDED_WATERMARKS_MARKER));
		expectedOutput.add(new StreamRecord<>("10"));
		expectedOutput.add(new Watermark(10));
		expectedOutput.add(new StreamRecord<>("20"));
		expectedOutput.add(new Watermark(20));
		expectedOutput.add(new StreamRecord<>("30"));
		expectedOutput.add(new Watermark(30));
		TestHarnessUtil.assertOutputEquals("Output was not correct.", expectedOutput, testHarness.getOutput());
		testHarness.processElement(StreamStatus.IDLE);
		testHarness.processElement(new StreamRecord<>(TriggerableFailOnWatermarkTestOperator.NO_FORWARDED_WATERMARKS_MARKER));
		testHarness.processElement(new StreamRecord<>("40"), 0, 0);
		testHarness.processElement(new StreamRecord<>("50"), 0, 0);
		testHarness.processElement(new StreamRecord<>("60"), 0, 0);
		testHarness.processElement(new Watermark(65)); 
		testHarness.waitForInputProcessing();
		expectedOutput.add(StreamStatus.IDLE);
		expectedOutput.add(new StreamRecord<>(TriggerableFailOnWatermarkTestOperator.NO_FORWARDED_WATERMARKS_MARKER));
		expectedOutput.add(new StreamRecord<>("40"));
		expectedOutput.add(new StreamRecord<>("50"));
		expectedOutput.add(new StreamRecord<>("60"));
		TestHarnessUtil.assertOutputEquals("Output was not correct.", expectedOutput, testHarness.getOutput());
		testHarness.processElement(StreamStatus.ACTIVE);
		testHarness.processElement(new StreamRecord<>(TriggerableFailOnWatermarkTestOperator.EXPECT_FORWARDED_WATERMARKS_MARKER));
		testHarness.processElement(new StreamRecord<>("70"), 0, 0);
		testHarness.processElement(new StreamRecord<>("80"), 0, 0);
		testHarness.processElement(new StreamRecord<>("90"), 0, 0);
		testHarness.waitForInputProcessing();
		expectedOutput.add(StreamStatus.ACTIVE);
		expectedOutput.add(new StreamRecord<>(TriggerableFailOnWatermarkTestOperator.EXPECT_FORWARDED_WATERMARKS_MARKER));
		expectedOutput.add(new StreamRecord<>("70"));
		expectedOutput.add(new Watermark(70));
		expectedOutput.add(new StreamRecord<>("80"));
		expectedOutput.add(new Watermark(80));
		expectedOutput.add(new StreamRecord<>("90"));
		expectedOutput.add(new Watermark(90));
		TestHarnessUtil.assertOutputEquals("Output was not correct.", expectedOutput, testHarness.getOutput());
		testHarness.endInput();
		testHarness.waitForTaskCompletion();
		List<String> resultElements = TestHarnessUtil.getRawElementsFromOutput(testHarness.getOutput());
		assertEquals(12, resultElements.size());
	}
@ Test public void testWatermarksNotForwardedWithinChainWhenIdle ( ) throws Exception { final OneInputStreamTask < String , String > testTask = new OneInputStreamTask < > ( ) ; final OneInputStreamTaskTestHarness < String , String > testHarness = new OneInputStreamTaskTestHarness < String , String > ( testTask , 1 , 1 , BasicTypeInfo . STRING_TYPE_INFO , BasicTypeInfo . STRING_TYPE_INFO ) ; TriggerableFailOnWatermarkTestOperator headOperator = new TriggerableFailOnWatermarkTestOperator ( ) ; WatermarkGeneratingTestOperator watermarkOperator = new WatermarkGeneratingTestOperator ( ) ; TriggerableFailOnWatermarkTestOperator tailOperator = new TriggerableFailOnWatermarkTestOperator ( ) ; testHarness . setupOperatorChain ( new OperatorID ( 42L , 42L ) , headOperator ) . chain ( new OperatorID ( 4711L , 42L ) , watermarkOperator , StringSerializer . INSTANCE ) . chain ( new OperatorID ( 123L , 123L ) , tailOperator , StringSerializer . INSTANCE ) . finish ( ) ; ConcurrentLinkedQueue < Object > expectedOutput = new ConcurrentLinkedQueue < Object > ( ) ; testHarness . invoke ( ) ; testHarness . waitForTaskRunning ( ) ; testHarness . processElement ( new StreamRecord < > ( TriggerableFailOnWatermarkTestOperator . EXPECT_FORWARDED_WATERMARKS_MARKER ) ) ; testHarness . processElement ( new StreamRecord < > ( "10" ) , 0 , 0 ) ; testHarness . processElement ( new Watermark ( 15 ) ) ; testHarness . processElement ( new StreamRecord < > ( "20" ) , 0 , 0 ) ; testHarness . processElement ( new StreamRecord < > ( "30" ) , 0 , 0 ) ; testHarness . waitForInputProcessing ( ) ; expectedOutput . add ( new StreamRecord < > ( TriggerableFailOnWatermarkTestOperator . EXPECT_FORWARDED_WATERMARKS_MARKER ) ) ; expectedOutput . add ( new StreamRecord < > ( "10" ) ) ; expectedOutput . add ( new Watermark ( 10 ) ) ; expectedOutput . add ( new StreamRecord < > ( "20" ) ) ; expectedOutput . add ( new Watermark ( 20 ) ) ; expectedOutput . add ( new StreamRecord < > ( "30" ) ) ; expectedOutput . add ( new Watermark ( 30 ) ) ; TestHarnessUtil . assertOutputEquals ( "Output was not correct." , expectedOutput , testHarness . getOutput ( ) ) ; testHarness . processElement ( StreamStatus . IDLE ) ; testHarness . processElement ( new StreamRecord < > ( TriggerableFailOnWatermarkTestOperator . NO_FORWARDED_WATERMARKS_MARKER ) ) ; testHarness . processElement ( new StreamRecord < > ( "40" ) , 0 , 0 ) ; testHarness . processElement ( new StreamRecord < > ( "50" ) , 0 , 0 ) ; testHarness . processElement ( new StreamRecord < > ( "60" ) , 0 , 0 ) ; testHarness . processElement ( new Watermark ( 65 ) ) ; testHarness . waitForInputProcessing ( ) ; expectedOutput . add ( StreamStatus . IDLE ) ; expectedOutput . add ( new StreamRecord < > ( TriggerableFailOnWatermarkTestOperator . NO_FORWARDED_WATERMARKS_MARKER ) ) ; expectedOutput . add ( new StreamRecord < > ( "40" ) ) ; expectedOutput . add ( new StreamRecord < > ( "50" ) ) ; expectedOutput . add ( new StreamRecord < > ( "60" ) ) ; TestHarnessUtil . assertOutputEquals ( "Output was not correct." , expectedOutput , testHarness . getOutput ( ) ) ; testHarness . processElement ( StreamStatus . ACTIVE ) ; testHarness . processElement ( new StreamRecord < > ( TriggerableFailOnWatermarkTestOperator . EXPECT_FORWARDED_WATERMARKS_MARKER ) ) ; testHarness . processElement ( new StreamRecord < > ( "70" ) , 0 , 0 ) ; testHarness . processElement ( new StreamRecord < > ( "80" ) , 0 , 0 ) ; testHarness . processElement ( new StreamRecord < > ( "90" ) , 0 , 0 ) ; testHarness . waitForInputProcessing ( ) ; expectedOutput . add ( StreamStatus . ACTIVE ) ; expectedOutput . add ( new StreamRecord < > ( TriggerableFailOnWatermarkTestOperator . EXPECT_FORWARDED_WATERMARKS_MARKER ) ) ; expectedOutput . add ( new StreamRecord < > ( "70" ) ) ; expectedOutput . add ( new Watermark ( 70 ) ) ; expectedOutput . add ( new StreamRecord < > ( "80" ) ) ; expectedOutput . add ( new Watermark ( 80 ) ) ; expectedOutput . add ( new StreamRecord < > ( "90" ) ) ; expectedOutput . add ( new Watermark ( 90 ) ) ; TestHarnessUtil . assertOutputEquals ( "Output was not correct." , expectedOutput , testHarness . getOutput ( ) ) ; testHarness . endInput ( ) ; testHarness . waitForTaskCompletion ( ) ; List < String > resultElements = TestHarnessUtil . getRawElementsFromOutput ( testHarness . getOutput ( ) ) ; assertEquals ( 12 , resultElements . size ( ) ) ; }
This test verifies that watermarks are not forwarded when the task is idle.
This test verifies that watermarks are not forwarded when the task is idle .
// --------------------- begin test --------------------- // the task starts as active, so all generated watermarks should be forwarded // this watermark will be forwarded since the task is currently active, // but should not be in the final output because it should be blocked by the watermark generator in the chain // now, toggle the task to be idle, and let the watermark generator produce some watermarks // after this, the operators will throw an exception if they are forwarded watermarks anywhere in the chain // NOTE: normally, tasks will not have records to process while idle; // we're doing this here only to mimic watermark generating in operators // the test will fail if any of the operators were forwarded this // the 40 - 60 watermarks should not be forwarded, only the stream status toggle element and records // re-toggle the task to be active and see if new watermarks are correctly forwarded again