Propagate Errors in executors to uncaught exception handler

[ywelsch]

This PR is a continuation of #28667 and has as goal to convert all executors to propagate errors to the uncaught exception handler. Notable missing ones were the direct executor and the scheduler. This PR also makes it the property of the executor, not the runnable, to ensure this property. A big part of this PR consists of vastly improving the test coverage in this area, which also uncovers other woes marked as TODO:

• executors of type fixed_auto_queue_size (i.e. search / search_throttled) wraps runnables into TimedRunnable, which is an AbstractRunnable. This is dangerous as it will silently swallow exceptions, and possibly miss calling a response listener.
• scheduler swallows exceptions. In many cases, the scheduler switches to an other executor upon execution, and so the issue mainly exists for the combination of scheduler + SAME executor.

/cc: @DaveCTurner

[elasticmachine]

Pinging @elastic/es-core-infra

[nik9000]

o makes it the property of the executor, not the runnable, to ensure this property

Oooooooooooooh! Nice.

[ywelsch]

ping @jasontedor

[jasontedor]

This is a great change. Thank you for taking it on. A few comments:

• I am concerned about a new executor being added in the future (like AutodetectWorkerExecutorService that would not do the right thing); can we detect this and enforce it? Similar to what we have done with LoggerUsageCheck and FeatureAwareCheck?
• The wrapping and unwrapping feels a little brittle with all the instance of checking; how about something like this?

