test_class.file: flink-streaming-java/src/test/java/org/apache/flink/streaming/runtime/operators/windowing/EvictingWindowOperatorTest.java
test_class.identifier: EvictingWindowOperatorTest
test_case.body: @Test
  	@SuppressWarnings("unchecked")
  	public void testCountTrigger() throws Exception {
  
  		final int WINDOW_SIZE = 4;
  		final int WINDOW_SLIDE = 2;
  
  		TypeInformation<Tuple2<String, Integer>> inputType = TypeInfoParser.parse("Tuple2<String, Integer>");
  
  
  		ListStateDescriptor<StreamRecord<Tuple2<String, Integer>>> stateDesc = new ListStateDescriptor<>("window-contents",
  			new StreamRecordSerializer<>(inputType.createSerializer(new ExecutionConfig())));
  
  
  		EvictingWindowOperator<String, Tuple2<String, Integer>, Tuple2<String, Integer>, GlobalWindow> operator = new EvictingWindowOperator<>(
  				GlobalWindows.create(),
  				new GlobalWindow.Serializer(),
  				new TupleKeySelector(),
  				BasicTypeInfo.STRING_TYPE_INFO.createSerializer(new ExecutionConfig()),
  				stateDesc,
  				new InternalIterableWindowFunction<>(new ReduceIterableWindowFunction<String, GlobalWindow, Tuple2<String, Integer>>(new SumReducer())),
  				CountTrigger.of(WINDOW_SLIDE),
  				CountEvictor.of(WINDOW_SIZE));
  
  		operator.setInputType(inputType, new ExecutionConfig());
  
  
  		OneInputStreamOperatorTestHarness<Tuple2<String, Integer>, Tuple2<String, Integer>> testHarness =
  				new OneInputStreamOperatorTestHarness<>(operator);
  
  		testHarness.configureForKeyedStream(new TupleKeySelector(), BasicTypeInfo.STRING_TYPE_INFO);
  
  		long initialTime = 0L;
  		ConcurrentLinkedQueue<Object> expectedOutput = new ConcurrentLinkedQueue<>();
  
  		testHarness.open();
  
  		// The global window actually ignores these timestamps...
  
  		// add elements out-of-order
  		testHarness.processElement(new StreamRecord<>(new Tuple2<>("key2", 1), initialTime + 3000));
  		testHarness.processElement(new StreamRecord<>(new Tuple2<>("key2", 1), initialTime + 3999));
  
  		testHarness.processElement(new StreamRecord<>(new Tuple2<>("key1", 1), initialTime + 20));
  		testHarness.processElement(new StreamRecord<>(new Tuple2<>("key1", 1), initialTime));
  		testHarness.processElement(new StreamRecord<>(new Tuple2<>("key1", 1), initialTime + 999));
  
  		testHarness.processElement(new StreamRecord<>(new Tuple2<>("key2", 1), initialTime + 1998));
  		testHarness.processElement(new StreamRecord<>(new Tuple2<>("key2", 1), initialTime + 1999));
  		testHarness.processElement(new StreamRecord<>(new Tuple2<>("key2", 1), initialTime + 1000));
  
  
  
  		expectedOutput.add(new StreamRecord<>(new Tuple2<>("key2", 2), Long.MAX_VALUE));
  		expectedOutput.add(new StreamRecord<>(new Tuple2<>("key2", 4), Long.MAX_VALUE));
  		expectedOutput.add(new StreamRecord<>(new Tuple2<>("key1", 2), Long.MAX_VALUE));
  
  		TestHarnessUtil.assertOutputEqualsSorted("Output was not correct.", expectedOutput, testHarness.getOutput(), new ResultSortComparator());
  
  		testHarness.processElement(new StreamRecord<>(new Tuple2<>("key1", 1), initialTime + 10999));
  		testHarness.processElement(new StreamRecord<>(new Tuple2<>("key2", 1), initialTime + 1000));
  
  		expectedOutput.add(new StreamRecord<>(new Tuple2<>("key1", 4), Long.MAX_VALUE));
  		expectedOutput.add(new StreamRecord<>(new Tuple2<>("key2", 4), Long.MAX_VALUE));
  
  		TestHarnessUtil.assertOutputEqualsSorted("Output was not correct.", expectedOutput, testHarness.getOutput(), new ResultSortComparator());
  
  		testHarness.close();
  	}
