code: public static KeyGroupRange computeKeyGroupRangeForOperatorIndex(
  		int maxParallelism,
  		int parallelism,
  		int operatorIndex) {
  
  		checkParallelismPreconditions(parallelism);
  		checkParallelismPreconditions(maxParallelism);
  
  		Preconditions.checkArgument(maxParallelism >= parallelism,
  			"Maximum parallelism must not be smaller than parallelism.");
  
  		int start = ((operatorIndex * maxParallelism + parallelism - 1) / parallelism);
  		int end = ((operatorIndex + 1) * maxParallelism - 1) / parallelism;
  		return new KeyGroupRange(start, end);
  	}
docs: Computes the range of key-groups that are assigned to a given operator under the given parallelism and maximum
  parallelism.
  
  IMPORTANT: maxParallelism must be <= Short.MAX_VALUE to avoid rounding problems in this method. If we ever want
  to go beyond this boundary, this method must perform arithmetic on long values.
  
  @param maxParallelism Maximal parallelism that the job was initially created with.
  @param parallelism    The current parallelism under which the job runs. Must be <= maxParallelism.
  @param operatorIndex  Id of a key-group. 0 <= keyGroupID < maxParallelism.
  @return the computed key-group range for the operator.