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 open() and close() are correctly called. This test also verifies
	 * that timestamps of emitted elements are correct. {@link StreamMap} assigns the input
	 * timestamp to emitted elements.
	 */
This test verifies that open() and close() are correctly called. This test also verifies
that timestamps of emitted elements are correct.  StreamMap assigns the input
timestamp to emitted elements.
This test verifies that open () and close () are correctly called . This test also verifies that timestamps of emitted elements are correct . StreamMap assigns the input timestamp to emitted elements .
@Test
	public void testOpenCloseAndTimestamps() throws Exception {
		final OneInputStreamTask<String, String> mapTask = new OneInputStreamTask<String, String>();
		final OneInputStreamTaskTestHarness<String, String> testHarness = new OneInputStreamTaskTestHarness<String, String>(mapTask, BasicTypeInfo.STRING_TYPE_INFO, BasicTypeInfo.STRING_TYPE_INFO);
		testHarness.setupOutputForSingletonOperatorChain();
		StreamConfig streamConfig = testHarness.getStreamConfig();
		StreamMap<String, String> mapOperator = new StreamMap<String, String>(new TestOpenCloseMapFunction());
		streamConfig.setStreamOperator(mapOperator);
		streamConfig.setOperatorID(new OperatorID());
		long initialTime = 0L;
		ConcurrentLinkedQueue<Object> expectedOutput = new ConcurrentLinkedQueue<Object>();
		testHarness.invoke();
		testHarness.waitForTaskRunning();
		testHarness.processElement(new StreamRecord<String>("Hello", initialTime + 1));
		testHarness.processElement(new StreamRecord<String>("Ciao", initialTime + 2));
		expectedOutput.add(new StreamRecord<String>("Hello", initialTime + 1));
		expectedOutput.add(new StreamRecord<String>("Ciao", initialTime + 2));
		testHarness.endInput();
		testHarness.waitForTaskCompletion();
		assertTrue("RichFunction methods where not called.", TestOpenCloseMapFunction.closeCalled);
		TestHarnessUtil.assertOutputEquals("Output was not correct.",
			expectedOutput,
			testHarness.getOutput());
	}
@ Test public void testOpenCloseAndTimestamps ( ) throws Exception { final OneInputStreamTask < String , String > mapTask = new OneInputStreamTask < String , String > ( ) ; final OneInputStreamTaskTestHarness < String , String > testHarness = new OneInputStreamTaskTestHarness < String , String > ( mapTask , BasicTypeInfo . STRING_TYPE_INFO , BasicTypeInfo . STRING_TYPE_INFO ) ; testHarness . setupOutputForSingletonOperatorChain ( ) ; StreamConfig streamConfig = testHarness . getStreamConfig ( ) ; StreamMap < String , String > mapOperator = new StreamMap < String , String > ( new TestOpenCloseMapFunction ( ) ) ; streamConfig . setStreamOperator ( mapOperator ) ; streamConfig . setOperatorID ( new OperatorID ( ) ) ; long initialTime = 0L ; ConcurrentLinkedQueue < Object > expectedOutput = new ConcurrentLinkedQueue < Object > ( ) ; testHarness . invoke ( ) ; testHarness . waitForTaskRunning ( ) ; testHarness . processElement ( new StreamRecord < String > ( "Hello" , initialTime + 1 ) ) ; testHarness . processElement ( new StreamRecord < String > ( "Ciao" , initialTime + 2 ) ) ; expectedOutput . add ( new StreamRecord < String > ( "Hello" , initialTime + 1 ) ) ; expectedOutput . add ( new StreamRecord < String > ( "Ciao" , initialTime + 2 ) ) ; testHarness . endInput ( ) ; testHarness . waitForTaskCompletion ( ) ; assertTrue ( "RichFunction methods where not called." , TestOpenCloseMapFunction . closeCalled ) ; TestHarnessUtil . assertOutputEquals ( "Output was not correct." , expectedOutput , testHarness . getOutput ( ) ) ; }
This test verifies that open() and close() are correctly called.
This test verifies that open () and close () are correctly called .