Use the QuarkusForkJoinWorkerThreadFactory for fork-join pools

.github/quarkus-bot.yml

@@ -328,6 +328,7 @@ triage:
         - core/deployment/src/main/java/io/quarkus/deployment/configuration/
         - core/runtime/src/main/java/io/quarkus/runtime/configuration/
     - labels: [area/core]
+      notify: [aloubyansky, gsmet, geoand, radcortez, Sanne, stuartwdouglas]
       directories:
         - core/
     - labels: [area/dependencies]

core/deployment/src/main/java/io/quarkus/deployment/ForkJoinPoolProcessor.java

@@ -0,0 +1,13 @@
+package io.quarkus.deployment;
+
+import io.quarkus.deployment.annotations.BuildStep;
+import io.quarkus.deployment.builditem.SystemPropertyBuildItem;
+
+public final class ForkJoinPoolProcessor {
+
+    @BuildStep
+    SystemPropertyBuildItem setProperty() {
+        return new SystemPropertyBuildItem("java.util.concurrent.ForkJoinPool.common.threadFactory",
+                "io.quarkus.bootstrap.forkjoin.QuarkusForkJoinWorkerThreadFactory");
+    }
+}

core/deployment/src/main/java/io/quarkus/deployment/jbang/JBangAugmentorImpl.java

