// TODO replace with state migration; note that key hash codes need to remain the same after migration
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();
				}
			}
		}
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");
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);
// 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();
}
}
}