Github user tillrohrmann commented on a diff in the pull request:
https://github.com/apache/flink/pull/6333#discussion_r202553765
--- Diff:
flink-runtime/src/main/java/org/apache/flink/runtime/state/heap/HeapKeyedStateBackend.java
---
@@ -342,16 +379,20 @@ private void
restorePartitionedState(Collection<KeyedStateHandle> state) throws
for (StateMetaInfoSnapshot restoredMetaInfo :
restoredMetaInfos) {
restoredKvStateMetaInfos.put(restoredMetaInfo.getName(), restoredMetaInfo);
- StateTable<K, ?, ?> stateTable =
stateTables.get(restoredMetaInfo.getName());
+ StateSnapshotRestore snapshotRestore =
registeredStates.get(restoredMetaInfo.getName());
//important: only create a new table we
did not already create it previously
- if (null == stateTable) {
+ if (null == snapshotRestore) {
-
RegisteredKeyedBackendStateMetaInfo<?, ?> registeredKeyedBackendStateMetaInfo =
- new
RegisteredKeyedBackendStateMetaInfo<>(restoredMetaInfo);
+ if
(restoredMetaInfo.getBackendStateType() ==
StateMetaInfoSnapshot.BackendStateType.KEY_VALUE) {
+
RegisteredKeyValueStateBackendMetaInfo<?, ?>
registeredKeyedBackendStateMetaInfo =
+ new
RegisteredKeyValueStateBackendMetaInfo<>(restoredMetaInfo);
- stateTable =
snapshotStrategy.newStateTable(registeredKeyedBackendStateMetaInfo);
-
stateTables.put(restoredMetaInfo.getName(), stateTable);
+ snapshotRestore =
snapshotStrategy.newStateTable(registeredKeyedBackendStateMetaInfo);
+
registeredStates.put(restoredMetaInfo.getName(), snapshotRestore);
+ } else {
--- End diff --
Maybe check that `(restoredMetaInfo.getBackendStateType() == PRIORITY_QUEUE`
---
