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
/**
	 * Tests that the stream operator can snapshot and restore the operator state of chained
	 * operators.
	 */
Tests that the stream operator can snapshot and restore the operator state of chained
operators.
Tests that the stream operator can snapshot and restore the operator state of chained operators .
@Test
	public void testSnapshottingAndRestoring() throws Exception {
		final Deadline deadline = new FiniteDuration(2, TimeUnit.MINUTES).fromNow();
		final OneInputStreamTask<String, String> streamTask = new OneInputStreamTask<String, String>();
		final OneInputStreamTaskTestHarness<String, String> testHarness = new OneInputStreamTaskTestHarness<String, String>(streamTask, BasicTypeInfo.STRING_TYPE_INFO, BasicTypeInfo.STRING_TYPE_INFO);
		testHarness.setupOutputForSingletonOperatorChain();
		IdentityKeySelector<String> keySelector = new IdentityKeySelector<>();
		testHarness.configureForKeyedStream(keySelector, BasicTypeInfo.STRING_TYPE_INFO);
		long checkpointId = 1L;
		long checkpointTimestamp = 1L;
		int numberChainedTasks = 11;
		StreamConfig streamConfig = testHarness.getStreamConfig();
		configureChainedTestingStreamOperator(streamConfig, numberChainedTasks);
		AcknowledgeStreamMockEnvironment env = new AcknowledgeStreamMockEnvironment(
			testHarness.jobConfig,
			testHarness.taskConfig,
			testHarness.executionConfig,
			testHarness.memorySize,
			new MockInputSplitProvider(),
			testHarness.bufferSize);
		TestingStreamOperator.numberRestoreCalls = 0;
		testHarness.invoke(env);
		testHarness.waitForTaskRunning(deadline.timeLeft().toMillis());
		CheckpointMetaData checkpointMetaData = new CheckpointMetaData(checkpointId, checkpointTimestamp);
		while (!streamTask.triggerCheckpoint(checkpointMetaData, CheckpointOptions.forCheckpoint())) {}
		assertEquals(0, TestingStreamOperator.numberRestoreCalls);
		env.getCheckpointLatch().await();
		assertEquals(checkpointId, env.getCheckpointId());
		testHarness.endInput();
		testHarness.waitForTaskCompletion(deadline.timeLeft().toMillis());
		final OneInputStreamTask<String, String> restoredTask = new OneInputStreamTask<String, String>();
		final OneInputStreamTaskTestHarness<String, String> restoredTaskHarness =
			new OneInputStreamTaskTestHarness<String, String>(restoredTask, BasicTypeInfo.STRING_TYPE_INFO, BasicTypeInfo.STRING_TYPE_INFO);
		restoredTaskHarness.configureForKeyedStream(keySelector, BasicTypeInfo.STRING_TYPE_INFO);
		StreamConfig restoredTaskStreamConfig = restoredTaskHarness.getStreamConfig();
		configureChainedTestingStreamOperator(restoredTaskStreamConfig, numberChainedTasks);
		TaskStateSnapshot stateHandles = env.getCheckpointStateHandles();
		Assert.assertEquals(numberChainedTasks, stateHandles.getSubtaskStateMappings().size());
		restoredTask.setInitialState(stateHandles);
		TestingStreamOperator.numberRestoreCalls = 0;
		restoredTaskHarness.invoke();
		restoredTaskHarness.endInput();
		restoredTaskHarness.waitForTaskCompletion(deadline.timeLeft().toMillis());
		assertEquals(numberChainedTasks, TestingStreamOperator.numberRestoreCalls);
		TestingStreamOperator.numberRestoreCalls = 0;
	}
@ Test public void testSnapshottingAndRestoring ( ) throws Exception { final Deadline deadline = new FiniteDuration ( 2 , TimeUnit . MINUTES ) . fromNow ( ) ; final OneInputStreamTask < String , String > streamTask = new OneInputStreamTask < String , String > ( ) ; final OneInputStreamTaskTestHarness < String , String > testHarness = new OneInputStreamTaskTestHarness < String , String > ( streamTask , BasicTypeInfo . STRING_TYPE_INFO , BasicTypeInfo . STRING_TYPE_INFO ) ; testHarness . setupOutputForSingletonOperatorChain ( ) ; IdentityKeySelector < String > keySelector = new IdentityKeySelector < > ( ) ; testHarness . configureForKeyedStream ( keySelector , BasicTypeInfo . STRING_TYPE_INFO ) ; long checkpointId = 1L ; long checkpointTimestamp = 1L ; int numberChainedTasks = 11 ; StreamConfig streamConfig = testHarness . getStreamConfig ( ) ; configureChainedTestingStreamOperator ( streamConfig , numberChainedTasks ) ; AcknowledgeStreamMockEnvironment env = new AcknowledgeStreamMockEnvironment ( testHarness . jobConfig , testHarness . taskConfig , testHarness . executionConfig , testHarness . memorySize , new MockInputSplitProvider ( ) , testHarness . bufferSize ) ; TestingStreamOperator . numberRestoreCalls = 0 ; testHarness . invoke ( env ) ; testHarness . waitForTaskRunning ( deadline . timeLeft ( ) . toMillis ( ) ) ; CheckpointMetaData checkpointMetaData = new CheckpointMetaData ( checkpointId , checkpointTimestamp ) ; while ( ! streamTask . triggerCheckpoint ( checkpointMetaData , CheckpointOptions . forCheckpoint ( ) ) ) { } assertEquals ( 0 , TestingStreamOperator . numberRestoreCalls ) ; env . getCheckpointLatch ( ) . await ( ) ; assertEquals ( checkpointId , env . getCheckpointId ( ) ) ; testHarness . endInput ( ) ; testHarness . waitForTaskCompletion ( deadline . timeLeft ( ) . toMillis ( ) ) ; final OneInputStreamTask < String , String > restoredTask = new OneInputStreamTask < String , String > ( ) ; final OneInputStreamTaskTestHarness < String , String > restoredTaskHarness = new OneInputStreamTaskTestHarness < String , String > ( restoredTask , BasicTypeInfo . STRING_TYPE_INFO , BasicTypeInfo . STRING_TYPE_INFO ) ; restoredTaskHarness . configureForKeyedStream ( keySelector , BasicTypeInfo . STRING_TYPE_INFO ) ; StreamConfig restoredTaskStreamConfig = restoredTaskHarness . getStreamConfig ( ) ; configureChainedTestingStreamOperator ( restoredTaskStreamConfig , numberChainedTasks ) ; TaskStateSnapshot stateHandles = env . getCheckpointStateHandles ( ) ; Assert . assertEquals ( numberChainedTasks , stateHandles . getSubtaskStateMappings ( ) . size ( ) ) ; restoredTask . setInitialState ( stateHandles ) ; TestingStreamOperator . numberRestoreCalls = 0 ; restoredTaskHarness . invoke ( ) ; restoredTaskHarness . endInput ( ) ; restoredTaskHarness . waitForTaskCompletion ( deadline . timeLeft ( ) . toMillis ( ) ) ; assertEquals ( numberChainedTasks , TestingStreamOperator . numberRestoreCalls ) ; TestingStreamOperator . numberRestoreCalls = 0 ; }
Tests that the stream operator can snapshot and restore the operator state of chained
 operators.
Tests that the stream operator can snapshot and restore the operator state of chained operators .
// reset number of restore calls // since no state was set, there shouldn't be restore calls // restore of every chained operator should have been called