private void restoreKVStateMetaData() throws IOException, StateMigrationException, RocksDBException {
			KeyedBackendSerializationProxy<K> serializationProxy =
					new KeyedBackendSerializationProxy<>(rocksDBKeyedStateBackend.userCodeClassLoader);
			serializationProxy.read(currentStateHandleInView);
			// check for key serializer compatibility; this also reconfigures the
			// key serializer to be compatible, if it is required and is possible
			if (CompatibilityUtil.resolveCompatibilityResult(
					serializationProxy.getKeySerializer(),
					UnloadableDummyTypeSerializer.class,
					serializationProxy.getKeySerializerConfigSnapshot(),
					rocksDBKeyedStateBackend.keySerializer)
				.isRequiresMigration()) {
				// TODO replace with state migration; note that key hash codes need to remain the same after migration
				throw new StateMigrationException("The new key serializer is not compatible to read previous keys. " +
					"Aborting now since state migration is currently not available");
			}
			this.keygroupStreamCompressionDecorator = serializationProxy.isUsingKeyGroupCompression() ?
				SnappyStreamCompressionDecorator.INSTANCE : UncompressedStreamCompressionDecorator.INSTANCE;
			List<RegisteredKeyedBackendStateMetaInfo.Snapshot<?, ?>> restoredMetaInfos =
					serializationProxy.getStateMetaInfoSnapshots();
			currentStateHandleKVStateColumnFamilies = new ArrayList<>(restoredMetaInfos.size());
			//rocksDBKeyedStateBackend.restoredKvStateMetaInfos = new HashMap<>(restoredMetaInfos.size());
			for (RegisteredKeyedBackendStateMetaInfo.Snapshot<?, ?> restoredMetaInfo : restoredMetaInfos) {
				Tuple2<ColumnFamilyHandle, RegisteredKeyedBackendStateMetaInfo<?, ?>> registeredColumn =
					rocksDBKeyedStateBackend.kvStateInformation.get(restoredMetaInfo.getName());
				if (registeredColumn == null) {
					ColumnFamilyDescriptor columnFamilyDescriptor = new ColumnFamilyDescriptor(
						restoredMetaInfo.getName().getBytes(ConfigConstants.DEFAULT_CHARSET),
						rocksDBKeyedStateBackend.columnOptions);
					RegisteredKeyedBackendStateMetaInfo<?, ?> stateMetaInfo =
							new RegisteredKeyedBackendStateMetaInfo<>(
								restoredMetaInfo.getStateType(),
								restoredMetaInfo.getName(),
								restoredMetaInfo.getNamespaceSerializer(),
								restoredMetaInfo.getStateSerializer());
					rocksDBKeyedStateBackend.restoredKvStateMetaInfos.put(restoredMetaInfo.getName(), restoredMetaInfo);
					ColumnFamilyHandle columnFamily = rocksDBKeyedStateBackend.db.createColumnFamily(columnFamilyDescriptor);
					registeredColumn = new Tuple2<ColumnFamilyHandle, RegisteredKeyedBackendStateMetaInfo<?, ?>>(columnFamily, stateMetaInfo);
					rocksDBKeyedStateBackend.kvStateInformation.put(stateMetaInfo.getName(), registeredColumn);
				} else {
					// TODO with eager state registration in place, check here for serializer migration strategies
				}
				currentStateHandleKVStateColumnFamilies.add(registeredColumn.f0);
			}
		}
