code: @SuppressWarnings("unchecked")
  	@Nonnull
  	@Override
  	public <T extends HeapPriorityQueueElement & PriorityComparable & Keyed> KeyGroupedInternalPriorityQueue<T> create(
  		@Nonnull String stateName,
  		@Nonnull TypeSerializer<T> byteOrderedElementSerializer) {
  
  		final HeapPriorityQueueSnapshotRestoreWrapper existingState = registeredPQStates.get(stateName);
  
  		if (existingState != null) {
  			// TODO we implement the simple way of supporting the current functionality, mimicking keyed state
  			// because this should be reworked in FLINK-9376 and then we should have a common algorithm over
  			// StateMetaInfoSnapshot that avoids this code duplication.
  
  			TypeSerializerSchemaCompatibility<T> compatibilityResult =
  				existingState.getMetaInfo().updateElementSerializer(byteOrderedElementSerializer);
  
  			if (compatibilityResult.isIncompatible()) {
  				throw new FlinkRuntimeException(new StateMigrationException("For heap backends, the new priority queue serializer must not be incompatible."));
  			} else {
  				registeredPQStates.put(
  					stateName,
  					existingState.forUpdatedSerializer(byteOrderedElementSerializer));
  			}
  
  			return existingState.getPriorityQueue();
  		} else {
  			final RegisteredPriorityQueueStateBackendMetaInfo<T> metaInfo =
  				new RegisteredPriorityQueueStateBackendMetaInfo<>(stateName, byteOrderedElementSerializer);
  			return createInternal(metaInfo);
  		}
  	}
docs: ------------------------------------------------------------------------