Handle throws on tasks submitted to thread pools

When we submit a task to a thread pool for asynchronous execution, we
are returned a future. Since we submitted to go asynchronous, these
futures are not inspected for failure (we would have to block a thread
to do that). While we have on failure handlers for exceptions that are
thrown during execution, we do not handle throwables that are not
exceptions and these end up silently lost. This commit adds a check
after the runnable returns that inspects the status of the future. If an
unhandled throwable occurred during execution, this throwable is
propogated out where it will land in the uncaught exception handler.

Relates #28667

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

@@ -0,0 +1,105 @@
+/*
+ * Licensed to Elasticsearch under one or more contributor
+ * license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright
+ * ownership. Elasticsearch licenses this file to you under
+ * the Apache License, Version 2.0 (the "License"); you may
+ * not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *    http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.elasticsearch.threadpool;
+
+import org.elasticsearch.test.ESTestCase;
+import org.junit.After;
+import org.junit.Before;
+
+import java.util.Optional;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.atomic.AtomicReference;
+import java.util.function.Consumer;
+import java.util.function.Supplier;
+
+import static org.hamcrest.Matchers.containsString;
+import static org.hamcrest.Matchers.hasToString;
+import static org.hamcrest.Matchers.instanceOf;
+
+public class EvilThreadPoolTests extends ESTestCase {
+
+    private ThreadPool threadPool;
+
+    @Before
+    public void setUpThreadPool() {
+        threadPool = new TestThreadPool(EvilThreadPoolTests.class.getName());
+    }
+
+    @After
+    public void tearDownThreadPool() throws InterruptedException {
+        terminate(threadPool);
+    }
+
+    public void testExecutionException() throws InterruptedException {
+        runExecutionExceptionTest(
+                () -> {
+                    throw new Error("future error");
+                },
+                true,
+                o -> {
+                    assertTrue(o.isPresent());
+                    assertThat(o.get(), instanceOf(Error.class));
+                    assertThat(o.get(), hasToString(containsString("future error")));
+                });
+        runExecutionExceptionTest(
+                () -> {
+                    throw new IllegalStateException("future exception");
+                },
+                false,
+                o -> assertFalse(o.isPresent()));
+    }
+
+    private void runExecutionExceptionTest(
+            final Runnable runnable,
+            final boolean expectThrowable,
+            final Consumer<Optional<Throwable>> consumer) throws InterruptedException {
+        final AtomicReference<Throwable> throwableReference = new AtomicReference<>();
+        final Thread.UncaughtExceptionHandler uncaughtExceptionHandler = Thread.getDefaultUncaughtExceptionHandler();
+        final CountDownLatch uncaughtExceptionHandlerLatch = new CountDownLatch(1);
+
+        try {
+            Thread.setDefaultUncaughtExceptionHandler((t, e) -> {
+                assertTrue(expectThrowable);
+                throwableReference.set(e);
+                uncaughtExceptionHandlerLatch.countDown();
+            });
+
+            final CountDownLatch supplierLatch = new CountDownLatch(1);
+
+            threadPool.generic().submit(() -> {
+                try {
+                    runnable.run();
+                } finally {
+                    supplierLatch.countDown();
+                }
+            });
+
+            supplierLatch.await();
+
+            if (expectThrowable) {
+                uncaughtExceptionHandlerLatch.await();
+            }
+            consumer.accept(Optional.ofNullable(throwableReference.get()));
+        } finally {
+            Thread.setDefaultUncaughtExceptionHandler(uncaughtExceptionHandler);
+        }
+    }
+
+}

server/src/main/java/org/elasticsearch/common/util/concurrent/ThreadContext.java

@@ -19,9 +19,11 @@
 package org.elasticsearch.common.util.concurrent;
 
 import org.apache.lucene.util.CloseableThreadLocal;
+import org.elasticsearch.ExceptionsHelper;
 import org.elasticsearch.common.io.stream.StreamInput;
 import org.elasticsearch.common.io.stream.StreamOutput;
 import org.elasticsearch.common.io.stream.Writeable;
+import org.elasticsearch.common.logging.ESLoggerFactory;
 import org.elasticsearch.common.settings.Setting;
 import org.elasticsearch.common.settings.Setting.Property;
 import org.elasticsearch.common.settings.Settings;
@@ -33,7 +35,12 @@
 import java.util.HashMap;
 import java.util.List;
 import java.util.Map;
+import java.util.Optional;
 import java.util.Set;
+import java.util.concurrent.CancellationException;
+import java.util.concurrent.ExecutionException;
+import java.util.concurrent.FutureTask;
+import java.util.concurrent.RunnableFuture;
 import java.util.concurrent.atomic.AtomicBoolean;
 import java.util.function.Function;
 import java.util.function.Supplier;
@@ -564,6 +571,36 @@ public void run() {
                 ctx.restore();
                 whileRunning = true;
                 in.run();
+                if (in instanceof RunnableFuture) {
+                    /*
+                     * The wrapped runnable arose from asynchronous submission of a task to an executor. If an uncaught exception was thrown
+                     * during the execution of this task, we need to inspect this runnable and see if it is an error that should be
+                     * propagated to the uncaught exception handler.
+                     */
+                    try {
+                        ((RunnableFuture) in).get();
+                    } catch (final Exception e) {
+                        /*
+                         * In theory, Future#get can only throw a cancellation exception, an interrupted exception, or an execution
+                         * exception. We want to ignore cancellation exceptions, restore the interrupt status on interrupted exceptions, and
+                         * inspect the cause of an execution. We are going to be extra paranoid here though and completely unwrap the
+                         * exception to ensure that there is not a buried error anywhere. We assume that a general exception has been
+                         * handled by the executed task or the task submitter.
+                         */
+                        assert e instanceof CancellationException
+                                || e instanceof InterruptedException
+                                || e instanceof ExecutionException : e;
+                        final Optional<Error> maybeError = ExceptionsHelper.maybeError(e, ESLoggerFactory.getLogger(ThreadContext.class));
+                        if (maybeError.isPresent()) {
+                            // throw this error where it will propagate to the uncaught exception handler
+                            throw maybeError.get();
+                        }
+                        if (e instanceof InterruptedException) {
+                            // restore the interrupt status
+                            Thread.currentThread().interrupt();
+                        }
+                    }
+                }
                 whileRunning = false;
             } catch (IllegalStateException ex) {
                 if (whileRunning || threadLocal.closed.get() == false) {