code: @Override
  	public <K> AbstractKeyedStateBackend<K> createKeyedStateBackend(
  		Environment env,
  		JobID jobID,
  		String operatorIdentifier,
  		TypeSerializer<K> keySerializer,
  		int numberOfKeyGroups,
  		KeyGroupRange keyGroupRange,
  		TaskKvStateRegistry kvStateRegistry,
  		TtlTimeProvider ttlTimeProvider,
  		MetricGroup metricGroup,
  		@Nonnull Collection<KeyedStateHandle> stateHandles,
  		CloseableRegistry cancelStreamRegistry) throws BackendBuildingException {
  
  		TaskStateManager taskStateManager = env.getTaskStateManager();
  		LocalRecoveryConfig localRecoveryConfig = taskStateManager.createLocalRecoveryConfig();
  		HeapPriorityQueueSetFactory priorityQueueSetFactory =
  			new HeapPriorityQueueSetFactory(keyGroupRange, numberOfKeyGroups, 128);
  
  		return new HeapKeyedStateBackendBuilder<>(
  			kvStateRegistry,
  			keySerializer,
  			env.getUserClassLoader(),
  			numberOfKeyGroups,
  			keyGroupRange,
  			env.getExecutionConfig(),
  			ttlTimeProvider,
  			stateHandles,
  			AbstractStateBackend.getCompressionDecorator(env.getExecutionConfig()),
  			localRecoveryConfig,
  			priorityQueueSetFactory,
  			isUsingAsynchronousSnapshots(),
  			cancelStreamRegistry).build();
  	}
docs: ------------------------------------------------------------------------