pnowojski commented on a change in pull request #14573:
URL: https://github.com/apache/flink/pull/14573#discussion_r553247220
##########
File path:
flink-streaming-java/src/test/java/org/apache/flink/streaming/runtime/tasks/StreamTaskTest.java
##########
@@ -1256,49 +1307,82 @@ public void testProcessWithUnAvailableOutput() throws
Exception {
final RunnableWithException completeFutureTask =
() -> {
- Thread.sleep(sleepTime + 1);
assertEquals(1, inputProcessor.currentNumProcessCalls);
assertTrue(task.mailboxProcessor.isDefaultActionUnavailable());
environment.getWriter(1).getAvailableFuture().complete(null);
};
+ // Make sure WaitingThread is started after Task starts processing.
executor.submit(
- () -> {
- executor.submit(
- completeFutureTask,
- "This task will complete the future to resume
process input action.");
- },
+ () ->
+ new WaitingThread(
+ executor,
+ completeFutureTask,
+ sleepTimeInsideMail,
+ sleepTimeOutsideMail)
+ .start(),
"This task will submit another task to execute after
processing input once.");
+ long startTs = System.currentTimeMillis();
TaskIOMetricGroup ioMetricGroup =
task.getEnvironment().getMetricGroup().getIOMetricGroup();
task.invoke();
+ long totalDuration = System.currentTimeMillis() - startTs;
assertThat(
ioMetricGroup.getBackPressuredTimePerSecond().getCount(),
- Matchers.greaterThanOrEqualTo(sleepTime));
+ Matchers.greaterThanOrEqualTo(sleepTimeOutsideMail));
+ assertThat(
+ ioMetricGroup.getBackPressuredTimePerSecond().getCount(),
+ Matchers.lessThanOrEqualTo(totalDuration -
sleepTimeInsideMail));
Review comment:
The test is failing without the fix. `WaitingThread` gives us 2
guarantees. It will wait X ms before submitting a mailbox action to unblock
task, and it will sleep Y ms inside that mailbox action. This directly
translates that we will have
```
idleTime > X // that's first assertion in the test, it was being tested on
master before my change
busyTime > Y
```
On the other hand we know
```
totalDuration > busyTime + idleTime
```
so...
```
totalDuration - idleTime > busyTime > Y
totalDuration - Y > idleTime // that's the new assertion in the test
```
So as long as we can assume elapsed time measured around `Thread.sleep(X)`
by java yields at least `X`, the test should be stable. I've also run in a loop
couple of hundreds times.
----------------------------------------------------------------
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:
us...@infra.apache.org
