XComp commented on code in PR #29562:
URL: https://github.com/apache/flink/pull/29562#discussion_r4535730191
##########
flink-runtime/src/test/java/org/apache/flink/runtime/rpc/RpcEndpointTest.java:
##########
@@ -413,9 +416,107 @@ private static void testCancelScheduledTask(
assertThat((Future<?>) scheduledFuture).isCancelled();
assertThat(actionFuture).isNotDone();
+ assertThat(gatewayFailures).isEmpty();
mainThreadExecutor.close();
}
+ @Test
+ void testScheduledRunnableFailureReachesTheMainThread() {
Review Comment:
The three new test methods are almost the same aside from the error type. We
could merge them into a parameterized test.
##########
flink-rpc/flink-rpc-core/src/main/java/org/apache/flink/runtime/rpc/RpcEndpoint.java:
##########
@@ -556,12 +563,38 @@ public <V> ScheduledFuture<V> schedule(Callable<V>
callable, long delay, TimeUni
"The scheduled executor service is shutdown and
ignores the callable {}",
callable);
} else {
- mainScheduledExecutor.schedule(
- () -> gateway.runAsync(ft), delayMillis,
TimeUnit.MILLISECONDS);
+ scheduleInMainThread(ft, delayMillis);
}
return new ScheduledFutureAdapter<>(ft, delayMillis,
TimeUnit.MILLISECONDS);
}
+ private void scheduleInMainThread(FutureTask<?> task, long
delayMillis) {
+ mainScheduledExecutor.schedule(
+ () -> gateway.runAsync(() -> runAndRethrowFailure(task)),
+ delayMillis,
+ TimeUnit.MILLISECONDS);
+ }
+
+ /**
+ * A {@link FutureTask} stores what its task throws instead of
throwing it; rethrowing lets
+ * the main thread log it and exit the process on a fatal error, as
for {@link #execute}.
+ * The failure stays in the task for the holder of the returned future.
+ */
Review Comment:
I feel like we can remove this comment. The code of the method itself is
good enough and the method is a private method. WDYT?
--
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: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
