code: private void repartitionUnionState(
  			Map<String, List<Tuple2<StreamStateHandle, OperatorStateHandle.StateMetaInfo>>> unionState,
  			List<Map<StreamStateHandle, OperatorStateHandle>> mergeMapList) {
  
  		for (Map<StreamStateHandle, OperatorStateHandle> mergeMap : mergeMapList) {
  			for (Map.Entry<String, List<Tuple2<StreamStateHandle, OperatorStateHandle.StateMetaInfo>>> e :
  					unionState.entrySet()) {
  
  				for (Tuple2<StreamStateHandle, OperatorStateHandle.StateMetaInfo> handleWithMetaInfo : e.getValue()) {
  					OperatorStateHandle operatorStateHandle = mergeMap.get(handleWithMetaInfo.f0);
  					if (operatorStateHandle == null) {
  						operatorStateHandle = new OperatorStreamStateHandle(
  							new HashMap<>(unionState.size()),
  							handleWithMetaInfo.f0);
  						mergeMap.put(handleWithMetaInfo.f0, operatorStateHandle);
  					}
  					operatorStateHandle.getStateNameToPartitionOffsets().put(e.getKey(), handleWithMetaInfo.f1);
  				}
  			}
  		}
  	}
docs: Repartition UNION state.