code: private List<Map<StreamStateHandle, OperatorStateHandle>> repartition(
  			GroupByStateNameResults nameToStateByMode,
  			int newParallelism) {
  
  		// We will use this to merge w.r.t. StreamStateHandles for each parallel subtask inside the maps
  		List<Map<StreamStateHandle, OperatorStateHandle>> mergeMapList = new ArrayList<>(newParallelism);
  
  		// Initialize
  		for (int i = 0; i < newParallelism; ++i) {
  			mergeMapList.add(new HashMap<>());
  		}
  
  		// Start with the state handles we distribute round robin by splitting by offsets
  		Map<String, List<Tuple2<StreamStateHandle, OperatorStateHandle.StateMetaInfo>>> nameToDistributeState =
  				nameToStateByMode.getByMode(OperatorStateHandle.Mode.SPLIT_DISTRIBUTE);
  
  		repartitionSplitState(nameToDistributeState, newParallelism, mergeMapList);
  
  		// Now we also add the state handles marked for union to all parallel instances
  		Map<String, List<Tuple2<StreamStateHandle, OperatorStateHandle.StateMetaInfo>>> nameToUnionState =
  				nameToStateByMode.getByMode(OperatorStateHandle.Mode.UNION);
  
  		repartitionUnionState(nameToUnionState, mergeMapList);
  
  		// Now we also add the state handles marked for uniform broadcast to all parallel instances
  		Map<String, List<Tuple2<StreamStateHandle, OperatorStateHandle.StateMetaInfo>>> nameToBroadcastState =
  				nameToStateByMode.getByMode(OperatorStateHandle.Mode.BROADCAST);
  
  		repartitionBroadcastState(nameToBroadcastState, mergeMapList);
  
  		return mergeMapList;
  	}
docs: Repartition all named states.