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 verifies that open() and close() are correctly called by the StreamTask.
	 */
This test verifies that open() and close() are correctly called by the StreamTask.
This test verifies that open () and close () are correctly called by the StreamTask .
@Test
	public void testOpenClose() throws Exception {
		final SourceStreamTask<String, SourceFunction<String>, StreamSource<String, SourceFunction<String>>> sourceTask = new SourceStreamTask<>();
		final StreamTaskTestHarness<String> testHarness = new StreamTaskTestHarness<String>(sourceTask, BasicTypeInfo.STRING_TYPE_INFO);
		testHarness.setupOutputForSingletonOperatorChain();
		StreamConfig streamConfig = testHarness.getStreamConfig();
		StreamSource<String, ?> sourceOperator = new StreamSource<>(new OpenCloseTestSource());
		streamConfig.setStreamOperator(sourceOperator);
		streamConfig.setOperatorID(new OperatorID());
		testHarness.invoke();
		testHarness.waitForTaskCompletion();
		Assert.assertTrue("RichFunction methods where not called.", OpenCloseTestSource.closeCalled);
		List<String> resultElements = TestHarnessUtil.getRawElementsFromOutput(testHarness.getOutput());
		Assert.assertEquals(10, resultElements.size());
	}
@ Test public void testOpenClose ( ) throws Exception { final SourceStreamTask < String , SourceFunction < String > , StreamSource < String , SourceFunction < String > > > sourceTask = new SourceStreamTask < > ( ) ; final StreamTaskTestHarness < String > testHarness = new StreamTaskTestHarness < String > ( sourceTask , BasicTypeInfo . STRING_TYPE_INFO ) ; testHarness . setupOutputForSingletonOperatorChain ( ) ; StreamConfig streamConfig = testHarness . getStreamConfig ( ) ; StreamSource < String , ? > sourceOperator = new StreamSource < > ( new OpenCloseTestSource ( ) ) ; streamConfig . setStreamOperator ( sourceOperator ) ; streamConfig . setOperatorID ( new OperatorID ( ) ) ; testHarness . invoke ( ) ; testHarness . waitForTaskCompletion ( ) ; Assert . assertTrue ( "RichFunction methods where not called." , OpenCloseTestSource . closeCalled ) ; List < String > resultElements = TestHarnessUtil . getRawElementsFromOutput ( testHarness . getOutput ( ) ) ; Assert . assertEquals ( 10 , resultElements . size ( ) ) ; }
This test verifies that open() and close() are correctly called by the StreamTask.
This test verifies that open () and close () are correctly called by the StreamTask .