[FLINK-9887][state] Integrate priority queue state with existing seri…

…alizer upgrade mechanism

This closes apache#6467.

flink-core/src/main/java/org/apache/flink/util/StateMigrationException.java

@@ -24,6 +24,8 @@
 public class StateMigrationException extends FlinkException {
  private static final long serialVersionUID = 8268516412747670839L;
 
+ public static final String MIGRATION_NOT_SUPPORTED_MSG = "State migration is currently not supported.";
+
  public StateMigrationException(String message) {
  super(message);
  }
@@ -35,4 +37,8 @@ public StateMigrationException(Throwable cause) {
  public StateMigrationException(String message, Throwable cause) {
  super(message, cause);
  }
+
+ public static StateMigrationException notSupported() {
+ return new StateMigrationException(MIGRATION_NOT_SUPPORTED_MSG);
+ }
 }

flink-runtime/src/main/java/org/apache/flink/runtime/state/AbstractKeyedStateBackend.java

@@ -313,7 +313,7 @@ StreamCompressionDecorator getKeyGroupCompressionDecorator() {
  * Returns the total number of state entries across all keys/namespaces.
  */
  @VisibleForTesting
- public abstract int numStateEntries();
+ public abstract int numKeyValueStateEntries();
 
  // TODO remove this once heap-based timers are working with RocksDB incremental snapshots!
  public boolean requiresLegacySynchronousTimerSnapshots() {

flink-runtime/src/main/java/org/apache/flink/runtime/state/DefaultOperatorStateBackend.java

@@ -260,7 +260,7 @@ public <K, V> BroadcastState<K, V> getBroadcastState(final MapStateDescriptor<K,
  // the new serializer; we're deliberately failing here for now to have equal functionality with
  // the RocksDB backend to avoid confusion for users.
 
- throw new StateMigrationException("State migration isn't supported, yet.");
+ throw StateMigrationException.notSupported();
  }
  }
 
@@ -781,7 +781,7 @@ private <S> ListState<S> getListState(
  // the new serializer; we're deliberately failing here for now to have equal functionality with
  // the RocksDB backend to avoid confusion for users.
 
- throw new StateMigrationException("State migration isn't supported, yet.");
+ throw StateMigrationException.notSupported();
  }
  }
 

flink-runtime/src/main/java/org/apache/flink/runtime/state/RegisteredKeyValueStateBackendMetaInfo.java

@@ -144,6 +144,12 @@ public static <N, S> RegisteredKeyValueStateBackendMetaInfo<N, S> resolveKvState
  TypeSerializer<N> newNamespaceSerializer,
  StateDescriptor<?, S> newStateDescriptor) throws StateMigrationException {
 
+ Preconditions.checkState(restoredStateMetaInfoSnapshot.getBackendStateType()
+ == StateMetaInfoSnapshot.BackendStateType.KEY_VALUE,
+ "Incompatible state types. " +
+ "Was [" + restoredStateMetaInfoSnapshot.getBackendStateType() + "], " +
+ "registered as [" + StateMetaInfoSnapshot.BackendStateType.KEY_VALUE + "].");
+
  Preconditions.checkState(
  Objects.equals(newStateDescriptor.getName(), restoredStateMetaInfoSnapshot.getName()),
  "Incompatible state names. " +
@@ -160,7 +166,7 @@ public static <N, S> RegisteredKeyValueStateBackendMetaInfo<N, S> resolveKvState
 
  Preconditions.checkState(
  newStateDescriptor.getType() == restoredType,
- "Incompatible state types. " +
+ "Incompatible key/value state types. " +
  "Was [" + restoredType + "], " +
  "registered with [" + newStateDescriptor.getType() + "].");
  }
@@ -184,7 +190,7 @@ public static <N, S> RegisteredKeyValueStateBackendMetaInfo<N, S> resolveKvState
 
  if (namespaceCompatibility.isRequiresMigration() || stateCompatibility.isRequiresMigration()) {
  // TODO state migration currently isn't possible.
- throw new StateMigrationException("State migration isn't supported, yet.");
+ throw StateMigrationException.notSupported();
  } else {
  return new RegisteredKeyValueStateBackendMetaInfo<>(
  newStateDescriptor.getType(),