code: @Internal
  	public <R> SingleOutputStreamOperator<R> process(
  			CoProcessFunction<IN1, IN2, R> coProcessFunction,
  			TypeInformation<R> outputType) {
  
  		TwoInputStreamOperator<IN1, IN2, R> operator;
  
  		if ((inputStream1 instanceof KeyedStream) && (inputStream2 instanceof KeyedStream)) {
  			operator = new KeyedCoProcessOperator<>(inputStream1.clean(coProcessFunction));
  		} else {
  			operator = new CoProcessOperator<>(inputStream1.clean(coProcessFunction));
  		}
  
  		return transform("Co-Process", outputType, operator);
  	}
docs: Applies the given {@link CoProcessFunction} on the connected input streams,
  thereby creating a transformed output stream.
  
  <p>The function will be called for every element in the input streams and can produce zero
  or more output elements. Contrary to the {@link #flatMap(CoFlatMapFunction)} function,
  this function can also query the time and set timers. When reacting to the firing of set
  timers the function can directly emit elements and/or register yet more timers.
  
  @param coProcessFunction The {@link CoProcessFunction} that is called for each element
  in the stream.
  
  @param <R> The type of elements emitted by the {@code CoProcessFunction}.
  
  @return The transformed {@link DataStream}.