docs: @Internal
  	public <R> SingleOutputStreamOperator<R> reduce(ReduceFunction<T> reduceFunction, ProcessWindowFunction<T, R, K, W> function, TypeInformation<R> resultType) {
  		if (reduceFunction instanceof RichFunction) {
  			throw new UnsupportedOperationException("ReduceFunction of apply can not be a RichFunction.");
  		}
  		//clean the closures
  		function = input.getExecutionEnvironment().clean(function);
  		reduceFunction = input.getExecutionEnvironment().clean(reduceFunction);
  
  		final String opName = generateOperatorName(windowAssigner, trigger, evictor, reduceFunction, function);
  		KeySelector<T, K> keySel = input.getKeySelector();
  
  		OneInputStreamOperator<T, R> operator;
  
  		if (evictor != null) {
  			@SuppressWarnings({"unchecked", "rawtypes"})
  			TypeSerializer<StreamRecord<T>> streamRecordSerializer =
  					(TypeSerializer<StreamRecord<T>>) new StreamElementSerializer(input.getType().createSerializer(getExecutionEnvironment().getConfig()));
  
  			ListStateDescriptor<StreamRecord<T>> stateDesc =
  					new ListStateDescriptor<>("window-contents", streamRecordSerializer);
  
  			operator =
  					new EvictingWindowOperator<>(windowAssigner,
  							windowAssigner.getWindowSerializer(getExecutionEnvironment().getConfig()),
  							keySel,
  							input.getKeyType().createSerializer(getExecutionEnvironment().getConfig()),
  							stateDesc,
  							new InternalIterableProcessWindowFunction<>(new ReduceApplyProcessWindowFunction<>(reduceFunction, function)),
  							trigger,
  							evictor,
  							allowedLateness,
  							lateDataOutputTag);
  
  		} else {
  			ReducingStateDescriptor<T> stateDesc = new ReducingStateDescriptor<>("window-contents",
  					reduceFunction,
  					input.getType().createSerializer(getExecutionEnvironment().getConfig()));
  
  			operator =
  					new WindowOperator<>(windowAssigner,
  							windowAssigner.getWindowSerializer(getExecutionEnvironment().getConfig()),
  							keySel,
  							input.getKeyType().createSerializer(getExecutionEnvironment().getConfig()),
  							stateDesc,
  							new InternalSingleValueProcessWindowFunction<>(function),
  							trigger,
  							allowedLateness,
  							lateDataOutputTag);
  		}
  
  		return input.transform(opName, resultType, operator);
  	}