docs: private void repartitionSplitState(
  			Map<String, List<Tuple2<StreamStateHandle, OperatorStateHandle.StateMetaInfo>>> nameToDistributeState,
  			int newParallelism,
  			List<Map<StreamStateHandle, OperatorStateHandle>> mergeMapList) {
  
  		int startParallelOp = 0;
  		// Iterate all named states and repartition one named state at a time per iteration
  		for (Map.Entry<String, List<Tuple2<StreamStateHandle, OperatorStateHandle.StateMetaInfo>>> e :
  				nameToDistributeState.entrySet()) {
  
  			List<Tuple2<StreamStateHandle, OperatorStateHandle.StateMetaInfo>> current = e.getValue();
  
  			// Determine actual number of partitions for this named state
  			int totalPartitions = 0;
  			for (Tuple2<StreamStateHandle, OperatorStateHandle.StateMetaInfo> offsets : current) {
  				totalPartitions += offsets.f1.getOffsets().length;
  			}
  
  			// Repartition the state across the parallel operator instances
  			int lstIdx = 0;
  			int offsetIdx = 0;
  			int baseFraction = totalPartitions / newParallelism;
  			int remainder = totalPartitions % newParallelism;
  
  			int newStartParallelOp = startParallelOp;
  
  			for (int i = 0; i < newParallelism; ++i) {
  
  				// Preparation: calculate the actual index considering wrap around
  				int parallelOpIdx = (i + startParallelOp) % newParallelism;
  
  				// Now calculate the number of partitions we will assign to the parallel instance in this round ...
  				int numberOfPartitionsToAssign = baseFraction;
  
  				// ... and distribute odd partitions while we still have some, one at a time
  				if (remainder > 0) {
  					++numberOfPartitionsToAssign;
  					--remainder;
  				} else if (remainder == 0) {
  					// We are out of odd partitions now and begin our next redistribution round with the current
  					// parallel operator to ensure fair load balance
  					newStartParallelOp = parallelOpIdx;
  					--remainder;
  				}
  
  				// Now start collection the partitions for the parallel instance into this list
  
  				while (numberOfPartitionsToAssign > 0) {
  					Tuple2<StreamStateHandle, OperatorStateHandle.StateMetaInfo> handleWithOffsets =
  							current.get(lstIdx);
  
  					long[] offsets = handleWithOffsets.f1.getOffsets();
  					int remaining = offsets.length - offsetIdx;
  					// Repartition offsets
  					long[] offs;
  					if (remaining > numberOfPartitionsToAssign) {
  						offs = Arrays.copyOfRange(offsets, offsetIdx, offsetIdx + numberOfPartitionsToAssign);
  						offsetIdx += numberOfPartitionsToAssign;
  					} else {
  						if (OPTIMIZE_MEMORY_USE) {
  							handleWithOffsets.f1 = null; // GC
  						}
  						offs = Arrays.copyOfRange(offsets, offsetIdx, offsets.length);
  						offsetIdx = 0;
  						++lstIdx;
  					}
  
  					numberOfPartitionsToAssign -= remaining;
  
  					// As a last step we merge partitions that use the same StreamStateHandle in a single
  					// OperatorStateHandle
  					Map<StreamStateHandle, OperatorStateHandle> mergeMap = mergeMapList.get(parallelOpIdx);
  					OperatorStateHandle operatorStateHandle = mergeMap.get(handleWithOffsets.f0);
  					if (operatorStateHandle == null) {
  						operatorStateHandle = new OperatorStreamStateHandle(
  							new HashMap<>(nameToDistributeState.size()),
  							handleWithOffsets.f0);
  						mergeMap.put(handleWithOffsets.f0, operatorStateHandle);
  					}
  					operatorStateHandle.getStateNameToPartitionOffsets().put(
  							e.getKey(),
  							new OperatorStateHandle.StateMetaInfo(offs, OperatorStateHandle.Mode.SPLIT_DISTRIBUTE));
  				}
  			}
  			startParallelOp = newStartParallelOp;
  			e.setValue(null);
  		}
  	}