You are viewing a plain text version of this content. The canonical link for it is here.
Posted to issues@flink.apache.org by GitBox <gi...@apache.org> on 2021/08/04 10:23:57 UTC
[GitHub] [flink] curcur commented on a change in pull request #16653: [FLINK-23558][streaming] Ignoring RejectedExecutionException during s…
curcur commented on a change in pull request #16653:
URL: https://github.com/apache/flink/pull/16653#discussion_r682472467
##########
File path: flink-streaming-java/src/test/java/org/apache/flink/streaming/runtime/tasks/StreamTaskTest.java
##########
@@ -1635,6 +1636,56 @@ public void testRestorePerformedFromInvoke() throws Exception {
assertThat(task.streamTask.restoreInvocationCount, is(1));
}
+ @Test
+ public void testQuiesceOfMailboxRightBeforeSubmittingActionViaTimerService() throws Exception {
+ // given: the stream task with configured handle async exception.
+ AtomicBoolean submitThroughputFail = new AtomicBoolean();
+ MockEnvironment mockEnvironment = new MockEnvironmentBuilder().build();
+
+ final UnAvailableTestInputProcessor inputProcessor = new UnAvailableTestInputProcessor();
+ RunningTask<StreamTask<?, ?>> task =
+ runTask(
+ () ->
+ new MockStreamTaskBuilder(mockEnvironment)
+ .setHandleAsyncException(
+ (str, t) -> submitThroughputFail.set(true))
+ .setStreamInputProcessor(inputProcessor)
+ .build());
+
+ waitTaskIsRunning(task.streamTask, task.invocationFuture);
+
+ TimerService timerService = task.streamTask.systemTimerService;
+ MailboxExecutor mainMailboxExecutor =
+ task.streamTask.mailboxProcessor.getMainMailboxExecutor();
+
+ CountDownLatch stoppintMailboxLatch = new CountDownLatch(1);
Review comment:
stopping?
##########
File path: flink-streaming-java/src/test/java/org/apache/flink/streaming/runtime/tasks/StreamTaskTest.java
##########
@@ -1778,8 +1829,10 @@ public long calculateThroughput() {
})
.build()) {
+ TaskMailboxImpl taskMailbox = new TaskMailboxImpl();
StreamTask task =
new MockStreamTaskBuilder(mockEnvironment)
+ .setTaskMailbox(taskMailbox)
Review comment:
should be removed.
##########
File path: flink-streaming-java/src/test/java/org/apache/flink/streaming/runtime/tasks/StreamTaskTest.java
##########
@@ -1635,6 +1636,56 @@ public void testRestorePerformedFromInvoke() throws Exception {
assertThat(task.streamTask.restoreInvocationCount, is(1));
}
+ @Test
+ public void testQuiesceOfMailboxRightBeforeSubmittingActionViaTimerService() throws Exception {
+ // given: the stream task with configured handle async exception.
+ AtomicBoolean submitThroughputFail = new AtomicBoolean();
+ MockEnvironment mockEnvironment = new MockEnvironmentBuilder().build();
+
+ final UnAvailableTestInputProcessor inputProcessor = new UnAvailableTestInputProcessor();
+ RunningTask<StreamTask<?, ?>> task =
+ runTask(
+ () ->
+ new MockStreamTaskBuilder(mockEnvironment)
+ .setHandleAsyncException(
+ (str, t) -> submitThroughputFail.set(true))
+ .setStreamInputProcessor(inputProcessor)
+ .build());
+
+ waitTaskIsRunning(task.streamTask, task.invocationFuture);
+
+ TimerService timerService = task.streamTask.systemTimerService;
+ MailboxExecutor mainMailboxExecutor =
+ task.streamTask.mailboxProcessor.getMainMailboxExecutor();
+
+ CountDownLatch stoppintMailboxLatch = new CountDownLatch(1);
+ timerService.registerTimer(
+ timerService.getCurrentProcessingTime(),
+ (time) -> {
+ stoppintMailboxLatch.await();
+ // The time to the start 'afterInvoke' inside of mailbox.
+ // 'afterInvoke' won't finish until this execution won't finish so it is
+ // impossible to wait on latch or something else.
+ Thread.sleep(5);
+ mainMailboxExecutor.submit(() -> {}, "test");
+ });
+
+ // when: Calling the quiesce for mailbox and finishing the timer service.
+ mainMailboxExecutor
+ .submit(
+ () -> {
+ stoppintMailboxLatch.countDown();
+ task.streamTask.afterInvoke();
+ },
+ "test")
+ .get();
+
+ // then: the exception handle wasn't invoked because the such situation is expected.
+ assertFalse(submitThroughputFail.get());
+
+ inputProcessor.availabilityProvider.getUnavailableToResetAvailable().complete(null);
Review comment:
why do we need this?
--
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.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org