@@ -124,6 +124,8 @@ public void accept(BuildChainBuilder builder) {
             resultMap.put("files", result);
             final List<String> javaargs = new ArrayList<>();
             javaargs.add("-Djava.util.logging.manager=org.jboss.logmanager.LogManager");
+            javaargs.add(
+                    "-Djava.util.concurrent.ForkJoinPool.common.threadFactory=io.quarkus.bootstrap.forkjoin.QuarkusForkJoinWorkerThreadFactory");
             resultMap.put("java-args", javaargs);
             resultMap.put("main-class", buildResult.consume(MainClassBuildItem.class).getClassName());
             if (nativeRequested) {

core/runtime/src/main/java/io/quarkus/runtime/Quarkus.java

@@ -5,7 +5,6 @@
 import java.util.function.BiConsumer;
 
 import org.jboss.logging.Logger;
-import org.jboss.logmanager.LogManager;
 
 import io.quarkus.launcher.QuarkusLauncher;
 
@@ -57,7 +56,9 @@ public static void run(Class<? extends QuarkusApplication> quarkusApplication, S
     public static void run(Class<? extends QuarkusApplication> quarkusApplication, BiConsumer<Integer, Throwable> exitHandler,
             String... args) {
         try {
-            System.setProperty("java.util.logging.manager", LogManager.class.getName());
+            System.setProperty("java.util.logging.manager", "org.jboss.logmanager.LogManager");
+            System.setProperty("java.util.concurrent.ForkJoinPool.common.threadFactory",
+                    "io.quarkus.bootstrap.forkjoin.QuarkusForkJoinWorkerThreadFactory");
             //production and common dev mode path
             //we already have an application, run it directly
             Class<? extends Application> appClass = (Class<? extends Application>) Class.forName(Application.APP_CLASS_NAME,

core/test-extension/deployment/src/test/java/io/quarkus/forkjoin/ForkJoinPoolAssertions.java

@@ -0,0 +1,66 @@
+package io.quarkus.forkjoin;
+
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.ForkJoinPool;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicInteger;
+
+/**
+ * Helper to collect assertions related to the JDK ForkJoinPool
+ * state.
+ */
+public class ForkJoinPoolAssertions {
+
+    /**
+     * We test which Classloader is being used by each thread
+     * in the common pool of ForkJoinPool.
+     * It is expected that a Quarkus application running in production mode
+     * will have set the io.quarkus.bootstrap.runner.RunnerClassLoader as
+     * context classloader on each of them, to prevent problems at runtime.
+     * 
+     * @return true only if all checks are successful.
+     */
+    static boolean isEachFJThreadUsingQuarkusClassloader() {
+        AtomicInteger testedOk = new AtomicInteger();
+        final int poolParallelism = ForkJoinPool.getCommonPoolParallelism();
+        CountDownLatch allDone = new CountDownLatch(poolParallelism);
+        CountDownLatch taskRelease = new CountDownLatch(1);
+        if (poolParallelism < 1) {
+            System.out
+                    .println("Can't test this when ForkJoinPool.getCommonPoolParallelism() has been forced to less than one.");
+            return false;
+        }
+        for (int i = 0; i < poolParallelism; ++i) {
+            ForkJoinPool.commonPool().execute(new Runnable() {
+                @Override
+                public void run() {
+                    final ClassLoader contextClassLoader = Thread.currentThread().getContextClassLoader();
+                    final String classLoaderImplementationName = contextClassLoader.getClass().getName();
+                    if (classLoaderImplementationName.equals(io.quarkus.bootstrap.runner.RunnerClassLoader.class.getName())) {
+                        testedOk.incrementAndGet();
+                    } else {
+                        System.out.println("Unexpected classloader name used in production: " + classLoaderImplementationName);
+                    }
+                    allDone.countDown();
+                    try {
+                        taskRelease.await();
+                    } catch (InterruptedException e) {
+                        e.printStackTrace();
+                    }
+                }
+            });
+        }
+        try {
+            if (!allDone.await(10, TimeUnit.SECONDS)) {
+                System.out.println("Timed out while trying to scale up the fork join pool");
+                return false;
+            }
+        } catch (InterruptedException e) {
+            e.printStackTrace();
+            return false;
+        } finally {
+            taskRelease.countDown();
+        }
+        return testedOk.get() == poolParallelism;
+    }
+}

core/test-extension/deployment/src/test/java/io/quarkus/forkjoin/ForkJoinProdModeTestCase.java

@@ -0,0 +1,31 @@
+package io.quarkus.forkjoin;
+
+import org.jboss.shrinkwrap.api.ShrinkWrap;
+import org.jboss.shrinkwrap.api.asset.StringAsset;
+import org.jboss.shrinkwrap.api.spec.JavaArchive;
+import org.junit.jupiter.api.Assertions;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.extension.RegisterExtension;
+
+import io.quarkus.test.QuarkusProdModeTest;
+
+public class ForkJoinProdModeTestCase {
+
+    @RegisterExtension
+    static final QuarkusProdModeTest config = new QuarkusProdModeTest()
+            .setArchiveProducer(() -> ShrinkWrap.create(JavaArchive.class)
+                    .addClasses(ProductionModeTestsEndpoint.class, ForkJoinPoolAssertions.class)
+                    .addAsResource(new StringAsset("quarkus.banner.enabled=false\nquarkus.log.level=ERROR"),
+                            "application.properties"))
+            .setApplicationName("prod-mode-quarkus")
+            .setApplicationVersion("0.1-SNAPSHOT")
+            .setRun(true)
+            .setExpectExit(true);
+
+    @Test
+    public void testForkJoinClassLoading() {
+        Assertions.assertEquals(0, config.getExitCode(), config.getStartupConsoleOutput());
+        Assertions.assertFalse(config.getStartupConsoleOutput().contains("FAIL"), config.getStartupConsoleOutput());
+        Assertions.assertTrue(config.getStartupConsoleOutput().contains("PASS"), config.getStartupConsoleOutput());
+    }
+}

core/test-extension/deployment/src/test/java/io/quarkus/forkjoin/ProductionModeTestsEndpoint.java

@@ -0,0 +1,32 @@
+package io.quarkus.forkjoin;
+
+import io.quarkus.runtime.QuarkusApplication;
+import io.quarkus.runtime.annotations.QuarkusMain;
+
+@QuarkusMain
+public class ProductionModeTestsEndpoint implements QuarkusApplication {
+
+    @Override
+    public int run(String... args) throws Exception {
+
+        //check system properties set
+        if (!"org.jboss.logmanager.LogManager".equals(System.getProperty("java.util.logging.manager"))) {
+            System.out.println("java.util.logging.manager not set");
+            System.out.println("FAIL");
+            return 1;
+        }
+        if (!"io.quarkus.bootstrap.forkjoin.QuarkusForkJoinWorkerThreadFactory"
+                .equals(System.getProperty("java.util.concurrent.ForkJoinPool.common.threadFactory"))) {
+            System.out.println("java.util.concurrent.ForkJoinPool.common.threadFactory not set");
+            System.out.println("FAIL");
+            return 1;
+        }
+        if (!ForkJoinPoolAssertions.isEachFJThreadUsingQuarkusClassloader()) {
+            System.out.println("fork join assertions failed");
+            System.out.println("FAIL");
+            return 1;
+        }
+        System.out.println("PASS");
+        return 0;
+    }
+}

independent-projects/bootstrap/runner/src/main/java/io/quarkus/bootstrap/forkjoin/QuarkusForkJoinWorkerThread.java

@@ -0,0 +1,30 @@
+package io.quarkus.bootstrap.forkjoin;
+
+import io.quarkus.bootstrap.runner.RunnerClassLoader;
+import java.util.concurrent.ForkJoinPool;
+import java.util.concurrent.ForkJoinWorkerThread;
+
+public class QuarkusForkJoinWorkerThread extends ForkJoinWorkerThread {
+
+    private static volatile ClassLoader qClassloader;
+
+    protected QuarkusForkJoinWorkerThread(ForkJoinPool pool) {
+        super(pool);
+    }
+
+    public static synchronized void setQuarkusAppClassloader(RunnerClassLoader runnerClassLoader) {
+        qClassloader = runnerClassLoader;
+    }
+
+    protected void onStart() {
+        super.onStart();
+        if (qClassloader != null) {
+            super.setContextClassLoader(qClassloader);
+        }
+        //When null: it means the task which is being scheduled was scheduled
+        //before the application was started - possibly by some parallel
+        //task during the preparation of the application classloader itself:
+        //should be safe to run with the existing (system) classsloader.
+    }
+
+}

independent-projects/bootstrap/runner/src/main/java/io/quarkus/bootstrap/forkjoin/QuarkusForkJoinWorkerThreadFactory.java

@@ -0,0 +1,23 @@
+package io.quarkus.bootstrap.forkjoin;
+
+import java.util.concurrent.ForkJoinPool;
+import java.util.concurrent.ForkJoinWorkerThread;
+
+/**
+ * This implementation of JDK's ForkJoinPool.ForkJoinWorkerThreadFactory
+ * needs to be enabled by setting system property {@code java.util.concurrent.ForkJoinPool.common.threadFactory}
+ * to this class name, so to allow Quarkus to set the contextual classloader to the correct
+ * runtime classloader for each thread being started by the common pool.
+ * Otherwise the system classloader will be set, which is unable to load all application resources.
+ */
+public final class QuarkusForkJoinWorkerThreadFactory implements ForkJoinPool.ForkJoinWorkerThreadFactory {
+
+    public QuarkusForkJoinWorkerThreadFactory() {
+    }
+
+    @Override
+    public ForkJoinWorkerThread newThread(ForkJoinPool pool) {
+        return new QuarkusForkJoinWorkerThread(pool);
+    }
+
+}

independent-projects/bootstrap/runner/src/main/java/io/quarkus/bootstrap/runner/QuarkusEntryPoint.java

@@ -1,5 +1,6 @@
 package io.quarkus.bootstrap.runner;
 
+import io.quarkus.bootstrap.forkjoin.QuarkusForkJoinWorkerThread;
 import java.io.BufferedInputStream;
 import java.io.File;
 import java.io.IOException;
@@ -21,6 +22,8 @@ public class QuarkusEntryPoint {
 
     public static void main(String... args) throws Throwable {
         System.setProperty("java.util.logging.manager", org.jboss.logmanager.LogManager.class.getName());
+        System.setProperty("java.util.concurrent.ForkJoinPool.common.threadFactory",
+                "io.quarkus.bootstrap.forkjoin.QuarkusForkJoinWorkerThreadFactory");
         Timing.staticInitStarted(false);
         doRun(args);
     }
@@ -42,12 +45,15 @@ private static void doRun(Object args) throws IOException, ClassNotFoundExceptio
                     24_576)) {
                 app = SerializedApplication.read(in, appRoot);
             }
+            final RunnerClassLoader appRunnerClassLoader = app.getRunnerClassLoader();
             try {
-                Thread.currentThread().setContextClassLoader(app.getRunnerClassLoader());
-                Class<?> mainClass = app.getRunnerClassLoader().loadClass(app.getMainClass());
+                Thread.currentThread().setContextClassLoader(appRunnerClassLoader);
+                QuarkusForkJoinWorkerThread.setQuarkusAppClassloader(appRunnerClassLoader);
+                Class<?> mainClass = appRunnerClassLoader.loadClass(app.getMainClass());
                 mainClass.getMethod("main", String[].class).invoke(null, args);
             } finally {
-                app.getRunnerClassLoader().close();
+                QuarkusForkJoinWorkerThread.setQuarkusAppClassloader(null);
+                appRunnerClassLoader.close();
             }
         }
     }