code: public ConnectedStreams<IN1, IN2> keyBy(int[] keyPositions1, int[] keyPositions2) {
  		return new ConnectedStreams<>(environment, inputStream1.keyBy(keyPositions1),
  				inputStream2.keyBy(keyPositions2));
  	}
docs: KeyBy operation for connected data stream. Assigns keys to the elements of
  input1 and input2 according to keyPositions1 and keyPositions2.
  
  @param keyPositions1
  The fields used to group the first input stream.
  @param keyPositions2
  The fields used to group the second input stream.
  @return The grouped {@link ConnectedStreams}