Call internal executor method to ensure at least 1 worker

Author: ldematte

build-tools-internal/src/main/java/org/elasticsearch/gradle/internal/ElasticsearchTestBasePlugin.java

@@ -112,6 +112,7 @@ public void execute(Task t) {
                 "-Xms" + System.getProperty("tests.heap.size", "512m"),
                 "-Dtests.testfeatures.enabled=true",
                 "--add-opens=java.base/java.util=ALL-UNNAMED",
+                "--add-opens=java.base/java.util.concurrent=ALL-UNNAMED",
                 // TODO: only open these for mockito when it is modularized
                 "--add-opens=java.base/java.security.cert=ALL-UNNAMED",
                 "--add-opens=java.base/java.nio.channels=ALL-UNNAMED",

distribution/tools/server-cli/src/main/java/org/elasticsearch/server/cli/SystemJvmOptions.java

@@ -68,7 +68,8 @@ static List<String> systemJvmOptions(Settings nodeSettings, final Map<String, St
                 "-Dlog4j2.formatMsgNoLookups=true",
                 "-Djava.locale.providers=CLDR",
                 // Pass through distribution type
-                "-Des.distribution.type=" + distroType
+                "-Des.distribution.type=" + distroType,
+                "--add-opens=java.base/java.util.concurrent=org.elasticsearch.server"
             ),
             maybeEnableNativeAccess(useEntitlements),
             maybeOverrideDockerCgroup(distroType),

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

@@ -17,8 +17,11 @@
 import org.elasticsearch.core.SuppressForbidden;
 import org.elasticsearch.node.Node;
 
+import java.lang.invoke.MethodHandle;
+import java.lang.invoke.MethodHandles;
 import java.security.AccessController;
 import java.security.PrivilegedAction;
+import java.security.PrivilegedExceptionAction;
 import java.util.List;
 import java.util.Optional;
 import java.util.concurrent.AbstractExecutorService;
@@ -109,34 +112,12 @@ public static EsThreadPoolExecutor newScaling(
     ) {
         ExecutorScalingQueue<Runnable> queue = new ExecutorScalingQueue<>();
         EsThreadPoolExecutor executor;
-
-        final int corePoolSize;
-        final long adjustedKeepAliveTime;
-        final boolean allowCoreThreadTimeOut;
-        if (min == 0) {
-            // If a min of 0 is requested we set the thread core count to 1, but we allow the thread cores to timeout.
-            // This way we still achieve the goal of having 0 threads when idle, but we do not run into the risk of running into a case
-            // where we have task(s) in the queue and no worker to consume them.
-            corePoolSize = 1;
-            allowCoreThreadTimeOut = true;
-            if (keepAliveTime == 0) {
-                // The JDK does not allow a timeout of 0 with allowCoreThreadTimeOut
-                adjustedKeepAliveTime = 1;
-            } else {
-                adjustedKeepAliveTime = keepAliveTime;
-            }
-        } else {
-            corePoolSize = min;
-            adjustedKeepAliveTime = keepAliveTime;
-            allowCoreThreadTimeOut = false;
-        }
-
         if (config.trackExecutionTime()) {
             executor = new TaskExecutionTimeTrackingEsThreadPoolExecutor(
                 name,
-                corePoolSize,
+                min,
                 max,
-                adjustedKeepAliveTime,
+                keepAliveTime,
                 unit,
                 queue,
                 TimedRunnable::new,
@@ -148,17 +129,16 @@ public static EsThreadPoolExecutor newScaling(
         } else {
             executor = new EsThreadPoolExecutor(
                 name,
-                corePoolSize,
+                min,
                 max,
-                adjustedKeepAliveTime,
+                keepAliveTime,
                 unit,
                 queue,
                 threadFactory,
                 new ForceQueuePolicy(rejectAfterShutdown),
                 contextHolder
             );
         }
-        executor.allowCoreThreadTimeOut(allowCoreThreadTimeOut);
         queue.executor = executor;
         return executor;
     }
@@ -479,6 +459,22 @@ static class ForceQueuePolicy extends EsRejectedExecutionHandler {
          */
         private final boolean rejectAfterShutdown;
 
+        private static final MethodHandle ensurePrestart$mh = lookupEnsurePrestart();
+
+        @SuppressForbidden(reason = "reflective access")
+        static MethodHandle lookupEnsurePrestart() {
+            try {
+                return AccessController.doPrivileged((PrivilegedExceptionAction<MethodHandle>) () -> {
+                    var ensurePrestartMethod = ThreadPoolExecutor.class.getDeclaredMethod("ensurePrestart");
+                    ensurePrestartMethod.setAccessible(true);
+
+                    return MethodHandles.lookup().unreflect(ensurePrestartMethod);
+                });
+            } catch (Exception e) {
+                throw new RuntimeException(e);
+            }
+        }
+
         /**
          * @param rejectAfterShutdown indicates if {@link Runnable} should be rejected once the thread pool is shutting down
          */
@@ -506,6 +502,11 @@ public void rejectedExecution(Runnable task, ThreadPoolExecutor executor) {
                         // b) var t = executor.getQueue().poll(); if (t != null) executor.execute(t); (re-enqueue) but this will likely
                         // have undesired side effects (it's a recursive call in disguise)
                         // c) executor.ensurePrestart(); which is not public nor protected, so we would have to call it reflectively
+                        try {
+                            ensurePrestart$mh.invokeExact(executor);
+                        } catch (Throwable e) {
+                            throw new RuntimeException(e);
+                        }
                     }
                 }
             } else {

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

@@ -66,28 +66,6 @@ public class EsThreadPoolExecutor extends ThreadPoolExecutor {
         this.contextHolder = contextHolder;
     }
 
-    @Override
-    public void setCorePoolSize(int corePoolSize) {
-        if (corePoolSize == 0) {
-            corePoolSize = 1;
-            // Enforce a minimum timeout > 0 so we can allow core threads timeout
-            if (getKeepAliveTime(TimeUnit.MILLISECONDS) == 0) {
-                setKeepAliveTime(1, TimeUnit.MILLISECONDS);
-            }
-            super.allowCoreThreadTimeOut(true);
-        }
-        super.setCorePoolSize(corePoolSize);
-    }
-
-    @Override
-    public void allowCoreThreadTimeOut(boolean value) {
-        if (value == false && getCorePoolSize() == 0) {
-            // Nope
-            return;
-        }
-        super.allowsCoreThreadTimeOut();
-    }
-
     @Override
     public void execute(Runnable command) {
         final Runnable wrappedRunnable = wrapRunnable(command);

server/src/main/resources/org/elasticsearch/bootstrap/security.policy

@@ -32,6 +32,10 @@ grant codeBase "${codebase.elasticsearch}" {
 
   // for plugin api dynamic settings instances
   permission java.lang.RuntimePermission "accessClassInPackage.jdk.internal.reflect";
+
+  // for ExExecutors internal function patch
+  permission java.lang.RuntimePermission "accessDeclaredMembers";
+  permission java.lang.reflect.ReflectPermission "suppressAccessChecks";
 };
 
 //// Very special jar permissions: