docs: @PublicEvolving
  	public <ACC, V, R> SingleOutputStreamOperator<R> aggregate(
  			AggregateFunction<T, ACC, V> aggregateFunction,
  			WindowFunction<V, R, K, W> windowFunction,
  			TypeInformation<ACC> accumulatorType,
  			TypeInformation<R> resultType) {
  
  		checkNotNull(aggregateFunction, "aggregateFunction");
  		checkNotNull(windowFunction, "windowFunction");
  		checkNotNull(accumulatorType, "accumulatorType");
  		checkNotNull(resultType, "resultType");
  
  		if (aggregateFunction instanceof RichFunction) {
  			throw new UnsupportedOperationException("This aggregate function cannot be a RichFunction.");
  		}
  
  		//clean the closures
  		windowFunction = input.getExecutionEnvironment().clean(windowFunction);
  		aggregateFunction = input.getExecutionEnvironment().clean(aggregateFunction);
  
  		final String opName = generateOperatorName(windowAssigner, trigger, evictor, aggregateFunction, windowFunction);
  		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 InternalIterableWindowFunction<>(new AggregateApplyWindowFunction<>(aggregateFunction, windowFunction)),
  					trigger,
  					evictor,
  					allowedLateness,
  					lateDataOutputTag);
  
  		} else {
  			AggregatingStateDescriptor<T, ACC, V> stateDesc = new AggregatingStateDescriptor<>("window-contents",
  					aggregateFunction, accumulatorType.createSerializer(getExecutionEnvironment().getConfig()));
  
  			operator = new WindowOperator<>(windowAssigner,
  					windowAssigner.getWindowSerializer(getExecutionEnvironment().getConfig()),
  					keySel,
  					input.getKeyType().createSerializer(getExecutionEnvironment().getConfig()),
  					stateDesc,
  					new InternalSingleValueWindowFunction<>(windowFunction),
  					trigger,
  					allowedLateness,
  					lateDataOutputTag);
  		}
  
  		return input.transform(opName, resultType, operator);
  	}