code: public static List<KeyGroupRange> createKeyGroupPartitions(int numberKeyGroups, int parallelism) {
  		Preconditions.checkArgument(numberKeyGroups >= parallelism);
  		List<KeyGroupRange> result = new ArrayList<>(parallelism);
  
  		for (int i = 0; i < parallelism; ++i) {
  			result.add(KeyGroupRangeAssignment.computeKeyGroupRangeForOperatorIndex(numberKeyGroups, parallelism, i));
  		}
  		return result;
  	}
docs: Groups the available set of key groups into key group partitions. A key group partition is
  the set of key groups which is assigned to the same task. Each set of the returned list
  constitutes a key group partition.
  <p>
  <b>IMPORTANT</b>: The assignment of key groups to partitions has to be in sync with the
  KeyGroupStreamPartitioner.
  
  @param numberKeyGroups Number of available key groups (indexed from 0 to numberKeyGroups - 1)
  @param parallelism     Parallelism to generate the key group partitioning for
  @return List of key group partitions