flink-instrumentation/workspace/flink-1.4.1-instrumented/flink-1.4.1/flink-runtime/src/test/java/org/apache/flink/runtime/state/StateBackendTestBase.java
/**
	 * Verify state restore resilience when:
	 *  - snapshot was taken without any Kryo registrations, specific serializers or default serializers for the state type
	 *  - restored with the state type registered (no specific serializer)
	 *
	 * This test should not fail, because de- / serialization of the state should noth be performed with Kryo's default
	 * {@link com.esotericsoftware.kryo.serializers.FieldSerializer}.
	 */
Verify state restore resilience when:
- snapshot was taken without any Kryo registrations, specific serializers or default serializers for the state type
- restored with the state type registered (no specific serializer)

This test should not fail, because de- / serialization of the state should noth be performed with Kryo's default
com.esotericsoftware.kryo.serializers.FieldSerializer.
Verify state restore resilience when : - snapshot was taken without any Kryo registrations specific serializers or default serializers for the state type - restored with the state type registered ( no specific serializer ) This test should not fail because de - / serialization of the state should noth be performed with Kryo ' s default com . esotericsoftware . kryo . serializers . FieldSerializer .
@Test
	public void testKryoRegisteringRestoreResilienceWithRegisteredType() throws Exception {
		CheckpointStreamFactory streamFactory = createStreamFactory();
		Environment env = new DummyEnvironment("test", 1, 0);
		AbstractKeyedStateBackend<Integer> backend = createKeyedBackend(IntSerializer.INSTANCE, env);
		TypeInformation<TestPojo> pojoType = new GenericTypeInfo<>(TestPojo.class);
		assertTrue(pojoType.createSerializer(env.getExecutionConfig()) instanceof KryoSerializer);
		ValueStateDescriptor<TestPojo> kvId = new ValueStateDescriptor<>("id", pojoType);
		ValueState<TestPojo> state = backend.getPartitionedState(VoidNamespace.INSTANCE, VoidNamespaceSerializer.INSTANCE, kvId);
		backend.setCurrentKey(1);
		state.update(new TestPojo("u1", 1));
		backend.setCurrentKey(2);
		state.update(new TestPojo("u2", 2));
		KeyedStateHandle snapshot = runSnapshot(backend.snapshot(
				682375462378L,
				2,
				streamFactory,
				CheckpointOptions.forCheckpoint()));
		backend.dispose();
		env.getExecutionConfig().registerKryoType(TestPojo.class);
		backend = restoreKeyedBackend(IntSerializer.INSTANCE, snapshot, env);
		snapshot.discardState();
		state = backend.getPartitionedState(VoidNamespace.INSTANCE, VoidNamespaceSerializer.INSTANCE, kvId);
		backend.setCurrentKey(1);
		assertEquals(state.value(), new TestPojo("u1", 1));
		backend.setCurrentKey(2);
		assertEquals(state.value(), new TestPojo("u2", 2));
		backend.dispose();
	}
@ Test public void testKryoRegisteringRestoreResilienceWithRegisteredType ( ) throws Exception { CheckpointStreamFactory streamFactory = createStreamFactory ( ) ; Environment env = new DummyEnvironment ( "test" , 1 , 0 ) ; AbstractKeyedStateBackend < Integer > backend = createKeyedBackend ( IntSerializer . INSTANCE , env ) ; TypeInformation < TestPojo > pojoType = new GenericTypeInfo < > ( TestPojo . class ) ; assertTrue ( pojoType . createSerializer ( env . getExecutionConfig ( ) ) instanceof KryoSerializer ) ; ValueStateDescriptor < TestPojo > kvId = new ValueStateDescriptor < > ( "id" , pojoType ) ; ValueState < TestPojo > state = backend . getPartitionedState ( VoidNamespace . INSTANCE , VoidNamespaceSerializer . INSTANCE , kvId ) ; backend . setCurrentKey ( 1 ) ; state . update ( new TestPojo ( "u1" , 1 ) ) ; backend . setCurrentKey ( 2 ) ; state . update ( new TestPojo ( "u2" , 2 ) ) ; KeyedStateHandle snapshot = runSnapshot ( backend . snapshot ( 682375462378L , 2 , streamFactory , CheckpointOptions . forCheckpoint ( ) ) ) ; backend . dispose ( ) ; env . getExecutionConfig ( ) . registerKryoType ( TestPojo . class ) ; backend = restoreKeyedBackend ( IntSerializer . INSTANCE , snapshot , env ) ; snapshot . discardState ( ) ; state = backend . getPartitionedState ( VoidNamespace . INSTANCE , VoidNamespaceSerializer . INSTANCE , kvId ) ; backend . setCurrentKey ( 1 ) ; assertEquals ( state . value ( ) , new TestPojo ( "u1" , 1 ) ) ; backend . setCurrentKey ( 2 ) ; assertEquals ( state . value ( ) , new TestPojo ( "u2" , 2 ) ) ; backend . dispose ( ) ; }
Verify state restore resilience when:
  - snapshot was taken without any Kryo registrations, specific serializers or default serializers for the state type
  - restored with the state type registered (no specific serializer)
Verify state restore resilience when : - snapshot was taken without any Kryo registrations specific serializers or default serializers for the state type - restored with the state type registered ( no specific serializer )
// make sure that we are in fact using the KryoSerializer // ============== create snapshot - no Kryo registration or specific / default serializers ============== // make some more modifications // ====================================== restore snapshot  ======================================