pnowojski commented on a change in pull request #15836:
URL: https://github.com/apache/flink/pull/15836#discussion_r626494148
##########
File path:
flink-streaming-java/src/test/java/org/apache/flink/streaming/api/operators/async/AsyncWaitOperatorTest.java
##########
@@ -241,6 +246,30 @@ public void timeout(Integer input, ResultFuture<Integer>
resultFuture) throws Ex
}
}
+ /**
+ * Checks if timeout has been called after the element has been completed
within the timeout.
+ */
+ private static class TimeoutAfterCompletionTestFunction
+ implements AsyncFunction<Integer, Integer> {
+ static final AtomicBoolean TIMED_OUT = new AtomicBoolean(false);
+
+ @Override
+ public void asyncInvoke(Integer input, ResultFuture<Integer>
resultFuture) {
+ ForkJoinPool.commonPool()
+ .submit(
+ () -> {
+ Thread.sleep(TIMEOUT / 2);
Review comment:
500ms might be low enough to cause test instabilities. Maybe instead of
away on a latch that you complete at the end of the unit test?
##########
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:
I'm also not sure if this will prove to be stable - I don't have a
concrete scenario, but generally speaking timeouts tend to brake.
Maybe replace it with:
```
while (timeOutHasNotHappened()) {
harness.processAll();
Thread.sleep(1);
}
assertFalse(...);
asyncFunctionWaitLatch.complete();
```
?
--
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]
