flink-instrumentation/workspace/flink-1.4.1-instrumented/flink-1.4.1/flink-runtime/src/test/java/org/apache/flink/runtime/state/MemoryStateBackendTest.java
/**
	 * Verifies that the operator state backend fails with appropriate error and message if
	 * previous serializer can not be restored.
	 */
Verifies that the operator state backend fails with appropriate error and message if
previous serializer can not be restored.
Verifies that the operator state backend fails with appropriate error and message if previous serializer can not be restored .
@Test
	public void testOperatorStateRestoreFailsIfSerializerDeserializationFails() throws Exception {
		AbstractStateBackend abstractStateBackend = new MemoryStateBackend(4096);
		Environment env = mock(Environment.class);
		when(env.getExecutionConfig()).thenReturn(new ExecutionConfig());
		when(env.getUserClassLoader()).thenReturn(OperatorStateBackendTest.class.getClassLoader());
		OperatorStateBackend operatorStateBackend = abstractStateBackend.createOperatorStateBackend(env, "test-op-name");
		ListStateDescriptor<Serializable> stateDescriptor1 = new ListStateDescriptor<>("test1", new JavaSerializer<>());
		ListStateDescriptor<Serializable> stateDescriptor2 = new ListStateDescriptor<>("test2", new JavaSerializer<>());
		ListStateDescriptor<Serializable> stateDescriptor3 = new ListStateDescriptor<>("test3", new JavaSerializer<>());
		ListState<Serializable> listState1 = operatorStateBackend.getListState(stateDescriptor1);
		ListState<Serializable> listState2 = operatorStateBackend.getListState(stateDescriptor2);
		ListState<Serializable> listState3 = operatorStateBackend.getUnionListState(stateDescriptor3);
		listState1.add(42);
		listState1.add(4711);
		listState2.add(7);
		listState2.add(13);
		listState2.add(23);
		listState3.add(17);
		listState3.add(18);
		listState3.add(19);
		listState3.add(20);
		CheckpointStreamFactory streamFactory = abstractStateBackend.createStreamFactory(new JobID(), "testOperator");
		RunnableFuture<OperatorStateHandle> runnableFuture =
			operatorStateBackend.snapshot(1, 1, streamFactory, CheckpointOptions.forCheckpoint());
		OperatorStateHandle stateHandle = FutureUtil.runIfNotDoneAndGet(runnableFuture);
		try {
			operatorStateBackend.close();
			operatorStateBackend.dispose();
			operatorStateBackend = abstractStateBackend.createOperatorStateBackend(
				env,
				"testOperator");
			TypeSerializerSerializationUtil.TypeSerializerSerializationProxy<?> mockProxy =
					mock(TypeSerializerSerializationUtil.TypeSerializerSerializationProxy.class);
			doThrow(new IOException()).when(mockProxy).read(any(DataInputViewStreamWrapper.class));
			PowerMockito.whenNew(TypeSerializerSerializationUtil.TypeSerializerSerializationProxy.class).withAnyArguments().thenReturn(mockProxy);
			operatorStateBackend.restore(Collections.singletonList(stateHandle));
			fail("The operator state restore should have failed if the previous state serializer could not be loaded.");
		} catch (IOException expected) {
			Assert.assertTrue(expected.getMessage().contains("Unable to restore operator state"));
		} finally {
			stateHandle.discardState();
		}
	}
@ Test public void testOperatorStateRestoreFailsIfSerializerDeserializationFails ( ) throws Exception { AbstractStateBackend abstractStateBackend = new MemoryStateBackend ( 4096 ) ; Environment env = mock ( Environment . class ) ; when ( env . getExecutionConfig ( ) ) . thenReturn ( new ExecutionConfig ( ) ) ; when ( env . getUserClassLoader ( ) ) . thenReturn ( OperatorStateBackendTest . class . getClassLoader ( ) ) ; OperatorStateBackend operatorStateBackend = abstractStateBackend . createOperatorStateBackend ( env , "test-op-name" ) ; ListStateDescriptor < Serializable > stateDescriptor1 = new ListStateDescriptor < > ( "test1" , new JavaSerializer < > ( ) ) ; ListStateDescriptor < Serializable > stateDescriptor2 = new ListStateDescriptor < > ( "test2" , new JavaSerializer < > ( ) ) ; ListStateDescriptor < Serializable > stateDescriptor3 = new ListStateDescriptor < > ( "test3" , new JavaSerializer < > ( ) ) ; ListState < Serializable > listState1 = operatorStateBackend . getListState ( stateDescriptor1 ) ; ListState < Serializable > listState2 = operatorStateBackend . getListState ( stateDescriptor2 ) ; ListState < Serializable > listState3 = operatorStateBackend . getUnionListState ( stateDescriptor3 ) ; listState1 . add ( 42 ) ; listState1 . add ( 4711 ) ; listState2 . add ( 7 ) ; listState2 . add ( 13 ) ; listState2 . add ( 23 ) ; listState3 . add ( 17 ) ; listState3 . add ( 18 ) ; listState3 . add ( 19 ) ; listState3 . add ( 20 ) ; CheckpointStreamFactory streamFactory = abstractStateBackend . createStreamFactory ( new JobID ( ) , "testOperator" ) ; RunnableFuture < OperatorStateHandle > runnableFuture = operatorStateBackend . snapshot ( 1 , 1 , streamFactory , CheckpointOptions . forCheckpoint ( ) ) ; OperatorStateHandle stateHandle = FutureUtil . runIfNotDoneAndGet ( runnableFuture ) ; try { operatorStateBackend . close ( ) ; operatorStateBackend . dispose ( ) ; operatorStateBackend = abstractStateBackend . createOperatorStateBackend ( env , "testOperator" ) ; TypeSerializerSerializationUtil . TypeSerializerSerializationProxy < ? > mockProxy = mock ( TypeSerializerSerializationUtil . TypeSerializerSerializationProxy . class ) ; doThrow ( new IOException ( ) ) . when ( mockProxy ) . read ( any ( DataInputViewStreamWrapper . class ) ) ; PowerMockito . whenNew ( TypeSerializerSerializationUtil . TypeSerializerSerializationProxy . class ) . withAnyArguments ( ) . thenReturn ( mockProxy ) ; operatorStateBackend . restore ( Collections . singletonList ( stateHandle ) ) ; fail ( "The operator state restore should have failed if the previous state serializer could not be loaded." ) ; } catch ( IOException expected ) { Assert . assertTrue ( expected . getMessage ( ) . contains ( "Unable to restore operator state" ) ) ; } finally { stateHandle . discardState ( ) ; } }
Verifies that the operator state backend fails with appropriate error and message if
 previous serializer can not be restored.
Verifies that the operator state backend fails with appropriate error and message if previous serializer can not be restored .
// write some state // mock failure when deserializing serializer