docs: @SuppressWarnings("unchecked, rawtype")
  	private GroupByStateNameResults groupByStateMode(List<List<OperatorStateHandle>> previousParallelSubtaskStates) {
  
  		//Reorganize: group by (State Name -> StreamStateHandle + StateMetaInfo)
  		EnumMap<OperatorStateHandle.Mode,
  				Map<String, List<Tuple2<StreamStateHandle, OperatorStateHandle.StateMetaInfo>>>> nameToStateByMode =
  				new EnumMap<>(OperatorStateHandle.Mode.class);
  
  		for (OperatorStateHandle.Mode mode : OperatorStateHandle.Mode.values()) {
  
  			nameToStateByMode.put(
  					mode,
  					new HashMap<>());
  		}
  
  		for (List<OperatorStateHandle> previousParallelSubtaskState : previousParallelSubtaskStates) {
  			for (OperatorStateHandle operatorStateHandle : previousParallelSubtaskState) {
  
  				if (operatorStateHandle == null) {
  					continue;
  				}
  
  				final Set<Map.Entry<String, OperatorStateHandle.StateMetaInfo>> partitionOffsetEntries =
  					operatorStateHandle.getStateNameToPartitionOffsets().entrySet();
  
  				for (Map.Entry<String, OperatorStateHandle.StateMetaInfo> e : partitionOffsetEntries) {
  					OperatorStateHandle.StateMetaInfo metaInfo = e.getValue();
  
  					Map<String, List<Tuple2<StreamStateHandle, OperatorStateHandle.StateMetaInfo>>> nameToState =
  						nameToStateByMode.get(metaInfo.getDistributionMode());
  
  					List<Tuple2<StreamStateHandle, OperatorStateHandle.StateMetaInfo>> stateLocations =
  						nameToState.computeIfAbsent(
  							e.getKey(),
  							k -> new ArrayList<>(previousParallelSubtaskStates.size() * partitionOffsetEntries.size()));
  
  					stateLocations.add(Tuple2.of(operatorStateHandle.getDelegateStateHandle(), e.getValue()));
  				}
  			}
  		}
  
  		return new GroupByStateNameResults(nameToStateByMode);
  	}