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); } } 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); @Override @SuppressWarnings("unchecked") void processElement(StreamRecord<IN> element); @VisibleForTesting Evictor<? super IN, ? super W> getEvictor(); @Override @VisibleForTesting @SuppressWarnings("unchecked, rawtypes") StateDescriptor<? extends MergingState<IN, Iterable<IN>>, ?> getStateDescriptor();  }
@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(); 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 @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(); 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()); }