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 memory-backed keyed state backend fails with appropriate error and message if
	 * previous serializer can not be restored.
	 */
Verifies that memory-backed keyed state backend fails with appropriate error and message if
previous serializer can not be restored.
Verifies that memory - backed keyed state backend fails with appropriate error and message if previous serializer can not be restored .
@Test
	public void testKeyedStateRestoreFailsIfSerializerDeserializationFails() throws Exception {
		CheckpointStreamFactory streamFactory = createStreamFactory();
		KeyedStateBackend<Integer> backend = createKeyedBackend(IntSerializer.INSTANCE);
		ValueStateDescriptor<String> kvId = new ValueStateDescriptor<>("id", String.class, null);
		kvId.initializeSerializerUnlessSet(new ExecutionConfig());
		HeapKeyedStateBackend<Integer> heapBackend = (HeapKeyedStateBackend<Integer>) backend;
		assertEquals(0, heapBackend.numStateEntries());
		ValueState<String> state = backend.getPartitionedState(VoidNamespace.INSTANCE, VoidNamespaceSerializer.INSTANCE, kvId);
		backend.setCurrentKey(0);
		state.update("hello");
		state.update("ciao");
		KeyedStateHandle snapshot = runSnapshot(((HeapKeyedStateBackend<Integer>) backend).snapshot(
			682375462378L,
			2,
			streamFactory,
			CheckpointOptions.forCheckpoint()));
		backend.dispose();
		Environment env = mock(Environment.class);
		when(env.getExecutionConfig()).thenReturn(new ExecutionConfig());
		when(env.getUserClassLoader()).thenReturn(OperatorStateBackendTest.class.getClassLoader());
		TypeSerializerSerializationUtil.TypeSerializerSerializationProxy<?> mockProxy =
				mock(TypeSerializerSerializationUtil.TypeSerializerSerializationProxy.class);
		doThrow(new IOException()).when(mockProxy).read(any(DataInputViewStreamWrapper.class));
		PowerMockito.whenNew(TypeSerializerSerializationUtil.TypeSerializerSerializationProxy.class).withAnyArguments().thenReturn(mockProxy);
		try {
			restoreKeyedBackend(IntSerializer.INSTANCE, snapshot, env);
			fail("The keyed 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 keyed state"));
		}
	}
@ Test public void testKeyedStateRestoreFailsIfSerializerDeserializationFails ( ) throws Exception { CheckpointStreamFactory streamFactory = createStreamFactory ( ) ; KeyedStateBackend < Integer > backend = createKeyedBackend ( IntSerializer . INSTANCE ) ; ValueStateDescriptor < String > kvId = new ValueStateDescriptor < > ( "id" , String . class , null ) ; kvId . initializeSerializerUnlessSet ( new ExecutionConfig ( ) ) ; HeapKeyedStateBackend < Integer > heapBackend = ( HeapKeyedStateBackend < Integer > ) backend ; assertEquals ( 0 , heapBackend . numStateEntries ( ) ) ; ValueState < String > state = backend . getPartitionedState ( VoidNamespace . INSTANCE , VoidNamespaceSerializer . INSTANCE , kvId ) ; backend . setCurrentKey ( 0 ) ; state . update ( "hello" ) ; state . update ( "ciao" ) ; KeyedStateHandle snapshot = runSnapshot ( ( ( HeapKeyedStateBackend < Integer > ) backend ) . snapshot ( 682375462378L , 2 , streamFactory , CheckpointOptions . forCheckpoint ( ) ) ) ; backend . dispose ( ) ; Environment env = mock ( Environment . class ) ; when ( env . getExecutionConfig ( ) ) . thenReturn ( new ExecutionConfig ( ) ) ; when ( env . getUserClassLoader ( ) ) . thenReturn ( OperatorStateBackendTest . class . getClassLoader ( ) ) ; TypeSerializerSerializationUtil . TypeSerializerSerializationProxy < ? > mockProxy = mock ( TypeSerializerSerializationUtil . TypeSerializerSerializationProxy . class ) ; doThrow ( new IOException ( ) ) . when ( mockProxy ) . read ( any ( DataInputViewStreamWrapper . class ) ) ; PowerMockito . whenNew ( TypeSerializerSerializationUtil . TypeSerializerSerializationProxy . class ) . withAnyArguments ( ) . thenReturn ( mockProxy ) ; try { restoreKeyedBackend ( IntSerializer . INSTANCE , snapshot , env ) ; fail ( "The keyed 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 keyed state" ) ) ; } }
Verifies that memory-backed keyed state backend fails with appropriate error and message if
 previous serializer can not be restored.
Verifies that memory - backed keyed state backend fails with appropriate error and message if previous serializer can not be restored .
// write some state // ========== restore snapshot ========== // mock failure when deserializing serializer