private void restoreKVStateMetaData() throws IOException, StateMigrationException, RocksDBException {
			KeyedBackendSerializationProxy<K> serializationProxy =
					new KeyedBackendSerializationProxy<>(rocksDBKeyedStateBackend.userCodeClassLoader);
			serializationProxy.read(currentStateHandleInView);
		
		
			if (CompatibilityUtil.resolveCompatibilityResult(
					serializationProxy.getKeySerializer(),
					UnloadableDummyTypeSerializer.class,
					serializationProxy.getKeySerializerConfigSnapshot(),
					rocksDBKeyedStateBackend.keySerializer)
				.isRequiresMigration()) {
			
				throw new StateMigrationException("The new key serializer is not compatible to read previous keys. " +
					"Aborting now since state migration is currently not available");
			}
			this.keygroupStreamCompressionDecorator = serializationProxy.isUsingKeyGroupCompression() ?
				SnappyStreamCompressionDecorator.INSTANCE : UncompressedStreamCompressionDecorator.INSTANCE;
			List<RegisteredKeyedBackendStateMetaInfo.Snapshot<?, ?>> restoredMetaInfos =
					serializationProxy.getStateMetaInfoSnapshots();
			currentStateHandleKVStateColumnFamilies = new ArrayList<>(restoredMetaInfos.size());
		
			for (RegisteredKeyedBackendStateMetaInfo.Snapshot<?, ?> restoredMetaInfo : restoredMetaInfos) {
				Tuple2<ColumnFamilyHandle, RegisteredKeyedBackendStateMetaInfo<?, ?>> registeredColumn =
					rocksDBKeyedStateBackend.kvStateInformation.get(restoredMetaInfo.getName());
				if (registeredColumn == null) {
					ColumnFamilyDescriptor columnFamilyDescriptor = new ColumnFamilyDescriptor(
						restoredMetaInfo.getName().getBytes(ConfigConstants.DEFAULT_CHARSET),
						rocksDBKeyedStateBackend.columnOptions);
					RegisteredKeyedBackendStateMetaInfo<?, ?> stateMetaInfo =
							new RegisteredKeyedBackendStateMetaInfo<>(
								restoredMetaInfo.getStateType(),
								restoredMetaInfo.getName(),
								restoredMetaInfo.getNamespaceSerializer(),
								restoredMetaInfo.getStateSerializer());
					rocksDBKeyedStateBackend.restoredKvStateMetaInfos.put(restoredMetaInfo.getName(), restoredMetaInfo);
					ColumnFamilyHandle columnFamily = rocksDBKeyedStateBackend.db.createColumnFamily(columnFamilyDescriptor);
					registeredColumn = new Tuple2<ColumnFamilyHandle, RegisteredKeyedBackendStateMetaInfo<?, ?>>(columnFamily, stateMetaInfo);
					rocksDBKeyedStateBackend.kvStateInformation.put(stateMetaInfo.getName(), registeredColumn);
				} else {
				
				}
				currentStateHandleKVStateColumnFamilies.add(registeredColumn.f0);
			}
		}
private void restorekvstatemetadata() throws ioexception, statemigrationexception, rocksdbexception { keyedbackendserializationproxy<k> serializationproxy = new keyedbackendserializationproxy<>(rocksdbkeyedstatebackend.usercodeclassloader); serializationproxy.read(currentstatehandleinview); if (compatibilityutil.resolvecompatibilityresult( serializationproxy.getkeyserializer(), unloadabledummytypeserializer.class, serializationproxy.getkeyserializerconfigsnapshot(), rocksdbkeyedstatebackend.keyserializer) .isrequiresmigration()) { throw new statemigrationexception("the new key serializer is not compatible to read previous keys. " + "aborting now since state migration is currently not available"); } this.keygroupstreamcompressiondecorator = serializationproxy.isusingkeygroupcompression() ? snappystreamcompressiondecorator.instance : uncompressedstreamcompressiondecorator.instance; list<registeredkeyedbackendstatemetainfo.snapshot<?, ?>> restoredmetainfos = serializationproxy.getstatemetainfosnapshots(); currentstatehandlekvstatecolumnfamilies = new arraylist<>(restoredmetainfos.size()); for (registeredkeyedbackendstatemetainfo.snapshot<?, ?> restoredmetainfo : restoredmetainfos) { tuple2<columnfamilyhandle, registeredkeyedbackendstatemetainfo<?, ?>> registeredcolumn = rocksdbkeyedstatebackend.kvstateinformation.get(restoredmetainfo.getname()); if (registeredcolumn == null) { columnfamilydescriptor columnfamilydescriptor = new columnfamilydescriptor( restoredmetainfo.getname().getbytes(configconstants.default_charset), rocksdbkeyedstatebackend.columnoptions); registeredkeyedbackendstatemetainfo<?, ?> statemetainfo = new registeredkeyedbackendstatemetainfo<>( restoredmetainfo.getstatetype(), restoredmetainfo.getname(), restoredmetainfo.getnamespaceserializer(), restoredmetainfo.getstateserializer()); rocksdbkeyedstatebackend.restoredkvstatemetainfos.put(restoredmetainfo.getname(), restoredmetainfo); columnfamilyhandle columnfamily = rocksdbkeyedstatebackend.db.createcolumnfamily(columnfamilydescriptor); registeredcolumn = new tuple2<columnfamilyhandle, registeredkeyedbackendstatemetainfo<?, ?>>(columnfamily, statemetainfo); rocksdbkeyedstatebackend.kvstateinformation.put(statemetainfo.getname(), registeredcolumn); } else { } currentstatehandlekvstatecolumnfamilies.add(registeredcolumn.f0); } }