Merge pull request quarkusio#44224 from mkouba/issue-44048

Quartz: introduce Nonconcurrent

extensions/quartz/deployment/src/main/java/io/quarkus/quartz/deployment/QuartzProcessor.java

@@ -5,6 +5,7 @@
 import java.sql.Connection;
 import java.util.ArrayList;
 import java.util.HashMap;
+import java.util.HashSet;
 import java.util.List;
 import java.util.Map;
 import java.util.Optional;
@@ -55,6 +56,7 @@
 import io.quarkus.deployment.builditem.nativeimage.NativeImageProxyDefinitionBuildItem;
 import io.quarkus.deployment.builditem.nativeimage.ReflectiveClassBuildItem;
 import io.quarkus.deployment.logging.LogCleanupFilterBuildItem;
+import io.quarkus.quartz.Nonconcurrent;
 import io.quarkus.quartz.runtime.QuarkusQuartzConnectionPoolProvider;
 import io.quarkus.quartz.runtime.QuartzBuildTimeConfig;
 import io.quarkus.quartz.runtime.QuartzExtensionPointConfig;
@@ -69,6 +71,7 @@
 import io.quarkus.quartz.runtime.jdbc.QuarkusStdJDBCDelegate;
 import io.quarkus.runtime.configuration.ConfigurationException;
 import io.quarkus.scheduler.Scheduled;
