code: public KeyedStream<T, Tuple> keyBy(int... fields) {
  		if (getType() instanceof BasicArrayTypeInfo || getType() instanceof PrimitiveArrayTypeInfo) {
  			return keyBy(KeySelectorUtil.getSelectorForArray(fields, getType()));
  		} else {
  			return keyBy(new Keys.ExpressionKeys<>(fields, getType()));
  		}
  	}
docs: Partitions the operator state of a {@link DataStream} by the given key positions.
  
  @param fields
  The position of the fields on which the {@link DataStream}
  will be grouped.
  @return The {@link DataStream} with partitioned state (i.e. KeyedStream)