Github user aljoscha commented on a diff in the pull request:
https://github.com/apache/flink/pull/2736#discussion_r86388455
--- Diff:
flink-streaming-java/src/test/java/org/apache/flink/streaming/runtime/operators/windowing/EvictingWindowOperatorTest.java
---
@@ -58,6 +61,458 @@
// For counting if close() is called the correct number of times on the
SumReducer
+ /**
+ * Tests CountEvictor evictAfter behavior
+ * @throws Exception
+ */
+ @Test
+ public void testCountEvictorEvictAfter() throws Exception {
+ AtomicInteger closeCalled = new AtomicInteger(0);
+ final int WINDOW_SIZE = 4;
+ final int TRIGGER_COUNT = 2;
+ final boolean EVICT_AFTER = true;
+
+ TypeInformation<Tuple2<String, Integer>> inputType =
TypeInfoParser.parse("Tuple2<String, Integer>");
+
+ @SuppressWarnings({"unchecked", "rawtypes"})
+ TypeSerializer<StreamRecord<Tuple2<String, Integer>>>
streamRecordSerializer =
+ (TypeSerializer<StreamRecord<Tuple2<String, Integer>>>)
new StreamElementSerializer(inputType.createSerializer(new ExecutionConfig()));
+
+ ListStateDescriptor<StreamRecord<Tuple2<String, Integer>>>
stateDesc =
+ new ListStateDescriptor<>("window-contents",
streamRecordSerializer);
+
+
+ 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(TRIGGER_COUNT),
+ CountEvictor.of(WINDOW_SIZE,EVICT_AFTER),
+ 0);
+
+
+ OneInputStreamOperatorTestHarness<Tuple2<String, Integer>,
Tuple2<String, Integer>> testHarness =
+ new KeyedOneInputStreamOperatorTestHarness<>(operator,
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", 6),
Long.MAX_VALUE));
+
+ TestHarnessUtil.assertOutputEqualsSorted("Output was not
correct.", expectedOutput, testHarness.getOutput(), new ResultSortComparator());
--- End diff --
As it is now, it doesn't seem this ever evicts elements because eviction
happens after triggering. If you add these lines that would exercise the
eviction:
```
testHarness.processElement(new StreamRecord<>(new Tuple2<>("key2", 1),
initialTime + 1000));
testHarness.processElement(new StreamRecord<>(new Tuple2<>("key2", 1),
initialTime + 1000));
expectedOutput.add(new StreamRecord<>(new Tuple2<>("key2", 6),
Long.MAX_VALUE));
TestHarnessUtil.assertOutputEqualsSorted("Output was not correct.",
expectedOutput, testHarness.getOutput(), new ResultSortComparator());
```
It checks whether the previous firing with 6 elements truncated back to 4.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at [email protected] or file a JIRA ticket
with INFRA.
---
