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