AHeise commented on a change in pull request #14573:
URL: https://github.com/apache/flink/pull/14573#discussion_r553528099
##########
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:
Ah I was irritated taht this is a field, would probably easier to
understand with local variable.
----------------------------------------------------------------
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