+import io.quarkus.scheduler.deployment.ScheduledBusinessMethodItem;
 import io.quarkus.scheduler.deployment.SchedulerImplementationBuildItem;
 
 public class QuartzProcessor {
@@ -79,6 +82,7 @@ public class QuartzProcessor {
     private static final DotName DELEGATE_HSQLDB = DotName.createSimple(QuarkusHSQLDBDelegate.class.getName());
     private static final DotName DELEGATE_MSSQL = DotName.createSimple(QuarkusMSSQLDelegate.class.getName());
     private static final DotName DELEGATE_STDJDBC = DotName.createSimple(QuarkusStdJDBCDelegate.class.getName());
+    private static final DotName NONCONCURRENT = DotName.createSimple(Nonconcurrent.class);
 
     @BuildStep
     FeatureBuildItem feature() {
@@ -313,12 +317,23 @@ public void start(BuildProducer<ServiceStartBuildItem> serviceStart,
     @Record(RUNTIME_INIT)
     public void quartzSupportBean(QuartzRuntimeConfig runtimeConfig, QuartzBuildTimeConfig buildTimeConfig,
             QuartzRecorder recorder,
-            BuildProducer<SyntheticBeanBuildItem> syntheticBeanBuildItemBuildProducer,
-            QuartzJDBCDriverDialectBuildItem driverDialect) {
+            QuartzJDBCDriverDialectBuildItem driverDialect,
+            List<ScheduledBusinessMethodItem> scheduledMethods,
+            BuildProducer<SyntheticBeanBuildItem> syntheticBeanBuildItemBuildProducer) {
+
+        Set<String> nonconcurrentMethods = new HashSet<>();
+        for (ScheduledBusinessMethodItem m : scheduledMethods) {
+            if (m.getMethod().hasAnnotation(NONCONCURRENT)) {
+                nonconcurrentMethods.add(m.getMethod().declaringClass().name() + "#" + m.getMethod().name());
+            }
+        }
 
         syntheticBeanBuildItemBuildProducer.produce(SyntheticBeanBuildItem.configure(QuartzSupport.class)
                 .scope(Singleton.class) // this should be @ApplicationScoped but it fails for some reason
                 .setRuntimeInit()
-                .supplier(recorder.quartzSupportSupplier(runtimeConfig, buildTimeConfig, driverDialect.getDriver())).done());
+                .supplier(recorder.quartzSupportSupplier(runtimeConfig, buildTimeConfig, driverDialect.getDriver(),
+                        nonconcurrentMethods))
+                .done());
     }
+
 }

extensions/quartz/deployment/src/test/java/io/quarkus/quartz/test/NonconcurrentJobDefinitionTest.java

@@ -0,0 +1,67 @@
+package io.quarkus.quartz.test;
+
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicInteger;
+
+import jakarta.inject.Inject;
+
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.extension.RegisterExtension;
+
+import io.quarkus.quartz.QuartzScheduler;
+import io.quarkus.scheduler.Scheduled;
+import io.quarkus.test.QuarkusUnitTest;
+
+public class NonconcurrentJobDefinitionTest {
+
+    @RegisterExtension
+    static final QuarkusUnitTest test = new QuarkusUnitTest()
+            .withApplicationRoot(root -> root.addClasses(Jobs.class))
+            .overrideConfigKey("quarkus.scheduler.start-mode", "forced");
+
+    @Inject
+    QuartzScheduler scheduler;
+
+    @Test
+    public void testExecution() throws InterruptedException {
+        scheduler.newJob("foo")
+                .setTask(se -> {
+                    Jobs.NONCONCURRENT_COUNTER.incrementAndGet();
+                    try {
+                        if (!Jobs.CONCURRENT_LATCH.await(10, TimeUnit.SECONDS)) {
+                            throw new IllegalStateException("nonconcurrent() execution blocked too long...");
+                        }
+                    } catch (InterruptedException e) {
+                        throw new IllegalStateException(e);
+                    }
+                    if (Jobs.NONCONCURRENT_COUNTER.get() == 1) {
+                        // concurrent() executed >= 5x and nonconcurrent() 1x
+                        Jobs.NONCONCURRENT_LATCH.countDown();
+                    }
+                })
+                .setInterval("1s")
+                .setNonconcurrent()
+                .schedule();
+
+        assertTrue(Jobs.NONCONCURRENT_LATCH.await(10, TimeUnit.SECONDS),
+                String.format("nonconcurrent() executed: %sx", Jobs.NONCONCURRENT_COUNTER.get()));
+    }
+
+    static class Jobs {
+
+        static final CountDownLatch NONCONCURRENT_LATCH = new CountDownLatch(1);
+        static final CountDownLatch CONCURRENT_LATCH = new CountDownLatch(5);
+
+        static final AtomicInteger NONCONCURRENT_COUNTER = new AtomicInteger(0);
+
+        @Scheduled(identity = "bar", every = "1s")
+        void concurrent() throws InterruptedException {
+            CONCURRENT_LATCH.countDown();
+        }
+
+    }
+
+}

extensions/quartz/deployment/src/test/java/io/quarkus/quartz/test/NonconcurrentOnQuartzThreadTest.java

@@ -0,0 +1,56 @@
+package io.quarkus.quartz.test;
+
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicInteger;
+
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.extension.RegisterExtension;
+
+import io.quarkus.quartz.Nonconcurrent;
+import io.quarkus.scheduler.Scheduled;
+import io.quarkus.test.QuarkusUnitTest;
+
+public class NonconcurrentOnQuartzThreadTest {
+
+    @RegisterExtension
+    static final QuarkusUnitTest test = new QuarkusUnitTest()
+            .withApplicationRoot(root -> root.addClasses(Jobs.class))
+            .overrideConfigKey("quarkus.quartz.run-blocking-scheduled-method-on-quartz-thread",
+                    "true");
+
+    @Test
+    public void testExecution() throws InterruptedException {
+        assertTrue(Jobs.NONCONCURRENT_LATCH.await(10, TimeUnit.SECONDS),
+                String.format("nonconcurrent() executed: %sx", Jobs.NONCONCURRENT_COUNTER.get()));
+    }
+
+    static class Jobs {
+
+        static final CountDownLatch NONCONCURRENT_LATCH = new CountDownLatch(1);
+        static final CountDownLatch CONCURRENT_LATCH = new CountDownLatch(5);
+
+        static final AtomicInteger NONCONCURRENT_COUNTER = new AtomicInteger(0);
+
+        @Nonconcurrent
+        @Scheduled(identity = "foo", every = "1s")
+        void nonconcurrent() throws InterruptedException {
+            NONCONCURRENT_COUNTER.incrementAndGet();
+            if (!CONCURRENT_LATCH.await(10, TimeUnit.SECONDS)) {
+                throw new IllegalStateException("nonconcurrent() execution blocked too long...");
+            }
+            if (NONCONCURRENT_COUNTER.get() == 1) {
+                // concurrent() executed >= 5x and nonconcurrent() 1x
+                NONCONCURRENT_LATCH.countDown();
+            }
+        }
+
+        @Scheduled(identity = "bar", every = "1s")
+        void concurrent() throws InterruptedException {
+            CONCURRENT_LATCH.countDown();
+        }
+
+    }
+}

extensions/quartz/deployment/src/test/java/io/quarkus/quartz/test/NonconcurrentProgrammaticTest.java

@@ -0,0 +1,91 @@
+package io.quarkus.quartz.test;
+
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicInteger;
+
+import jakarta.inject.Inject;
+
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.extension.RegisterExtension;
+import org.quartz.DisallowConcurrentExecution;
+import org.quartz.Job;
+import org.quartz.JobBuilder;
+import org.quartz.JobDetail;
+import org.quartz.JobExecutionContext;
+import org.quartz.JobExecutionException;
+import org.quartz.SchedulerException;
+import org.quartz.SimpleScheduleBuilder;
+import org.quartz.Trigger;
+import org.quartz.TriggerBuilder;
+
+import io.quarkus.quartz.QuartzScheduler;
+import io.quarkus.scheduler.Scheduled;
+import io.quarkus.scheduler.Scheduler;
+import io.quarkus.test.QuarkusUnitTest;
+
+public class NonconcurrentProgrammaticTest {
+
+    @RegisterExtension
+    static final QuarkusUnitTest test = new QuarkusUnitTest()
+            .withApplicationRoot(root -> root
+                    .addClasses(Jobs.class))
+            .overrideConfigKey("quarkus.scheduler.start-mode", "halted");
+
+    @Inject
+    QuartzScheduler scheduler;
+
+    @Test
+    public void testExecution() throws SchedulerException, InterruptedException {
+        JobDetail job = JobBuilder.newJob(Jobs.class)
+                .withIdentity("foo", Scheduler.class.getName())
+                .build();
+        Trigger trigger = TriggerBuilder.newTrigger()
+                .withIdentity("foo", Scheduler.class.getName())
+                .startNow()
+                .withSchedule(SimpleScheduleBuilder.simpleSchedule()
+                        .withIntervalInSeconds(1)
+                        .repeatForever())
+                .build();
+        scheduler.getScheduler().scheduleJob(job, trigger);
+
+        scheduler.resume();
+
+        assertTrue(Jobs.NONCONCURRENT_LATCH.await(10, TimeUnit.SECONDS),
+                String.format("nonconcurrent() executed: %sx", Jobs.NONCONCURRENT_COUNTER.get()));
+    }
+
+    @DisallowConcurrentExecution
+    static class Jobs implements Job {
+
+        static final CountDownLatch NONCONCURRENT_LATCH = new CountDownLatch(1);
+        static final CountDownLatch CONCURRENT_LATCH = new CountDownLatch(5);
+
+        static final AtomicInteger NONCONCURRENT_COUNTER = new AtomicInteger(0);
+
+        @Scheduled(identity = "bar", every = "1s")
+        void concurrent() throws InterruptedException {
+            CONCURRENT_LATCH.countDown();
+        }
+
+        @Override
+        public void execute(JobExecutionContext context) throws JobExecutionException {
+            Jobs.NONCONCURRENT_COUNTER.incrementAndGet();
+            try {
+                if (!Jobs.CONCURRENT_LATCH.await(10, TimeUnit.SECONDS)) {
+                    throw new IllegalStateException("nonconcurrent() execution blocked too long...");
+                }
+            } catch (InterruptedException e) {
+                throw new IllegalStateException(e);
+            }
+            if (Jobs.NONCONCURRENT_COUNTER.get() == 1) {
+                // concurrent() executed >= 5x and nonconcurrent() 1x
+                Jobs.NONCONCURRENT_LATCH.countDown();
+            }
+        }
+
+    }
+
+}

extensions/quartz/deployment/src/test/java/io/quarkus/quartz/test/NonconcurrentTest.java

@@ -0,0 +1,54 @@
+package io.quarkus.quartz.test;
+
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicInteger;
+
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.extension.RegisterExtension;
+
+import io.quarkus.quartz.Nonconcurrent;
+import io.quarkus.scheduler.Scheduled;
+import io.quarkus.test.QuarkusUnitTest;
+
+public class NonconcurrentTest {
+
+    @RegisterExtension
+    static final QuarkusUnitTest test = new QuarkusUnitTest()
+            .withApplicationRoot(root -> root.addClasses(Jobs.class));
+
+    @Test
+    public void testExecution() throws InterruptedException {
+        assertTrue(Jobs.NONCONCURRENT_LATCH.await(10, TimeUnit.SECONDS),
+                String.format("nonconcurrent() executed: %sx", Jobs.NONCONCURRENT_COUNTER.get()));
+    }
+
+    static class Jobs {
+
+        static final CountDownLatch NONCONCURRENT_LATCH = new CountDownLatch(1);
+        static final CountDownLatch CONCURRENT_LATCH = new CountDownLatch(5);
+
+        static final AtomicInteger NONCONCURRENT_COUNTER = new AtomicInteger(0);
+
+        @Nonconcurrent
+        @Scheduled(identity = "foo", every = "1s")
+        void nonconcurrent() throws InterruptedException {
+            NONCONCURRENT_COUNTER.incrementAndGet();
+            if (!CONCURRENT_LATCH.await(10, TimeUnit.SECONDS)) {
+                throw new IllegalStateException("nonconcurrent() execution blocked too long...");
+            }
+            if (NONCONCURRENT_COUNTER.get() == 1) {
+                // concurrent() executed >= 5x and nonconcurrent() 1x
+                NONCONCURRENT_LATCH.countDown();
+            }
+        }
+
+        @Scheduled(identity = "bar", every = "1s")
+        void concurrent() throws InterruptedException {
+            CONCURRENT_LATCH.countDown();
+        }
+
+    }
+}

extensions/quartz/deployment/src/test/java/io/quarkus/quartz/test/programmatic/ProgrammaticJobsTest.java

@@ -69,7 +69,7 @@ public void testJobs() throws InterruptedException {
                 .setSkipPredicate(AlwaysSkipPredicate.class)
                 .schedule();
 
-        Scheduler.JobDefinition job1 = scheduler.newJob("foo")
+        Scheduler.JobDefinition<?> job1 = scheduler.newJob("foo")
                 .setInterval("1s")
                 .setTask(ec -> {
                     assertTrue(Arc.container().requestContext().isActive());
@@ -79,7 +79,7 @@ public void testJobs() throws InterruptedException {
         assertEquals("Sync task was already set",
                 assertThrows(IllegalStateException.class, () -> job1.setAsyncTask(ec -> null)).getMessage());
 
-        Scheduler.JobDefinition job2 = scheduler.newJob("foo").setCron("0/5 * * * * ?");
+        Scheduler.JobDefinition<?> job2 = scheduler.newJob("foo").setCron("0/5 * * * * ?");
         assertEquals("Either sync or async task must be set",
                 assertThrows(IllegalStateException.class, () -> job2.schedule()).getMessage());
         job2.setTask(ec -> {
@@ -117,7 +117,7 @@ public void testJobs() throws InterruptedException {
     @Test
     public void testAsyncJob() throws InterruptedException, SchedulerException {
         String identity = "fooAsync";
-        JobDefinition asyncJob = scheduler.newJob(identity)
+        JobDefinition<?> asyncJob = scheduler.newJob(identity)
                 .setInterval("1s")
                 .setAsyncTask(ec -> {
                     assertTrue(Context.isOnEventLoopThread() && VertxContext.isOnDuplicatedContext());

extensions/quartz/runtime/src/main/java/io/quarkus/quartz/Nonconcurrent.java

@@ -0,0 +1,35 @@
+package io.quarkus.quartz;
+
+import static java.lang.annotation.ElementType.METHOD;
+import static java.lang.annotation.RetentionPolicy.RUNTIME;
+
+import java.lang.annotation.Retention;
+import java.lang.annotation.Target;
+
+import org.quartz.DisallowConcurrentExecution;
+import org.quartz.Job;
+
+import io.quarkus.scheduler.Scheduled;
+import io.quarkus.scheduler.SkippedExecution;
+
+/**
+ * A scheduled method annotated with this annotation may not be executed concurrently. The behavior is identical to a
+ * {@link Job} class annotated with {@link DisallowConcurrentExecution}.
+ * <p>
+ * If {@code quarkus.quartz.run-blocking-scheduled-method-on-quartz-thread} is set to
+ * {@code false} the execution of a scheduled method is offloaded to a specific Quarkus thread pool but the triggering Quartz
+ * thread is blocked until the execution is finished. Therefore, make sure the Quartz thread pool is configured appropriately.
+ * <p>
+ * If {@code quarkus.quartz.run-blocking-scheduled-method-on-quartz-thread} is set to {@code true} the scheduled method is
+ * invoked on a thread managed by Quartz.
+ * <p>
+ * Unlike with {@link Scheduled.ConcurrentExecution#SKIP} the {@link SkippedExecution} event is never fired if a method
+ * execution is skipped by Quartz.
+ *
+ * @see DisallowConcurrentExecution
+ */
+@Target(METHOD)
+@Retention(RUNTIME)
+public @interface Nonconcurrent {
+
+}