docs: private Map<String, List<Tuple2<StreamStateHandle, OperatorStateHandle.StateMetaInfo>>> collectUnionStates(
  		List<List<OperatorStateHandle>> parallelSubtaskStates) {
  
  		Map<String, List<Tuple2<StreamStateHandle, OperatorStateHandle.StateMetaInfo>>> unionStates =
  			new HashMap<>(parallelSubtaskStates.size());
  
  		for (List<OperatorStateHandle> subTaskState : parallelSubtaskStates) {
  			for (OperatorStateHandle operatorStateHandle : subTaskState) {
  				if (operatorStateHandle == null) {
  					continue;
  				}
  
  				final Set<Map.Entry<String, OperatorStateHandle.StateMetaInfo>> partitionOffsetEntries =
  					operatorStateHandle.getStateNameToPartitionOffsets().entrySet();
  
  				partitionOffsetEntries.stream()
  					.filter(entry -> entry.getValue().getDistributionMode().equals(OperatorStateHandle.Mode.UNION))
  					.forEach(entry -> {
  						List<Tuple2<StreamStateHandle, OperatorStateHandle.StateMetaInfo>> stateLocations =
  							unionStates.computeIfAbsent(entry.getKey(), k -> new ArrayList<>(parallelSubtaskStates.size() * partitionOffsetEntries.size()));
  
  						stateLocations.add(Tuple2.of(operatorStateHandle.getDelegateStateHandle(), entry.getValue()));
  					});
  			}
  		}
  
  		return unionStates;
  	}