class EvictingWindowOperator extends WindowOperator<K, IN, Iterable<IN>, OUT, W> {

	@Override
	@SuppressWarnings("unchecked")
	public void processElement(StreamRecord<IN> element) throws Exception {
		Collection<W> elementWindows = windowAssigner.assignWindows(element.getValue(), element.getTimestamp());

		K key = (K) getStateBackend().getCurrentKey();

		for (W window: elementWindows) {

			ListState<StreamRecord<IN>> windowState = getPartitionedState(window, windowSerializer,
				windowStateDescriptor);

			windowState.add(element);

			context.key = key;
			context.window = window;
			TriggerResult triggerResult = context.onElement(element);

			processTriggerResult(triggerResult, key, window);
		}
	}

	public  EvictingWindowOperator(WindowAssigner<? super IN, W> windowAssigner,
		TypeSerializer<W> windowSerializer,
		KeySelector<IN, K> keySelector,
		TypeSerializer<K> keySerializer,
		StateDescriptor<? extends ListState<StreamRecord<IN>>, ?> windowStateDescriptor,
		InternalWindowFunction<Iterable<IN>, OUT, K, W> windowFunction,
		Trigger<? super IN, ? super W> trigger,
		Evictor<? super IN, ? super W> evictor);

}