Executes scheduled methods on duplicated contexts

- Remove the executor from the ScheduledContext as it was confusing and can only be used for blocking methods
- Create a duplicated context for both blocking and non-blocking invocation
- Blocking invocation use executeBlocking, which uses the same thread pool as the previous executor
- It is not possible to create the duplicated context in the thread factory (an idea discussed with Bruno and Martin because it creates a chicken-and-egg problem)
- Simplify the dev console @scheduled support to use the same strategy - no need to play with the classloader anymore as there is no more in-thread direct invocation and the Vert.x thread that will be used has the right TCCL configured

extensions/scheduler/common/src/main/java/io/quarkus/scheduler/common/runtime/SchedulerContext.java

@@ -2,14 +2,11 @@
 
 import java.lang.reflect.InvocationTargetException;
 import java.util.List;
-import java.util.concurrent.ExecutorService;
 
 import com.cronutils.model.CronType;
 
 public interface SchedulerContext {
 
-    ExecutorService getExecutor();
-
     CronType getCronType();
 
     List<ScheduledMethodMetadata> getScheduledMethods();

extensions/scheduler/deployment/src/main/java/io/quarkus/scheduler/deployment/SchedulerProcessor.java

@@ -61,7 +61,6 @@
 import io.quarkus.deployment.annotations.BuildStep;
 import io.quarkus.deployment.annotations.Record;
 import io.quarkus.deployment.builditem.AnnotationProxyBuildItem;
-import io.quarkus.deployment.builditem.ExecutorBuildItem;
 import io.quarkus.deployment.builditem.FeatureBuildItem;
 import io.quarkus.deployment.builditem.GeneratedClassBuildItem;
 import io.quarkus.deployment.builditem.nativeimage.ReflectiveClassBuildItem;
@@ -277,7 +276,7 @@ public List<UnremovableBeanBuildItem> unremovableBeans() {
     public FeatureBuildItem build(SchedulerConfig config, BuildProducer<SyntheticBeanBuildItem> syntheticBeans,
             SchedulerRecorder recorder, List<ScheduledBusinessMethodItem> scheduledMethods,
             BuildProducer<GeneratedClassBuildItem> generatedClasses, BuildProducer<ReflectiveClassBuildItem> reflectiveClass,
-            AnnotationProxyBuildItem annotationProxy, ExecutorBuildItem executor) {
+            AnnotationProxyBuildItem annotationProxy) {
 
         List<ScheduledMethodMetadata> scheduledMetadata = new ArrayList<>();
         ClassOutput classOutput = new GeneratedClassGizmoAdaptor(generatedClasses, new Function<String, String>() {
@@ -311,7 +310,7 @@ public String apply(String name) {
         }
 
         syntheticBeans.produce(SyntheticBeanBuildItem.configure(SchedulerContext.class).setRuntimeInit()
-                .supplier(recorder.createContext(config, executor.getExecutorProxy(), scheduledMetadata))
+                .supplier(recorder.createContext(config, scheduledMetadata))
                 .done());
 
         return new FeatureBuildItem(Feature.SCHEDULER);

extensions/scheduler/deployment/src/test/java/io/quarkus/scheduler/test/DuplicatedContextTest.java

@@ -0,0 +1,81 @@
+package io.quarkus.scheduler.test;
+
+import static org.awaitility.Awaitility.await;
+
+import java.time.Duration;
+import java.util.concurrent.atomic.AtomicInteger;
+
+import javax.inject.Inject;
+
+import org.junit.jupiter.api.Assertions;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.extension.RegisterExtension;
+
+import io.quarkus.scheduler.Scheduled;
+import io.quarkus.test.QuarkusUnitTest;
+import io.smallrye.common.vertx.VertxContext;
+import io.smallrye.mutiny.Uni;
+import io.vertx.core.Context;
+import io.vertx.core.Vertx;
+
+/**
+ * Verifies that the @Scheduled method are called on a duplicated context.
+ */
+public class DuplicatedContextTest {
+    @RegisterExtension
+    static final QuarkusUnitTest test = new QuarkusUnitTest()
+            .withApplicationRoot((jar) -> jar
+                    .addClasses(MyScheduledClass.class));
+
+    @Inject
+    MyScheduledClass scheduled;
+
+    @Test
+    public void testBlocking() {
+        await()
+                .atMost(Duration.ofSeconds(3))
+                .until(() -> scheduled.blockingCalled() > 0);
+    }
+
+    @Test
+    public void testNonBlocking() {
+        await()
+                .atMost(Duration.ofSeconds(3))
+                .until(() -> scheduled.nonBlockingCalled() > 0);
+    }
+
+    public static class MyScheduledClass {
+
+        private final AtomicInteger blockingCalled = new AtomicInteger();
+        private final AtomicInteger nonBlockingCalled = new AtomicInteger();
+
+        @Scheduled(every = "1m")
+        public void blocking() {
+            Context context = Vertx.currentContext();
+            Assertions.assertNotNull(context);
+            Assertions.assertTrue(VertxContext.isDuplicatedContext(context));
+            Assertions.assertTrue(VertxContext.isOnDuplicatedContext());
+
+            blockingCalled.incrementAndGet();
+        }
+
+        @Scheduled(every = "1m")
+        public Uni<Void> nonblocking() {
+            Context context = Vertx.currentContext();
+            Assertions.assertNotNull(context);
+            Assertions.assertTrue(VertxContext.isDuplicatedContext(context));
+            Assertions.assertTrue(VertxContext.isOnDuplicatedContext());
+
+            nonBlockingCalled.incrementAndGet();
+            return Uni.createFrom().voidItem();
+        }
+
+        public int blockingCalled() {
+            return blockingCalled.get();
+        }
+
+        public int nonBlockingCalled() {
+            return nonBlockingCalled.get();
+        }
+    }
+}

extensions/scheduler/runtime/src/main/java/io/quarkus/scheduler/runtime/SchedulerRecorder.java

@@ -1,7 +1,6 @@
 package io.quarkus.scheduler.runtime;
 
 import java.util.List;
-import java.util.concurrent.ExecutorService;
 import java.util.function.Supplier;
 
 import com.cronutils.model.CronType;
@@ -13,18 +12,13 @@
 @Recorder
 public class SchedulerRecorder {
 
-    public Supplier<Object> createContext(SchedulerConfig config, ExecutorService executorService,
+    public Supplier<Object> createContext(SchedulerConfig config,
             List<ScheduledMethodMetadata> scheduledMethods) {
         return new Supplier<Object>() {
             @Override
             public Object get() {
                 return new SchedulerContext() {
 
-                    @Override
-                    public ExecutorService getExecutor() {
-                        return executorService;
-                    }
-
                     @Override
                     public CronType getCronType() {
                         return config.cronType;

extensions/scheduler/runtime/src/main/java/io/quarkus/scheduler/runtime/SimpleScheduler.java

@@ -10,8 +10,6 @@
 import java.util.Objects;
 import java.util.Optional;
 import java.util.OptionalLong;
-import java.util.concurrent.ExecutorService;
-import java.util.concurrent.RejectedExecutionException;
 import java.util.concurrent.ScheduledExecutorService;
 import java.util.concurrent.TimeUnit;
 import java.util.stream.Collectors;
@@ -55,6 +53,7 @@
 import io.smallrye.common.vertx.VertxContext;
 import io.vertx.core.Context;
 import io.vertx.core.Handler;
+import io.vertx.core.Promise;
 import io.vertx.core.Vertx;
 
 @Typed(Scheduler.class)
@@ -67,7 +66,6 @@ public class SimpleScheduler implements Scheduler {
     private static final long CHECK_PERIOD = 1000L;
 
     private final ScheduledExecutorService scheduledExecutor;
-    private final ExecutorService executor;
     private final Vertx vertx;
     private volatile boolean running;
     private final List<ScheduledTask> scheduledTasks;
@@ -79,7 +77,6 @@ public SimpleScheduler(SchedulerContext context, SchedulerRuntimeConfig schedule
         this.running = true;
         this.enabled = schedulerRuntimeConfig.enabled;
         this.scheduledTasks = new ArrayList<>();
-        this.executor = context.getExecutor();
         this.vertx = vertx;
 
         if (!schedulerRuntimeConfig.enabled) {
@@ -155,7 +152,7 @@ void checkTriggers() {
         ZonedDateTime now = ZonedDateTime.now();
         LOG.tracef("Check triggers at %s", now);
         for (ScheduledTask task : scheduledTasks) {
-            task.execute(now, executor, vertx);
+            task.execute(now, vertx);
         }
     }
 
@@ -301,26 +298,26 @@ static class ScheduledTask {
             this.invoker = invoker;
         }
 
-        void execute(ZonedDateTime now, ExecutorService executor, Vertx vertx) {
+        void execute(ZonedDateTime now, Vertx vertx) {
             if (!trigger.isRunning()) {
                 return;
             }
             ZonedDateTime scheduledFireTime = trigger.evaluate(now);
             if (scheduledFireTime != null) {
+                Context context = VertxContext.getOrCreateDuplicatedContext(vertx);
+                VertxContextSafetyToggle.setContextSafe(context, true);
                 if (invoker.isBlocking()) {
-                    try {
-                        executor.execute(new Runnable() {
-                            @Override
-                            public void run() {
+                    context.executeBlocking(new Handler<Promise<Object>>() {
+                        @Override
+                        public void handle(Promise<Object> p) {
+                            try {
                                 doInvoke(now, scheduledFireTime);
+                            } finally {
+                                p.complete();
                             }
-                        });
-                    } catch (RejectedExecutionException e) {
-                        LOG.warnf("Rejected execution of a scheduled task for trigger %s", trigger);
-                    }
+                        }
+                    }, false);
                 } else {
-                    Context context = VertxContext.getOrCreateDuplicatedContext(vertx);
-                    VertxContextSafetyToggle.setContextSafe(context, true);
                     context.runOnContext(new Handler<Void>() {
                         @Override
                         public void handle(Void event) {

extensions/scheduler/runtime/src/main/java/io/quarkus/scheduler/runtime/devconsole/SchedulerDevConsoleRecorder.java

@@ -17,7 +17,9 @@
 import io.quarkus.scheduler.common.runtime.ScheduledMethodMetadata;
 import io.quarkus.scheduler.common.runtime.SchedulerContext;
 import io.quarkus.scheduler.common.runtime.util.SchedulerUtils;
+import io.quarkus.vertx.core.runtime.context.VertxContextSafetyToggle;
 import io.smallrye.common.vertx.VertxContext;
+import io.vertx.core.Context;
 import io.vertx.core.Handler;
 import io.vertx.core.MultiMap;
 import io.vertx.core.Vertx;
@@ -80,39 +82,36 @@ protected void handlePost(RoutingContext ctx, MultiMap form) throws Exception {
                     SchedulerContext context = Arc.container().instance(SchedulerContext.class).get();
                     for (ScheduledMethodMetadata metadata : context.getScheduledMethods()) {
                         if (metadata.getMethodDescription().equals(name)) {
-                            context.getExecutor().execute(new Runnable() {
-                                @Override
-                                public void run() {
-                                    final ClassLoader previousCl = Thread.currentThread().getContextClassLoader();
-                                    try {
-                                        Thread.currentThread().setContextClassLoader(currentCl);
-                                        ScheduledInvoker invoker = context
-                                                .createInvoker(metadata.getInvokerClassName());
-                                        if (invoker.isBlocking()) {
+                            Vertx vertx = Arc.container().instance(Vertx.class).get();
+                            Context vdc = VertxContext.getOrCreateDuplicatedContext(vertx);
+                            VertxContextSafetyToggle.setContextSafe(vdc, true);
+                            try {
+                                ScheduledInvoker invoker = context
+                                        .createInvoker(metadata.getInvokerClassName());
+                                if (invoker.isBlocking()) {
+                                    vdc.executeBlocking(p -> {
+                                        try {
                                             invoker.invoke(new DevModeScheduledExecution());
-                                        } else {
-                                            Vertx vertx = Arc.container().instance(Vertx.class).get();
-                                            VertxContext.getOrCreateDuplicatedContext(vertx).runOnContext(new Handler<Void>() {
-                                                @Override
-                                                public void handle(Void event) {
-                                                    try {
-                                                        invoker.invoke(new DevModeScheduledExecution());
-                                                    } catch (Exception ignored) {
-                                                    }
-                                                }
-                                            });
+                                        } catch (Exception ignored) {
+                                        } finally {
+                                            p.complete();
                                         }
-                                        LOG.infof("Invoked scheduled method %s via Dev UI", name);
-                                    } catch (Exception e) {
-                                        LOG.error(
-                                                "Unable to invoke a @Scheduled method: "
-                                                        + metadata.getMethodDescription(),
-                                                e);
-                                    } finally {
-                                        Thread.currentThread().setContextClassLoader(previousCl);
-                                    }
+                                    }, false);
+                                } else {
+                                    vdc.runOnContext(x -> {
+                                        try {
+                                            invoker.invoke(new DevModeScheduledExecution());
+                                        } catch (Exception ignored) {
+                                        }
+                                    });
                                 }
-                            });
+                                LOG.infof("Invoked scheduled method %s via Dev UI", name);
+                            } catch (Exception e) {
+                                LOG.error(
+                                        "Unable to invoke a @Scheduled method: "
+                                                + metadata.getMethodDescription(),
+                                        e);
+                            }
                             flashMessage(ctx, "Action invoked");
                             return;
                         }