diff --git a/server/src/main/java/org/elasticsearch/common/util/concurrent/PrioritizedEsThreadPoolExecutor.java b/server/src/main/java/org/elasticsearch/common/util/concurrent/PrioritizedEsThreadPoolExecutor.java
index 644b6d067fc..729ad4c52de 100644
--- a/server/src/main/java/org/elasticsearch/common/util/concurrent/PrioritizedEsThreadPoolExecutor.java
+++ b/server/src/main/java/org/elasticsearch/common/util/concurrent/PrioritizedEsThreadPoolExecutor.java
@@ -151,8 +151,8 @@ public class PrioritizedEsThreadPoolExecutor extends EsThreadPoolExecutor {
 
     @Override
     protected Runnable unwrap(Runnable runnable) {
-        if (runnable instanceof TieBreakingPrioritizedRunnable) {
-            return super.unwrap(((TieBreakingPrioritizedRunnable) runnable).unwrap());
+        if (runnable instanceof WrappedRunnable) {
+            return ((WrappedRunnable) runnable).unwrap();
         } else {
             return super.unwrap(runnable);
         }
@@ -256,9 +256,10 @@ public class PrioritizedEsThreadPoolExecutor extends EsThreadPoolExecutor {
             }
         }
 
-        Runnable unwrap() {
+        public Runnable unwrap() {
             return runnable;
         }
+
     }
 
     private static final class PrioritizedFutureTask<T> extends FutureTask<T> implements Comparable<PrioritizedFutureTask> {
diff --git a/server/src/main/java/org/elasticsearch/common/util/concurrent/PrioritizedRunnable.java b/server/src/main/java/org/elasticsearch/common/util/concurrent/PrioritizedRunnable.java
index 7ef2e96e2c5..6773e2884f8 100644
--- a/server/src/main/java/org/elasticsearch/common/util/concurrent/PrioritizedRunnable.java
+++ b/server/src/main/java/org/elasticsearch/common/util/concurrent/PrioritizedRunnable.java
@@ -23,7 +23,7 @@ import org.elasticsearch.common.Priority;
 import java.util.concurrent.TimeUnit;
 import java.util.function.LongSupplier;
 
-public abstract class PrioritizedRunnable implements Runnable, Comparable<PrioritizedRunnable> {
+public abstract class PrioritizedRunnable implements WrappedRunnable, Comparable<PrioritizedRunnable> {
 
     private final Priority priority;
     private final long creationDate;
@@ -82,5 +82,11 @@ public abstract class PrioritizedRunnable implements Runnable, Comparable<Priori
         public void run() {
             runnable.run();
         }
+
+        @Override
+        public Runnable unwrap() {
+            return runnable;
+        }
+
     }
 }
diff --git a/server/src/main/java/org/elasticsearch/common/util/concurrent/QueueResizingEsThreadPoolExecutor.java b/server/src/main/java/org/elasticsearch/common/util/concurrent/QueueResizingEsThreadPoolExecutor.java
index 7fbc9ed91e8..c1d2f1f43f8 100644
--- a/server/src/main/java/org/elasticsearch/common/util/concurrent/QueueResizingEsThreadPoolExecutor.java
+++ b/server/src/main/java/org/elasticsearch/common/util/concurrent/QueueResizingEsThreadPoolExecutor.java
@@ -85,8 +85,8 @@ public final class QueueResizingEsThreadPoolExecutor extends EsThreadPoolExecuto
     @Override
     protected Runnable unwrap(Runnable runnable) {
         final Runnable unwrapped = super.unwrap(runnable);
-        if (unwrapped instanceof TimedRunnable) {
-            return ((TimedRunnable) unwrapped).unwrap();
+        if (unwrapped instanceof WrappedRunnable) {
+            return ((WrappedRunnable) unwrapped).unwrap();
         } else {
             return unwrapped;
         }
diff --git a/server/src/main/java/org/elasticsearch/common/util/concurrent/ThreadContext.java b/server/src/main/java/org/elasticsearch/common/util/concurrent/ThreadContext.java
index 83ee7f63cb8..b95dd3c01af 100644
--- a/server/src/main/java/org/elasticsearch/common/util/concurrent/ThreadContext.java
+++ b/server/src/main/java/org/elasticsearch/common/util/concurrent/ThreadContext.java
@@ -347,11 +347,8 @@ public final class ThreadContext implements Closeable, Writeable {
      * Unwraps a command that was previously wrapped by {@link #preserveContext(Runnable)}.
      */
     public Runnable unwrap(Runnable command) {
-        if (command instanceof ContextPreservingAbstractRunnable) {
-            return ((ContextPreservingAbstractRunnable) command).unwrap();
-        }
-        if (command instanceof ContextPreservingRunnable) {
-            return ((ContextPreservingRunnable) command).unwrap();
+        if (command instanceof WrappedRunnable) {
+            return ((WrappedRunnable) command).unwrap();
         }
         return command;
     }
@@ -677,7 +674,7 @@ public final class ThreadContext implements Closeable, Writeable {
     /**
      * Wraps an AbstractRunnable to preserve the thread context.
      */
-    private class ContextPreservingAbstractRunnable extends AbstractRunnable {
+    private class ContextPreservingAbstractRunnable extends AbstractRunnable implements WrappedRunnable {
         private final AbstractRunnable in;
         private final ThreadContext.StoredContext creatorsContext;
 
diff --git a/server/src/main/java/org/elasticsearch/common/util/concurrent/TimedRunnable.java b/server/src/main/java/org/elasticsearch/common/util/concurrent/TimedRunnable.java
index ea264aa1608..3f3589a7226 100644
--- a/server/src/main/java/org/elasticsearch/common/util/concurrent/TimedRunnable.java
+++ b/server/src/main/java/org/elasticsearch/common/util/concurrent/TimedRunnable.java
@@ -23,7 +23,7 @@ package org.elasticsearch.common.util.concurrent;
  * A class used to wrap a {@code Runnable} that allows capturing the time of the task since creation
  * through execution as well as only execution time.
  */
-class TimedRunnable extends AbstractRunnable {
+class TimedRunnable extends AbstractRunnable implements WrappedRunnable {
     private final Runnable original;
     private final long creationTimeNanos;
     private long startTimeNanos;
@@ -94,7 +94,8 @@ class TimedRunnable extends AbstractRunnable {
         return Math.max(finishTimeNanos - startTimeNanos, 1);
     }
 
-    Runnable unwrap() {
+    public Runnable unwrap() {
         return original;
     }
+
 }
diff --git a/server/src/main/java/org/elasticsearch/common/util/concurrent/WrappedRunnable.java b/server/src/main/java/org/elasticsearch/common/util/concurrent/WrappedRunnable.java
index 2bd031ee2b4..bdb4b6b36b6 100644
--- a/server/src/main/java/org/elasticsearch/common/util/concurrent/WrappedRunnable.java
+++ b/server/src/main/java/org/elasticsearch/common/util/concurrent/WrappedRunnable.java
@@ -1,4 +1,7 @@
 package org.elasticsearch.common.util.concurrent;
 
-public class WrappedRunnable {
+public interface WrappedRunnable extends Runnable {
+
+    Runnable unwrap();
+
 }

[ywelsch]

@elasticmachine run the Gradle build tests 1

[ywelsch]

pinging my reviewers here

[original-brownbear]

LGTM :)

[jasontedor]

LGTM. Thank you for the iteration to prevent new executors from having the same problem.

[ywelsch]

@elasticmachine please run the Gradle build tests 1

[ywelsch]

@elasticmachine please run the Gradle build tests 1

[ywelsch]

@elasticmachine please run the Gradle build tests 1

[ywelsch]

@elasticmachine please run the Gradle build tests 1