code: private static <IN, OUT> SingleOutputStreamOperator<OUT> addOperator(
  			DataStream<IN> in,
  			AsyncFunction<IN, OUT> func,
  			long timeout,
  			int bufSize,
  			OutputMode mode) {
  
  		TypeInformation<OUT> outTypeInfo = TypeExtractor.getUnaryOperatorReturnType(
  			func,
  			AsyncFunction.class,
  			0,
  			1,
  			new int[]{1, 0},
  			in.getType(),
  			Utils.getCallLocationName(),
  			true);
  
  		// create transform
  		AsyncWaitOperator<IN, OUT> operator = new AsyncWaitOperator<>(
  			in.getExecutionEnvironment().clean(func),
  			timeout,
  			bufSize,
  			mode);
  
  		return in.transform("async wait operator", outTypeInfo, operator);
  	}
docs: Add an AsyncWaitOperator.
  
  @param in The {@link DataStream} where the {@link AsyncWaitOperator} will be added.
  @param func {@link AsyncFunction} wrapped inside {@link AsyncWaitOperator}.
  @param timeout for the asynchronous operation to complete
  @param bufSize The max number of inputs the {@link AsyncWaitOperator} can hold inside.
  @param mode Processing mode for {@link AsyncWaitOperator}.
  @param <IN> Input type.
  @param <OUT> Output type.
  @return A new {@link SingleOutputStreamOperator}