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 testCountTriggerWithApply() throws Exception {
  		AtomicInteger closeCalled = new AtomicInteger(0);
  
  		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 RichSumReducer<GlobalWindow>(closeCalled)),
  			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();
  
  		Assert.assertEquals("Close was not called.", 1, closeCalled.get());
  	}
test_case.class_method_signature: EvictingWindowOperatorTest.testCountTriggerWithApply()
test_case.full_signature: @Test @SuppressWarnings("unchecked") public void testCountTriggerWithApply()
test_case.identifier: testCountTriggerWithApply
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 assertEquals get
test_case.modifiers: @Test @SuppressWarnings("unchecked") public
test_case.parameters: ()
test_case.return: void
test_case.signature: void testCountTriggerWithApply()
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)