test_case.class_method_signature: EvictingWindowOperatorTest.testCountTrigger()
test_case.full_signature: @Test @SuppressWarnings("unchecked") public void testCountTrigger()
test_case.identifier: testCountTrigger
test_case.invocations: parse createSerializer create createSerializer of of setInputType configureForKeyedStream open processElement processElement processElement processElement processElement processElement processElement processElement add add add assertOutputEqualsSorted getOutput processElement processElement add add assertOutputEqualsSorted getOutput close
test_case.modifiers: @Test @SuppressWarnings("unchecked") public
test_case.parameters: ()
test_case.return: void
test_case.signature: void testCountTrigger()
focal_class.declarator: serialVersionUID = 1L
focal_class.modifier: private static final
focal_class.original_string: private static final long serialVersionUID = 1L;
focal_class.type: long
focal_class.var_name: serialVersionUID
focal_class.declarator: evictor
focal_class.modifier: private final
focal_class.original_string: private final Evictor<? super IN, ? super W> evictor;
focal_class.type: Evictor<? super IN, ? super W>
focal_class.var_name: evictor
focal_class.declarator: windowStateDescriptor
focal_class.modifier: private final
focal_class.original_string: private final StateDescriptor<? extends ListState<StreamRecord<IN>>, ?> windowStateDescriptor;
focal_class.type: StateDescriptor<? extends ListState<StreamRecord<IN>>, ?>
focal_class.var_name: windowStateDescriptor
focal_class.file: flink-streaming-java/src/main/java/org/apache/flink/streaming/runtime/operators/windowing/EvictingWindowOperator.java
focal_class.identifier: EvictingWindowOperator
focal_class.class_method_signature: EvictingWindowOperator.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)
focal_class.full_signature: 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)
focal_class.identifier: EvictingWindowOperator
focal_class.modifiers: public
focal_class.parameters: (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)
focal_class.signature:  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)
focal_class.class_method_signature: EvictingWindowOperator.processElement(StreamRecord<IN> element)
focal_class.full_signature: @Override @SuppressWarnings("unchecked") public void processElement(StreamRecord<IN> element)
focal_class.identifier: processElement
focal_class.modifiers: @Override @SuppressWarnings("unchecked") public
focal_class.parameters: (StreamRecord<IN> element)
focal_class.return: void
focal_class.signature: void processElement(StreamRecord<IN> element)
focal_class.class_method_signature: EvictingWindowOperator.processTriggerResult(TriggerResult triggerResult, K key, W window)
focal_class.full_signature: @Override @SuppressWarnings("unchecked,rawtypes") protected void processTriggerResult(TriggerResult triggerResult, K key, W window)
focal_class.identifier: processTriggerResult
focal_class.modifiers: @Override @SuppressWarnings("unchecked,rawtypes") protected
focal_class.parameters: (TriggerResult triggerResult, K key, W window)
focal_class.return: void
focal_class.signature: void processTriggerResult(TriggerResult triggerResult, K key, W window)
focal_class.class_method_signature: EvictingWindowOperator.getEvictor()
focal_class.full_signature: @VisibleForTesting public Evictor<? super IN, ? super W> getEvictor()
focal_class.identifier: getEvictor
focal_class.modifiers: @VisibleForTesting public
focal_class.parameters: ()
focal_class.return: Evictor<? super IN, ? super W>
focal_class.signature: Evictor<? super IN, ? super W> getEvictor()
focal_class.class_method_signature: EvictingWindowOperator.getStateDescriptor()
focal_class.full_signature: @Override @VisibleForTesting @SuppressWarnings("unchecked, rawtypes") public StateDescriptor<? extends MergingState<IN, Iterable<IN>>, ?> getStateDescriptor()
focal_class.identifier: getStateDescriptor
focal_class.modifiers: @Override @VisibleForTesting @SuppressWarnings("unchecked, rawtypes") public
focal_class.parameters: ()
focal_class.return: StateDescriptor<? extends MergingState<IN, Iterable<IN>>, ?>
focal_class.signature: StateDescriptor<? extends MergingState<IN, Iterable<IN>>, ?> getStateDescriptor()
focal_class.superclass: extends WindowOperator<K, IN, Iterable<IN>, OUT, W>
focal_method.body: @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);
  		}
  	}
focal_method.class_method_signature: EvictingWindowOperator.processElement(StreamRecord<IN> element)
focal_method.full_signature: @Override @SuppressWarnings("unchecked") public void processElement(StreamRecord<IN> element)
focal_method.identifier: processElement
focal_method.invocations: assignWindows getValue getTimestamp getCurrentKey getStateBackend getPartitionedState add onElement processTriggerResult
focal_method.modifiers: @Override @SuppressWarnings("unchecked") public
focal_method.parameters: (StreamRecord<IN> element)
focal_method.return: void
focal_method.signature: void processElement(StreamRecord<IN> element)