[GitHub] [flink] zjureel commented on a change in pull request #18303: [FLINK-25085][runtime] Add a scheduled thread pool for periodic tasks in RpcEndpoint

[GitBox <gi...@apache.org>]

zjureel commented on a change in pull request #18303:
URL: https://github.com/apache/flink/pull/18303#discussion_r821275803



##########
File path: flink-rpc/flink-rpc-core/src/main/java/org/apache/flink/runtime/rpc/RpcEndpoint.java
##########
@@ -401,38 +451,90 @@ public void validateRunsInMainThread() {
 
     /** Executor which executes runnables in the main thread context. */
     protected static class MainThreadExecutor implements ComponentMainThreadExecutor {
+        private static final Logger log = LoggerFactory.getLogger(MainThreadExecutor.class);
 
         private final MainThreadExecutable gateway;
         private final Runnable mainThreadCheck;
-
-        MainThreadExecutor(MainThreadExecutable gateway, Runnable mainThreadCheck) {
+        /**
+         * The main scheduled executor manages the scheduled tasks and send them to gateway when
+         * they should be executed.
+         */
+        private final ScheduledThreadPoolExecutor mainScheduledExecutor;
+
+        MainThreadExecutor(
+                MainThreadExecutable gateway, Runnable mainThreadCheck, String endpointId) {
             this.gateway = Preconditions.checkNotNull(gateway);
             this.mainThreadCheck = Preconditions.checkNotNull(mainThreadCheck);
-        }
-
-        private void scheduleRunAsync(Runnable runnable, long delayMillis) {
-            gateway.scheduleRunAsync(runnable, delayMillis);
+            this.mainScheduledExecutor =
+                    new ScheduledThreadPoolExecutor(
+                            1, new ExecutorThreadFactory(endpointId + "-main-scheduler"));
+            this.mainScheduledExecutor.setRemoveOnCancelPolicy(true);
         }
 
         @Override
         public void execute(@Nonnull Runnable command) {
             gateway.runAsync(command);
         }
 
+        /**
+         * The mainScheduledExecutor manages the task and sends it to the gateway after the given
+         * delay.
+         *
+         * @param command the task to execute in the future
+         * @param delay the time from now to delay the execution
+         * @param unit the time unit of the delay parameter
+         * @return a ScheduledFuture representing the completion of the scheduled task
+         */
         @Override
         public ScheduledFuture<?> schedule(Runnable command, long delay, TimeUnit unit) {
-            final long delayMillis = TimeUnit.MILLISECONDS.convert(delay, unit);
-            FutureTask<Void> ft = new FutureTask<>(command, null);
-            scheduleRunAsync(ft, delayMillis);
-            return new ScheduledFutureAdapter<>(ft, delayMillis, TimeUnit.MILLISECONDS);
+            // If the scheduled executor service is shutdown, the command won't be executed.
+            if (mainScheduledExecutor.isShutdown()) {
+                log.warn(
+                        "The scheduled executor service is shutdown and return throwing scheduled future for command {}",
+                        command);
+                return ThrowingScheduledFuture.getInstance();
+            } else {
+                final long delayMillis = TimeUnit.MILLISECONDS.convert(delay, unit);
+                FutureTask<Void> ft = new FutureTask<>(command, null);
+                ScheduledFuture<?> scheduledFuture =
+                        mainScheduledExecutor.schedule(
+                                () -> gateway.scheduleRunAsync(ft, 0L),
+                                delayMillis,
+                                TimeUnit.MILLISECONDS);
+                return new ScheduledFutureAdapter<>(
+                        scheduledFuture, ft, delayMillis, TimeUnit.MILLISECONDS);
+            }
         }
 
+        /**
+         * The mainScheduledExecutor manages the given callable and sends it to the gateway after
+         * the given delay. The result of the callable is returned as a {@link ScheduledFuture}.
+         *
+         * @param callable the callable to execute
+         * @param delay the time from now to delay the execution
+         * @param unit the time unit of the delay parameter
+         * @param <V> result type of the callable
+         * @return a ScheduledFuture which holds the future value of the given callable
+         */
         @Override
         public <V> ScheduledFuture<V> schedule(Callable<V> callable, long delay, TimeUnit unit) {
-            final long delayMillis = TimeUnit.MILLISECONDS.convert(delay, unit);
-            FutureTask<V> ft = new FutureTask<>(callable);
-            scheduleRunAsync(ft, delayMillis);
-            return new ScheduledFutureAdapter<>(ft, delayMillis, TimeUnit.MILLISECONDS);
+            // If the scheduled executor service is shutdown, the callable won't be executed.
+            if (mainScheduledExecutor.isShutdown()) {
+                log.warn(
+                        "The scheduled executor service is shutdown and return throwing scheduled future for callable {}",
+                        callable);
+                return ThrowingScheduledFuture.getInstance();

Review comment:
       @xintongsong Sorry to reply later. In fact we can't throw exception here even when the mainScheduledExecutor is shutdown. For example, when the check idle slot task is executed in akka thread pool, it will submit the next task, and if we throw the exception here, it may cause the job failed before the job is terminated. 
   In flink runtime project, it most uses main thread executor to schedule the periodic tasks such as resource checker, heartbeat and ect. These tasks don't need to check the result future of 'schedule' method and add a warn log is enough for them.
   For the other tasks with delay==0, we can submit them to akka thread pool even the mainScheduledExecutor is shutdown. What do you think? THX @xintongsong 




-- 
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