Bubble-up exceptions from scheduler

qa/evil-tests/src/test/java/org/elasticsearch/threadpool/EvilThreadPoolTests.java

@@ -19,19 +19,13 @@
 
 package org.elasticsearch.threadpool;
 
-import org.apache.logging.log4j.Level;
-import org.apache.logging.log4j.LogManager;
-import org.apache.logging.log4j.Logger;
-import org.apache.logging.log4j.core.LogEvent;
-import org.elasticsearch.common.logging.Loggers;
 import org.elasticsearch.common.settings.Settings;
 import org.elasticsearch.common.unit.TimeValue;
 import org.elasticsearch.common.util.concurrent.AbstractRunnable;
 import org.elasticsearch.common.util.concurrent.EsExecutors;
 import org.elasticsearch.common.util.concurrent.EsThreadPoolExecutor;
 import org.elasticsearch.common.util.concurrent.PrioritizedEsThreadPoolExecutor;
 import org.elasticsearch.test.ESTestCase;
-import org.elasticsearch.test.MockLogAppender;
 import org.junit.After;
 import org.junit.Before;
 
@@ -44,7 +38,6 @@
 import java.util.function.Consumer;
 
 import static org.hamcrest.Matchers.containsString;
-import static org.hamcrest.Matchers.equalTo;
 import static org.hamcrest.Matchers.hasToString;
 import static org.hamcrest.Matchers.instanceOf;
 
@@ -367,63 +360,28 @@ private void runExecutionTest(
                 uncaughtExceptionHandlerLatch.countDown();
             });
 
-
             final CountDownLatch supplierLatch = new CountDownLatch(1);
 
