code: public static List<KeyedStateHandle> getManagedKeyedStateHandles(
  		OperatorState operatorState,
  		KeyGroupRange subtaskKeyGroupRange) {
  
  		final int parallelism = operatorState.getParallelism();
  
  		List<KeyedStateHandle> subtaskKeyedStateHandles = null;
  
  		for (int i = 0; i < parallelism; i++) {
  			if (operatorState.getState(i) != null) {
  
  				Collection<KeyedStateHandle> keyedStateHandles = operatorState.getState(i).getManagedKeyedState();
  
  				if (subtaskKeyedStateHandles == null) {
  					subtaskKeyedStateHandles = new ArrayList<>(parallelism * keyedStateHandles.size());
  				}
  
  				extractIntersectingState(
  					keyedStateHandles,
  					subtaskKeyGroupRange,
  					subtaskKeyedStateHandles);
  			}
  		}
  
  		return subtaskKeyedStateHandles;
  	}
docs: Collect {@link KeyGroupsStateHandle  managedKeyedStateHandles} which have intersection with given
  {@link KeyGroupRange} from {@link TaskState operatorState}
  
  @param operatorState        all state handles of a operator
  @param subtaskKeyGroupRange the KeyGroupRange of a subtask
  @return all managedKeyedStateHandles which have intersection with given KeyGroupRange