AHeise commented on a change in pull request #15836:
URL: https://github.com/apache/flink/pull/15836#discussion_r626660347
##########
File path:
flink-streaming-java/src/test/java/org/apache/flink/streaming/api/operators/async/AsyncWaitOperatorTest.java
##########
@@ -783,6 +812,28 @@ public void testTimeoutCleanup() throws Exception {
assertEquals(0,
harness.getProcessingTimeService().getNumActiveTimers());
}
+ @Test
+ public void testTimeoutAfterComplete() throws Exception {
+ StreamTaskMailboxTestHarnessBuilder<Integer> builder =
+ new StreamTaskMailboxTestHarnessBuilder<>(
+ OneInputStreamTask::new,
BasicTypeInfo.INT_TYPE_INFO)
+ .addInput(BasicTypeInfo.INT_TYPE_INFO);
+ TimeoutAfterCompletionTestFunction.TIMED_OUT.set(false);
+ try (StreamTaskMailboxTestHarness<Integer> harness =
+ builder.setupOutputForSingletonOperatorChain(
+ new AsyncWaitOperatorFactory<>(
+ new
TimeoutAfterCompletionTestFunction(),
+ TIMEOUT,
+ 1,
+ AsyncDataStream.OutputMode.UNORDERED))
+ .build()) {
+ harness.processElement(new StreamRecord<>(1));
+ Thread.sleep(2 * TIMEOUT);
Review comment:
The test is testing that **no** timeout happens, so it's tricky to test.
I now had the idea to just register a timer in the test that is known to be
processed after the timer under test. So we just wait until this timer has been
processed and we should get a stable test. PTAL
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
[email protected]