-            Runnable job = () -> {
-                try {
-                    runnable.run();
-                } finally {
-                    supplierLatch.countDown();
-                }
-            };
-
-            // snoop on logging to also handle the cases where exceptions are simply logged in Scheduler.
-            final Logger schedulerLogger = LogManager.getLogger(Scheduler.SafeScheduledThreadPoolExecutor.class);
-            final MockLogAppender appender = new MockLogAppender();
-            appender.addExpectation(
-                new MockLogAppender.LoggingExpectation() {
-                    @Override
-                    public void match(LogEvent event) {
-                        if (event.getLevel() == Level.WARN) {
-                            assertThat("no other warnings than those expected",
-                                event.getMessage().getFormattedMessage(),
-                                equalTo("uncaught exception in scheduled thread [" + Thread.currentThread().getName() + "]"));
-                            assertTrue(expectThrowable);
-                            assertNotNull(event.getThrown());
-                            assertTrue("only one message allowed", throwableReference.compareAndSet(null, event.getThrown()));
-                            uncaughtExceptionHandlerLatch.countDown();
-                        }
-                    }
-
-                    @Override
-                    public void assertMatched() {
+            try {
+                runner.accept(() -> {
+                    try {
+                        runnable.run();
+                    } finally {
+                        supplierLatch.countDown();
                     }
                 });
+            } catch (Throwable t) {
+                consumer.accept(Optional.of(t));
+                return;
+            }
 
-            appender.start();
-            Loggers.addAppender(schedulerLogger, appender);
-            try {
-                try {
-                    runner.accept(job);
-                } catch (Throwable t) {
-                    consumer.accept(Optional.of(t));
-                    return;
-                }
-
-                supplierLatch.await();
+            supplierLatch.await();
 
-                if (expectThrowable) {
-                    uncaughtExceptionHandlerLatch.await();
-                }
-            } finally {
-                Loggers.removeAppender(schedulerLogger, appender);
-                appender.stop();
+            if (expectThrowable) {
+                uncaughtExceptionHandlerLatch.await();
             }
 
             consumer.accept(Optional.ofNullable(throwableReference.get()));
-        } catch (IllegalAccessException e) {
-            throw new RuntimeException(e);
         } finally {
             Thread.setDefaultUncaughtExceptionHandler(uncaughtExceptionHandler);
         }

server/src/main/java/org/elasticsearch/index/shard/GlobalCheckpointListeners.java

@@ -135,7 +135,7 @@ synchronized void add(final long waitingForGlobalCheckpoint, final GlobalCheckpo
                                                  * before we could be cancelled by the notification. In this case, our listener here would
                                                  * not be in the map and we should not fire the timeout logic.
                                                  */
-                                                removed = listeners.remove(listener).v2() != null;
+                                                removed = listeners.remove(listener) != null;
                                             }
                                             if (removed) {
                                                 final TimeoutException e = new TimeoutException(timeout.getStringRep());

server/src/main/java/org/elasticsearch/threadpool/Scheduler.java

@@ -21,7 +21,7 @@
 
 import org.apache.logging.log4j.LogManager;
 import org.apache.logging.log4j.Logger;
-import org.apache.logging.log4j.message.ParameterizedMessage;
+import org.elasticsearch.ExceptionsHelper;
 import org.elasticsearch.common.SuppressForbidden;
 import org.elasticsearch.common.settings.Settings;
 import org.elasticsearch.common.unit.TimeValue;
@@ -47,8 +47,8 @@ public interface Scheduler {
     /**
      * Create a scheduler that can be used client side. Server side, please use <code>ThreadPool.schedule</code> instead.
      *
-     * Notice that if any scheduled jobs fail with an exception, they will be logged as a warning. This includes jobs started
-     * using execute, submit and schedule.
+     * Notice that if any scheduled jobs fail with an exception, these will bubble up to the uncaught exception handler where they will
+     * be logged as a warning. This includes jobs started using execute, submit and schedule.
      * @param settings the settings to use
      * @return executor
      */
@@ -250,7 +250,8 @@ public void onAfter() {
     }
 
     /**
-     * This subclass ensures to properly bubble up Throwable instances of type Error and logs exceptions thrown in submitted/scheduled tasks
+     * This subclass ensures to properly bubble up Throwable instances of both type Error and Exception thrown in submitted/scheduled
+     * tasks to the uncaught exception handler
      */
     class SafeScheduledThreadPoolExecutor extends ScheduledThreadPoolExecutor {
         private static final Logger logger = LogManager.getLogger(SafeScheduledThreadPoolExecutor.class);
@@ -272,12 +273,10 @@ public SafeScheduledThreadPoolExecutor(int corePoolSize) {
 
         @Override
         protected void afterExecute(Runnable r, Throwable t) {
-            Throwable exception = EsExecutors.rethrowErrors(r);
-            if (exception != null) {
-                logger.warn(() ->
-                    new ParameterizedMessage("uncaught exception in scheduled thread [{}]", Thread.currentThread().getName()),
-                    exception);
-            }
+            if (t != null) return;
+            // Scheduler only allows Runnable's so we expect no checked exceptions here. If anyone uses submit directly on `this`, we
+            // accept the wrapped exception in the output.
+            ExceptionsHelper.reThrowIfNotNull(EsExecutors.rethrowErrors(r));
         }
     }
 }

server/src/main/java/org/elasticsearch/threadpool/ThreadPool.java

@@ -354,8 +354,11 @@ public ScheduledCancellable schedule(Runnable command, TimeValue delay, String e
     }
 
     public void scheduleUnlessShuttingDown(TimeValue delay, String executor, Runnable command) {
+        if (!Names.SAME.equals(executor)) {
+            command = new ThreadedRunnableAllowShutdown(command, executor(executor));
+        }
         try {
-            schedule(command, delay, executor);
+            scheduler.schedule(command, delay.millis(), TimeUnit.MILLISECONDS);
         } catch (EsRejectedExecutionException e) {
             if (e.isExecutorShutdown()) {
                 logger.debug(new ParameterizedMessage("could not schedule execution of [{}] after [{}] on [{}] as executor is shut down",
@@ -520,6 +523,36 @@ public String toString() {
         }
     }
 
+    class ThreadedRunnableAllowShutdown implements Runnable {
+        private final Runnable runnable;
+
+        private final Executor executor;
+
+        ThreadedRunnableAllowShutdown(Runnable runnable, Executor executor) {
+            this.runnable = runnable;
+            this.executor = executor;
+        }
+
+        @Override
+        public void run() {
+            try {
+                executor.execute(runnable);
+            } catch (EsRejectedExecutionException e) {
+                if (e.isExecutorShutdown()) {
+                    logger.debug(new ParameterizedMessage("could not schedule execution of [{}] on [{}] as executor is shut down",
+                        runnable, executor), e);
+                } else {
+                    throw e;
+                }
+            }
+        }
+
+        @Override
+        public String toString() {
+            return "[threaded-allow-shutdown]" + runnable.toString();
+        }
+    }
+
     /**
      * A thread to cache millisecond time values from
      * {@link System#nanoTime()} and {@link System#currentTimeMillis()}.

server/src/test/java/org/elasticsearch/transport/TransportKeepAliveTests.java

@@ -215,5 +215,10 @@ public ScheduledCancellable schedule(Runnable task, TimeValue delay, String exec
             scheduledTasks.add(new Tuple<>(delay, task));
             return null;
         }
+
+        @Override
+        public void scheduleUnlessShuttingDown(TimeValue delay, String executor, Runnable command) {
+            schedule(command, delay, executor);
+        }
     }
 }

test/framework/src/main/java/org/elasticsearch/cluster/coordination/DeterministicTaskQueue.java

@@ -388,6 +388,12 @@ public boolean isCancelled() {
                 };
             }
 
+            @Override
+            public void scheduleUnlessShuttingDown(TimeValue delay, String executor, Runnable command) {
+                // reject on shutdown not implemented for deterministic task queue, so this is equivalent to schedule here.
+                schedule(command, delay, executor);
+            }
+
             @Override
             public Cancellable scheduleWithFixedDelay(Runnable command, TimeValue interval, String executor) {
                 return super.scheduleWithFixedDelay(command, interval, executor);