AHeise commented on a change in pull request #14573: URL: https://github.com/apache/flink/pull/14573#discussion_r553285345
########## File path: flink-streaming-java/src/test/java/org/apache/flink/streaming/runtime/tasks/StreamTaskTest.java ########## @@ -1256,54 +1309,97 @@ 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); }; + waitingThread = + new WaitingThread( + executor, + completeFutureTask, + sleepTimeInsideMail, + sleepTimeOutsideMail); + // 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."); - }, + waitingThread::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(sleepTimeOutsideMail)); assertThat( ioMetricGroup.getBackPressuredTimePerSecond().getCount(), - Matchers.greaterThanOrEqualTo(sleepTime)); + Matchers.lessThanOrEqualTo(totalDuration - sleepTimeInsideMail)); assertThat(ioMetricGroup.getIdleTimeMsPerSecond().getCount(), is(0L)); assertEquals(numberOfProcessCalls, inputProcessor.currentNumProcessCalls); + } finally { + if (waitingThread != null) { Review comment: I was thinking after `task.invoke()` but I guess this is working as well. ########## File path: flink-streaming-java/src/test/java/org/apache/flink/streaming/runtime/tasks/StreamTaskTest.java ########## @@ -1241,10 +1241,63 @@ public void testProcessWithAvailableOutput() throws Exception { } } + /** + * In this weird construct, we are: + * + * <ul> + * <li>1. We start a thread, which will... + * <li>2. ... sleep for X ms, and enqueue another mail, that will... + * <li>3. ... sleep for Y ms, and make the output available again + * </ul> + * + * <p>2nd step is to check that back pressure or idle counter is at least X. In the last 3rd + * step, we test whether this counter was paused for the duration of processing mails. + */ + private static class WaitingThread extends Thread { + private final MailboxExecutor executor; + private final RunnableWithException resumeTask; + private final long sleepTimeInsideMail; + private final long sleepTimeOutsideMail; + + @Nullable private Exception asyncException; + + public WaitingThread( + MailboxExecutor executor, + RunnableWithException resumeTask, + long sleepTimeInsideMail, + long sleepTimeOutsideMail) { + this.executor = executor; + this.resumeTask = resumeTask; + this.sleepTimeInsideMail = sleepTimeInsideMail; + this.sleepTimeOutsideMail = sleepTimeOutsideMail; + } + + @Override + public void run() { + try { + Thread.sleep(sleepTimeOutsideMail); + } catch (InterruptedException e) { + asyncException = e; + } + executor.submit( + () -> { + if (asyncException != null) { Review comment: Is this working visibility-wise? `asyncException` should probably be volatile. ---------------------------------------------------------------- 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