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 a default serializer for the state type
	 *
	 * <p> The default serializer used on restore is {@link CustomKryoTestSerializer}, which deliberately
	 * fails only on deserialization. We use the deliberate deserialization failure to acknowledge test success.
	 *
	 * @throws Exception expects {@link ExpectedKryoTestException} to be thrown.
	 */
Verify state restore resilience when:
- snapshot was taken without any Kryo registrations, specific serializers or default serializers for the state type
- restored with a default serializer for the state type

 The default serializer used on restore is  CustomKryoTestSerializer, which deliberately
fails only on deserialization. We use the deliberate deserialization failure to acknowledge test success.

@throws Exception expects  ExpectedKryoTestException to be thrown.
Verify state restore resilience when : - snapshot was taken without any Kryo registrations specific serializers or default serializers for the state type - restored with a default serializer for the state type The default serializer used on restore is CustomKryoTestSerializer which deliberately fails only on deserialization . We use the deliberate deserialization failure to acknowledge test success . @throws Exception expects ExpectedKryoTestException to be thrown .
@Test
	@SuppressWarnings("unchecked")
	public void testKryoRegisteringRestoreResilienceWithDefaultSerializer() throws Exception {
		CheckpointStreamFactory streamFactory = createStreamFactory();
		SharedStateRegistry sharedStateRegistry = new SharedStateRegistry();
		Environment env = new DummyEnvironment("test", 1, 0);
		AbstractKeyedStateBackend<Integer> backend = null;
		try {
			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()));
			snapshot.registerSharedStates(sharedStateRegistry);
			backend.dispose();
			env.getExecutionConfig().addDefaultKryoSerializer(TestPojo.class, (Class) CustomKryoTestSerializer.class);
			backend = restoreKeyedBackend(IntSerializer.INSTANCE, snapshot, env);
			kvId = new ValueStateDescriptor<>("id", pojoType);
			state = backend.getPartitionedState(VoidNamespace.INSTANCE, VoidNamespaceSerializer.INSTANCE, kvId);
			backend.setCurrentKey(1);
			state.update(new TestPojo("u1", 11));
			KeyedStateHandle snapshot2 = runSnapshot(backend.snapshot(
				682375462378L,
				2,
				streamFactory,
				CheckpointOptions.forCheckpoint()));
			snapshot2.registerSharedStates(sharedStateRegistry);
			snapshot.discardState();
			backend.dispose();
			env.getExecutionConfig().addDefaultKryoSerializer(TestPojo.class, (Class) CustomKryoTestSerializer.class);
			expectedException.expect(ExpectedKryoTestException.class);
			backend = restoreKeyedBackend(IntSerializer.INSTANCE, snapshot2, env);
			state = backend.getPartitionedState(VoidNamespace.INSTANCE, VoidNamespaceSerializer.INSTANCE, kvId);
			backend.setCurrentKey(1);
			state.value();
			snapshot2.discardState();
			backend.dispose();
		} finally {
			IOUtils.closeQuietly(backend);
			backend.dispose();
		}
	}
@ Test @ SuppressWarnings ( "unchecked" ) public void testKryoRegisteringRestoreResilienceWithDefaultSerializer ( ) throws Exception { CheckpointStreamFactory streamFactory = createStreamFactory ( ) ; SharedStateRegistry sharedStateRegistry = new SharedStateRegistry ( ) ; Environment env = new DummyEnvironment ( "test" , 1 , 0 ) ; AbstractKeyedStateBackend < Integer > backend = null ; try { 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 ( ) ) ) ; snapshot . registerSharedStates ( sharedStateRegistry ) ; backend . dispose ( ) ; env . getExecutionConfig ( ) . addDefaultKryoSerializer ( TestPojo . class , ( Class ) CustomKryoTestSerializer . class ) ; backend = restoreKeyedBackend ( IntSerializer . INSTANCE , snapshot , env ) ; kvId = new ValueStateDescriptor < > ( "id" , pojoType ) ; state = backend . getPartitionedState ( VoidNamespace . INSTANCE , VoidNamespaceSerializer . INSTANCE , kvId ) ; backend . setCurrentKey ( 1 ) ; state . update ( new TestPojo ( "u1" , 11 ) ) ; KeyedStateHandle snapshot2 = runSnapshot ( backend . snapshot ( 682375462378L , 2 , streamFactory , CheckpointOptions . forCheckpoint ( ) ) ) ; snapshot2 . registerSharedStates ( sharedStateRegistry ) ; snapshot . discardState ( ) ; backend . dispose ( ) ; env . getExecutionConfig ( ) . addDefaultKryoSerializer ( TestPojo . class , ( Class ) CustomKryoTestSerializer . class ) ; expectedException . expect ( ExpectedKryoTestException . class ) ; backend = restoreKeyedBackend ( IntSerializer . INSTANCE , snapshot2 , env ) ; state = backend . getPartitionedState ( VoidNamespace . INSTANCE , VoidNamespaceSerializer . INSTANCE , kvId ) ; backend . setCurrentKey ( 1 ) ; state . value ( ) ; snapshot2 . discardState ( ) ; backend . dispose ( ) ; } finally { IOUtils . closeQuietly ( backend ) ; 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 a default serializer for the state type
Verify state restore resilience when : - snapshot was taken without any Kryo registrations specific serializers or default serializers for the state type - restored with a default serializer for the state type
// 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 - should use default serializer (ONLY SERIALIZATION) ========== // cast because our test serializer is not typed to TestPojo // re-initialize to ensure that we create the KryoSerializer from scratch, otherwise // initializeSerializerUnlessSet would not pick up our new config // update to test state backends that eagerly serialize, such as RocksDB // ========= restore snapshot - should use default serializer (FAIL ON DESERIALIZATION) ========= // cast because our test serializer is not typed to TestPojo // on the second restore, since the custom serializer will be used for // deserialization, we expect the deliberate failure to be thrown // state backends that eagerly deserializes (such as the memory state backend) will fail here // state backends that lazily deserializes (such as RocksDB) will fail here // ensure to release native resources even when we exit through exception