private List<RegisteredKeyedBackendStateMetaInfo.Snapshot<?, ?>> readMetaData(
				StreamStateHandle metaStateHandle) throws Exception {
			FSDataInputStream inputStream = null;
			try {
				inputStream = metaStateHandle.openInputStream();
				stateBackend.cancelStreamRegistry.registerClosable(inputStream);
				KeyedBackendSerializationProxy<T> serializationProxy =
					new KeyedBackendSerializationProxy<>(stateBackend.userCodeClassLoader);
				DataInputView in = new DataInputViewStreamWrapper(inputStream);
				serializationProxy.read(in);
				// 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(),
						stateBackend.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");
				}
				return serializationProxy.getStateMetaInfoSnapshots();
			} finally {
				if (inputStream != null) {
					stateBackend.cancelStreamRegistry.unregisterClosable(inputStream);
					inputStream.close();
				}
			}
		}
private List<RegisteredKeyedBackendStateMetaInfo.Snapshot<?, ?>> readMetaData(
				StreamStateHandle metaStateHandle) throws Exception {
			FSDataInputStream inputStream = null;
			try {
				inputStream = metaStateHandle.openInputStream();
				stateBackend.cancelStreamRegistry.registerClosable(inputStream);
				KeyedBackendSerializationProxy<T> serializationProxy =
					new KeyedBackendSerializationProxy<>(stateBackend.userCodeClassLoader);
				DataInputView in = new DataInputViewStreamWrapper(inputStream);
				serializationProxy.read(in);
			
			
				if (CompatibilityUtil.resolveCompatibilityResult(
						serializationProxy.getKeySerializer(),
						UnloadableDummyTypeSerializer.class,
						serializationProxy.getKeySerializerConfigSnapshot(),
						stateBackend.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");
				}
				return serializationProxy.getStateMetaInfoSnapshots();
			} finally {
				if (inputStream != null) {
					stateBackend.cancelStreamRegistry.unregisterClosable(inputStream);
					inputStream.close();
				}
			}
		}
private list<registeredkeyedbackendstatemetainfo.snapshot<?, ?>> readmetadata( streamstatehandle metastatehandle) throws exception { fsdatainputstream inputstream = null; try { inputstream = metastatehandle.openinputstream(); statebackend.cancelstreamregistry.registerclosable(inputstream); keyedbackendserializationproxy<t> serializationproxy = new keyedbackendserializationproxy<>(statebackend.usercodeclassloader); datainputview in = new datainputviewstreamwrapper(inputstream); serializationproxy.read(in); if (compatibilityutil.resolvecompatibilityresult( serializationproxy.getkeyserializer(), unloadabledummytypeserializer.class, serializationproxy.getkeyserializerconfigsnapshot(), statebackend.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"); } return serializationproxy.getstatemetainfosnapshots(); } finally { if (inputstream != null) { statebackend.cancelstreamregistry.unregisterclosable(inputstream); inputstream.close(); } } }