[hotfix] Restore KeySerializer only once

flink-runtime/src/main/java/org/apache/flink/runtime/state/heap/HeapKeyedStateBackend.java

@@ -362,6 +362,8 @@ public class HeapKeyedStateBackend<K> extends AbstractKeyedStateBackend<K> {
 		int numRegisteredKvStates = 0;
 		stateTables.clear();
 
+		boolean keySerializerRestored = false;
+
 		for (KeyedStateHandle keyedStateHandle : state) {
 
 			if (keyedStateHandle == null) {
@@ -386,20 +388,24 @@ public class HeapKeyedStateBackend<K> extends AbstractKeyedStateBackend<K> {
 
 				serializationProxy.read(inView);
 
-				// check for key serializer compatibility; this also reconfigures the
-				// key serializer to be compatible, if it is required and is possible
-				if (StateMigrationUtil.resolveCompatibilityResult(
+				if (!keySerializerRestored) {
+					// check for key serializer compatibility; this also reconfigures the
+					// key serializer to be compatible, if it is required and is possible
+					if (StateMigrationUtil.resolveCompatibilityResult(
 						serializationProxy.getKeySerializer(),
 						TypeSerializerSerializationProxy.ClassNotFoundDummyTypeSerializer.class,
 						serializationProxy.getKeySerializerConfigSnapshot(),
 						(TypeSerializer) keySerializer)
-					.isRequiresMigration()) {
+						.isRequiresMigration()) {
 
-					// TODO replace with state migration; note that key hash codes need to remain the same after migration
-					throw new RuntimeException("The new key serializer is not compatible to read previous keys. " +
-						"Aborting now since state migration is currently not available");
-				}
+						// TODO replace with state migration; note that key hash codes need to remain the same after migration
+						throw new RuntimeException("The new key serializer is not compatible to read previous keys. " +
+							"Aborting now since state migration is currently not available");
+					}
 
+					keySerializerRestored = true;
+				}
+				
 				List<RegisteredKeyedBackendStateMetaInfo.Snapshot<?, ?>> restoredMetaInfos =
 						serializationProxy.getStateMetaInfoSnapshots();