diff --git a/extensions/quartz/deployment/src/main/java/io/quarkus/quartz/deployment/QuartzProcessor.java b/extensions/quartz/deployment/src/main/java/io/quarkus/quartz/deployment/QuartzProcessor.java index 30f0727e9c68b1..c74d35b710596e 100644 --- a/extensions/quartz/deployment/src/main/java/io/quarkus/quartz/deployment/QuartzProcessor.java +++ b/extensions/quartz/deployment/src/main/java/io/quarkus/quartz/deployment/QuartzProcessor.java @@ -64,10 +64,9 @@ import io.quarkus.quartz.runtime.jdbc.QuarkusPostgreSQLDelegate; import io.quarkus.quartz.runtime.jdbc.QuarkusStdJDBCDelegate; import io.quarkus.runtime.configuration.ConfigurationException; +import io.quarkus.scheduler.Scheduled; +import io.quarkus.scheduler.deployment.SchedulerImplementationBuildItem; -/** - * - */ public class QuartzProcessor { private static final DotName JOB = DotName.createSimple(Job.class.getName()); @@ -77,6 +76,11 @@ FeatureBuildItem feature() { return new FeatureBuildItem(Feature.QUARTZ); } + @BuildStep + SchedulerImplementationBuildItem implementation() { + return new SchedulerImplementationBuildItem(Scheduled.QUARTZ, DotName.createSimple(QuartzSchedulerImpl.class), 1); + } + @BuildStep AdditionalBeanBuildItem beans() { return new AdditionalBeanBuildItem(QuartzSchedulerImpl.class); diff --git a/extensions/quartz/deployment/src/test/java/io/quarkus/quartz/test/composite/CompositeSchedulerNotUsedTest.java b/extensions/quartz/deployment/src/test/java/io/quarkus/quartz/test/composite/CompositeSchedulerNotUsedTest.java new file mode 100644 index 00000000000000..08a8dd58a85b9f --- /dev/null +++ b/extensions/quartz/deployment/src/test/java/io/quarkus/quartz/test/composite/CompositeSchedulerNotUsedTest.java @@ -0,0 +1,36 @@ +package io.quarkus.quartz.test.composite; + +import static org.assertj.core.api.Assertions.assertThat; +import static org.junit.jupiter.api.Assertions.fail; + +import org.junit.jupiter.api.Test; +import org.junit.jupiter.api.extension.RegisterExtension; + +import io.quarkus.scheduler.Scheduled; +import io.quarkus.test.QuarkusUnitTest; + +public class CompositeSchedulerNotUsedTest { + + @RegisterExtension + static final QuarkusUnitTest test = new QuarkusUnitTest() + .withApplicationRoot(root -> root + .addClasses(Jobs.class)) + .assertException(t -> { + assertThat(t).cause().isInstanceOf(IllegalStateException.class) + .hasMessageContaining( + "The required scheduler implementation is not available because the composite scheduler is not used: SIMPLE"); + }); + + @Test + public void test() { + fail(); + } + + static class Jobs { + + @Scheduled(every = "1s", executeWith = Scheduled.SIMPLE) + void quartz() { + } + + } +} diff --git a/extensions/quartz/deployment/src/test/java/io/quarkus/quartz/test/composite/CompositeSchedulerTest.java b/extensions/quartz/deployment/src/test/java/io/quarkus/quartz/test/composite/CompositeSchedulerTest.java new file mode 100644 index 00000000000000..d08607b05e317e --- /dev/null +++ b/extensions/quartz/deployment/src/test/java/io/quarkus/quartz/test/composite/CompositeSchedulerTest.java @@ -0,0 +1,99 @@ +package io.quarkus.quartz.test.composite; + +import static org.junit.jupiter.api.Assertions.assertFalse; +import static org.junit.jupiter.api.Assertions.assertNotNull; +import static org.junit.jupiter.api.Assertions.assertNull; +import static org.junit.jupiter.api.Assertions.assertTrue; + +import java.util.concurrent.CountDownLatch; +import java.util.concurrent.TimeUnit; + +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.scheduler.Scheduler; +import io.quarkus.scheduler.runtime.Constituent; +import io.quarkus.test.QuarkusUnitTest; +import io.smallrye.common.annotation.Identifier; + +public class CompositeSchedulerTest { + + @RegisterExtension + static final QuarkusUnitTest test = new QuarkusUnitTest() + .withApplicationRoot(root -> root + .addClasses(Jobs.class)) + .overrideConfigKey("quarkus.scheduler.use-composite-scheduler", "true"); + + @Constituent + QuartzScheduler quartz; + + @Constituent + @Identifier("SIMPLE") + Scheduler simple; + + @Inject + Scheduler composite; + + @Test + public void testExecution() throws InterruptedException { + assertTrue(Jobs.simpleLatch.await(5, TimeUnit.SECONDS)); + assertTrue(Jobs.quartzLatch.await(5, TimeUnit.SECONDS)); + assertTrue(Jobs.autoLatch.await(5, TimeUnit.SECONDS)); + + assertNull(quartz.getScheduledJob("simple")); + assertNotNull(quartz.getScheduledJob("quartz")); + assertNotNull(quartz.getScheduledJob("auto")); + + assertNotNull(simple.getScheduledJob("simple")); + assertNull(simple.getScheduledJob("quartz")); + assertNull(simple.getScheduledJob("auto")); + + assertNotNull(composite.getScheduledJob("quartz")); + assertNotNull(composite.getScheduledJob("auto")); + assertNotNull(composite.getScheduledJob("simple")); + + composite.pause(); + Jobs.reset(); + assertFalse(composite.isRunning()); + assertFalse(Jobs.simpleLatch.await(2, TimeUnit.SECONDS)); + + composite.resume(); + assertTrue(composite.isRunning()); + assertTrue(Jobs.simpleLatch.await(5, TimeUnit.SECONDS)); + assertTrue(Jobs.quartzLatch.await(5, TimeUnit.SECONDS)); + assertTrue(Jobs.autoLatch.await(5, TimeUnit.SECONDS)); + } + + static class Jobs { + + static CountDownLatch simpleLatch = new CountDownLatch(1); + static CountDownLatch quartzLatch = new CountDownLatch(1); + static CountDownLatch autoLatch = new CountDownLatch(1); + + static void reset() { + simpleLatch = new CountDownLatch(1); + quartzLatch = new CountDownLatch(1); + autoLatch = new CountDownLatch(1); + } + + @Scheduled(identity = "simple", every = "1s", executeWith = Scheduled.SIMPLE) + void simple() { + simpleLatch.countDown(); + } + + @Scheduled(identity = "quartz", every = "1s", executeWith = Scheduled.QUARTZ) + void quartz() { + quartzLatch.countDown(); + } + + @Scheduled(identity = "auto", every = "1s") + void auto() { + autoLatch.countDown(); + } + + } +} diff --git a/extensions/quartz/runtime/src/main/java/io/quarkus/quartz/runtime/QuartzSchedulerImpl.java b/extensions/quartz/runtime/src/main/java/io/quarkus/quartz/runtime/QuartzSchedulerImpl.java index 7e08b2a596de56..78c859eefe8ffd 100644 --- a/extensions/quartz/runtime/src/main/java/io/quarkus/quartz/runtime/QuartzSchedulerImpl.java +++ b/extensions/quartz/runtime/src/main/java/io/quarkus/quartz/runtime/QuartzSchedulerImpl.java @@ -198,7 +198,8 @@ public QuartzSchedulerImpl(SchedulerContext context, QuartzSupport quartzSupport if (!enabled) { LOGGER.info("Quartz scheduler is disabled by config property and will not be started"); this.scheduler = null; - } else if (!forceStart && context.getScheduledMethods().isEmpty() && !context.forceSchedulerStart()) { + } else if (!forceStart && context.getScheduledMethods(Scheduled.QUARTZ).isEmpty() + && !context.forceSchedulerStart()) { LOGGER.info("No scheduled business methods found - Quartz scheduler will not be started"); this.scheduler = null; } else { @@ -232,10 +233,13 @@ public org.quartz.Trigger apply(TriggerKey triggerKey) { } }; - for (ScheduledMethod method : context.getScheduledMethods()) { + for (ScheduledMethod method : context.getScheduledMethods(Scheduled.QUARTZ)) { int nameSequence = 0; for (Scheduled scheduled : method.getSchedules()) { + if (!context.matchesImplementation(scheduled, Scheduled.QUARTZ)) { + continue; + } String identity = SchedulerUtils.lookUpPropertyValue(scheduled.identity()); if (identity.isEmpty()) { identity = ++nameSequence + "_" + method.getInvokerClassName(); @@ -345,6 +349,11 @@ public org.quartz.Scheduler getScheduler() { return scheduler; } + @Override + public String implementation() { + return Scheduled.QUARTZ; + } + @Override public void pause() { if (!enabled) { @@ -893,7 +902,7 @@ public Trigger schedule() { } scheduled = true; SyntheticScheduled scheduled = new SyntheticScheduled(identity, cron, every, 0, TimeUnit.MINUTES, delayed, - overdueGracePeriod, concurrentExecution, skipPredicate, timeZone); + overdueGracePeriod, concurrentExecution, skipPredicate, timeZone, implementation); return createJobDefinitionQuartzTrigger(this, scheduled, null); } diff --git a/extensions/scheduler/api/src/main/java/io/quarkus/scheduler/Scheduled.java b/extensions/scheduler/api/src/main/java/io/quarkus/scheduler/Scheduled.java index e13e64c127ae2a..6cae49cf5d037f 100644 --- a/extensions/scheduler/api/src/main/java/io/quarkus/scheduler/Scheduled.java +++ b/extensions/scheduler/api/src/main/java/io/quarkus/scheduler/Scheduled.java @@ -62,6 +62,28 @@ */ String DEFAULT_TIMEZONE = "<>"; + /** + * Constant value for {@link #executeWith()} indicating that the implementation should be selected automatically, i.e. the + * implementation with highest priority is used. + */ + String AUTO = "<>"; + + /** + * Constant value for {@link #executeWith()} indicating that the simple in-memory implementation provided by the + * {@code quarkus-scheduler} extension should be used. + *

+ * This implementation has priority {@code 0}. + */ + String SIMPLE = "SIMPLE"; + + /** + * Constant value for {@link #executeWith()} indicating that the Quartz implementation provided by the + * {@code quarkus-quartz} extension should be used. + *

+ * This implementation has priority {@code 1}. + */ + String QUARTZ = "QUARTZ"; + /** * Optionally defines a unique identifier for this job. *

@@ -205,6 +227,20 @@ */ String timeZone() default DEFAULT_TIMEZONE; + /** + * The scheduler implementation used to execute this scheduled method. + *

+ * By default, the implementation with highest priority is selected automatically. + *

+ * If the required implementation is not available, then the build fails. + * + * @return the implementation to execute this scheduled method + * @see #AUTO + * @see #SIMPLE + * @see #QUARTZ + */ + String executeWith() default AUTO; + @Retention(RUNTIME) @Target(METHOD) @interface Schedules { diff --git a/extensions/scheduler/api/src/main/java/io/quarkus/scheduler/Scheduler.java b/extensions/scheduler/api/src/main/java/io/quarkus/scheduler/Scheduler.java index f5dd65e7440c6a..be5e6575df7f25 100644 --- a/extensions/scheduler/api/src/main/java/io/quarkus/scheduler/Scheduler.java +++ b/extensions/scheduler/api/src/main/java/io/quarkus/scheduler/Scheduler.java @@ -44,10 +44,14 @@ public interface Scheduler { /** * Identity must not be null and {@code false} is returned for non-existent identity. + *

+ * Note that this method only returns {@code true} if the job was explicitly paused. I.e. it does not reflect a paused + * scheduler. * * @param identity * @return {@code true} if the job with the given identity is paused, {@code false} otherwise * @see Scheduled#identity() + * @see #pause(String) */ boolean isPaused(String identity); @@ -88,6 +92,13 @@ public interface Scheduler { */ Trigger unscheduleJob(String identity); + /** + * + * @return the implementation + * @see Scheduled#executeWith() + */ + String implementation(); + /** * The job definition is a builder-like API that can be used to define a job programmatically. *

@@ -177,6 +188,16 @@ interface JobDefinition { */ JobDefinition setTimeZone(String timeZone); + /** + * {@link Scheduled#executeWith()} + * + * @param implementation + * @return self + * @throws IllegalArgumentException If the selected implementation is not available + * @see Scheduled#executeWith() + */ + JobDefinition setExecuteWith(String implementation); + /** * * @param task diff --git a/extensions/scheduler/common/src/main/java/io/quarkus/scheduler/common/runtime/AbstractJobDefinition.java b/extensions/scheduler/common/src/main/java/io/quarkus/scheduler/common/runtime/AbstractJobDefinition.java index eef7fe84d7c27a..16e24a2779e766 100644 --- a/extensions/scheduler/common/src/main/java/io/quarkus/scheduler/common/runtime/AbstractJobDefinition.java +++ b/extensions/scheduler/common/src/main/java/io/quarkus/scheduler/common/runtime/AbstractJobDefinition.java @@ -28,6 +28,7 @@ public abstract class AbstractJobDefinition implements JobDefinition { protected boolean scheduled = false; protected String timeZone = Scheduled.DEFAULT_TIMEZONE; protected boolean runOnVirtualThread; + protected String implementation = Scheduled.AUTO; public AbstractJobDefinition(String identity) { this.identity = identity; @@ -88,6 +89,12 @@ public JobDefinition setTimeZone(String timeZone) { return this; } + @Override + public JobDefinition setExecuteWith(String implementation) { + this.implementation = implementation; + return this; + } + @Override public JobDefinition setTask(Consumer task, boolean runOnVirtualThread) { checkScheduled(); diff --git a/extensions/scheduler/common/src/main/java/io/quarkus/scheduler/common/runtime/SchedulerContext.java b/extensions/scheduler/common/src/main/java/io/quarkus/scheduler/common/runtime/SchedulerContext.java index 5b2a406467a26f..210cff53bd946e 100644 --- a/extensions/scheduler/common/src/main/java/io/quarkus/scheduler/common/runtime/SchedulerContext.java +++ b/extensions/scheduler/common/src/main/java/io/quarkus/scheduler/common/runtime/SchedulerContext.java @@ -5,6 +5,8 @@ import com.cronutils.model.CronType; +import io.quarkus.scheduler.Scheduled; + public interface SchedulerContext { CronType getCronType(); @@ -13,6 +15,10 @@ public interface SchedulerContext { boolean forceSchedulerStart(); + List getScheduledMethods(String implementation); + + boolean matchesImplementation(Scheduled scheduled, String implementation); + @SuppressWarnings("unchecked") default ScheduledInvoker createInvoker(String invokerClassName) { try { diff --git a/extensions/scheduler/common/src/main/java/io/quarkus/scheduler/common/runtime/SyntheticScheduled.java b/extensions/scheduler/common/src/main/java/io/quarkus/scheduler/common/runtime/SyntheticScheduled.java index a598ef744ad6ff..8345b061c925c8 100644 --- a/extensions/scheduler/common/src/main/java/io/quarkus/scheduler/common/runtime/SyntheticScheduled.java +++ b/extensions/scheduler/common/src/main/java/io/quarkus/scheduler/common/runtime/SyntheticScheduled.java @@ -22,10 +22,11 @@ public final class SyntheticScheduled extends AnnotationLiteral imple private final ConcurrentExecution concurrentExecution; private final SkipPredicate skipPredicate; private final String timeZone; + private final String implementation; public SyntheticScheduled(String identity, String cron, String every, long delay, TimeUnit delayUnit, String delayed, String overdueGracePeriod, ConcurrentExecution concurrentExecution, - SkipPredicate skipPredicate, String timeZone) { + SkipPredicate skipPredicate, String timeZone, String implementation) { this.identity = Objects.requireNonNull(identity); this.cron = Objects.requireNonNull(cron); this.every = Objects.requireNonNull(every); @@ -36,6 +37,7 @@ public SyntheticScheduled(String identity, String cron, String every, long delay this.concurrentExecution = Objects.requireNonNull(concurrentExecution); this.skipPredicate = skipPredicate; this.timeZone = timeZone; + this.implementation = implementation; } @Override @@ -88,6 +90,11 @@ public String timeZone() { return timeZone; } + @Override + public String executeWith() { + return implementation; + } + public String toJson() { if (skipPredicate != null) { throw new IllegalStateException("A skipPredicate instance may not be serialized"); @@ -102,6 +109,7 @@ public String toJson() { json.put("overdueGracePeriod", overdueGracePeriod); json.put("concurrentExecution", concurrentExecution.toString()); json.put("timeZone", timeZone); + json.put("executeWith", implementation); return json.encode(); } @@ -110,7 +118,7 @@ public static SyntheticScheduled fromJson(String json) { return new SyntheticScheduled(jsonObj.getString("identity"), jsonObj.getString("cron"), jsonObj.getString("every"), jsonObj.getLong("delay"), TimeUnit.valueOf(jsonObj.getString("delayUnit")), jsonObj.getString("delayed"), jsonObj.getString("overdueGracePeriod"), ConcurrentExecution.valueOf(jsonObj.getString("concurrentExecution")), - null, jsonObj.getString("timeZone")); + null, jsonObj.getString("timeZone"), jsonObj.getString("executeWith")); } @Override diff --git a/extensions/scheduler/common/src/test/java/io/quarkus/scheduler/common/runtime/util/SchedulerUtilsTest.java b/extensions/scheduler/common/src/test/java/io/quarkus/scheduler/common/runtime/util/SchedulerUtilsTest.java index 09083fce197d24..493d4c10ea6025 100644 --- a/extensions/scheduler/common/src/test/java/io/quarkus/scheduler/common/runtime/util/SchedulerUtilsTest.java +++ b/extensions/scheduler/common/src/test/java/io/quarkus/scheduler/common/runtime/util/SchedulerUtilsTest.java @@ -134,6 +134,12 @@ public Class skipExecutionIf() { public String delayed() { return delayed; } + + @Override + public String executeWith() { + return AUTO; + } + }; } } diff --git a/extensions/scheduler/common/src/test/java/io/quarkus/scheduler/common/runtime/util/SyntheticScheduledTest.java b/extensions/scheduler/common/src/test/java/io/quarkus/scheduler/common/runtime/util/SyntheticScheduledTest.java index b80460db3b380a..a8119398ff8b3c 100644 --- a/extensions/scheduler/common/src/test/java/io/quarkus/scheduler/common/runtime/util/SyntheticScheduledTest.java +++ b/extensions/scheduler/common/src/test/java/io/quarkus/scheduler/common/runtime/util/SyntheticScheduledTest.java @@ -15,7 +15,7 @@ public class SyntheticScheduledTest { @Test public void testJson() { SyntheticScheduled s1 = new SyntheticScheduled("foo", "", "2s", 0, TimeUnit.SECONDS, "1s", "15m", - ConcurrentExecution.PROCEED, null, Scheduled.DEFAULT_TIMEZONE); + ConcurrentExecution.PROCEED, null, Scheduled.DEFAULT_TIMEZONE, Scheduled.AUTO); SyntheticScheduled s2 = SyntheticScheduled.fromJson(s1.toJson()); assertEquals(s1.identity(), s2.identity()); assertEquals(s1.concurrentExecution(), s2.concurrentExecution()); @@ -26,6 +26,7 @@ public void testJson() { assertEquals(s1.delayed(), s2.delayed()); assertEquals(s1.overdueGracePeriod(), s2.overdueGracePeriod()); assertEquals(s1.timeZone(), s2.timeZone()); + assertEquals(s1.executeWith(), s2.executeWith()); } } diff --git a/extensions/scheduler/deployment/src/main/java/io/quarkus/scheduler/deployment/DiscoveredImplementationsBuildItem.java b/extensions/scheduler/deployment/src/main/java/io/quarkus/scheduler/deployment/DiscoveredImplementationsBuildItem.java new file mode 100644 index 00000000000000..a9b6eda0f3f438 --- /dev/null +++ b/extensions/scheduler/deployment/src/main/java/io/quarkus/scheduler/deployment/DiscoveredImplementationsBuildItem.java @@ -0,0 +1,63 @@ +package io.quarkus.scheduler.deployment; + +import java.util.Objects; +import java.util.Set; + +import io.quarkus.builder.item.SimpleBuildItem; +import io.quarkus.scheduler.runtime.CompositeScheduler; +import io.quarkus.scheduler.runtime.Constituent; +import io.quarkus.scheduler.runtime.SchedulerConfig; +import io.smallrye.common.annotation.Identifier; + +/** + * This build item holds all discovered {@link io.quarkus.scheduler.Scheduler} implementations sorted by priority. Higher + * priority goes first. + */ +public final class DiscoveredImplementationsBuildItem extends SimpleBuildItem { + + private final String autoImplementation; + + private final Set implementations; + + private final boolean useCompositeScheduler; + + DiscoveredImplementationsBuildItem(String autoImplementation, Set implementations, boolean useCompositeScheduler) { + this.autoImplementation = Objects.requireNonNull(autoImplementation); + this.implementations = Objects.requireNonNull(implementations); + this.useCompositeScheduler = useCompositeScheduler; + } + + /** + * + * @return the implementation with highest priority + */ + public String getAutoImplementation() { + return autoImplementation; + } + + public Set getImplementations() { + return implementations; + } + + /** + * A composite scheduler is used if multiple scheduler implementations are found and + * {@link SchedulerConfig#useCompositeScheduler} is set to {@code true}. + *

+ * The extension will add: + *

+ * + * @return {@code true} if a composite scheduler is used + * @see CompositeScheduler + */ + public boolean isCompositeSchedulerUsed() { + return useCompositeScheduler && implementations.size() > 1; + } + + public boolean isAutoImplementation(String implementation) { + return autoImplementation.equals(implementation); + } + +} diff --git a/extensions/scheduler/deployment/src/main/java/io/quarkus/scheduler/deployment/SchedulerImplementationBuildItem.java b/extensions/scheduler/deployment/src/main/java/io/quarkus/scheduler/deployment/SchedulerImplementationBuildItem.java new file mode 100644 index 00000000000000..e43f78d281f104 --- /dev/null +++ b/extensions/scheduler/deployment/src/main/java/io/quarkus/scheduler/deployment/SchedulerImplementationBuildItem.java @@ -0,0 +1,52 @@ +package io.quarkus.scheduler.deployment; + +import org.jboss.jandex.DotName; + +import io.quarkus.builder.item.MultiBuildItem; +import io.quarkus.scheduler.Scheduled; +import io.quarkus.scheduler.Scheduler; + +/** + * An extension that provides an implementation of {@link Scheduler} must produce this build item. + *

+ * If multiple extensions produce this build item with the same {@link #implementation} value then the build fails. + */ +public final class SchedulerImplementationBuildItem extends MultiBuildItem { + + private final String implementation; + + private final DotName schedulerBeanClass; + + private final int priority; + + public SchedulerImplementationBuildItem(String implementation, DotName schedulerBeanClass, int priority) { + this.implementation = implementation; + this.schedulerBeanClass = schedulerBeanClass; + this.priority = priority; + } + + public String getImplementation() { + return implementation; + } + + public DotName getSchedulerBeanClass() { + return schedulerBeanClass; + } + + /** + * The implementation with highest priority is selected if {@link Implementation#AUTO} is used. + * + * @return the priority + * @see Scheduled.Implementation#AUTO + */ + public int getPriority() { + return priority; + } + + @Override + public String toString() { + return "SchedulerImplementationBuildItem [" + (implementation != null ? "implementation=" + implementation + ", " : "") + + "priority=" + priority + "]"; + } + +} diff --git a/extensions/scheduler/deployment/src/main/java/io/quarkus/scheduler/deployment/SchedulerProcessor.java b/extensions/scheduler/deployment/src/main/java/io/quarkus/scheduler/deployment/SchedulerProcessor.java index 1188ec62004088..2db5a37c62741f 100644 --- a/extensions/scheduler/deployment/src/main/java/io/quarkus/scheduler/deployment/SchedulerProcessor.java +++ b/extensions/scheduler/deployment/src/main/java/io/quarkus/scheduler/deployment/SchedulerProcessor.java @@ -10,6 +10,7 @@ import java.time.Duration; import java.time.ZoneId; import java.util.ArrayList; +import java.util.Comparator; import java.util.HashMap; import java.util.HashSet; import java.util.List; @@ -19,8 +20,10 @@ import java.util.concurrent.CompletableFuture; import java.util.concurrent.CompletionStage; import java.util.function.Function; +import java.util.stream.Collectors; import org.jboss.jandex.AnnotationInstance; +import org.jboss.jandex.AnnotationTransformation; import org.jboss.jandex.AnnotationValue; import org.jboss.jandex.ClassInfo; import org.jboss.jandex.DotName; @@ -82,9 +85,12 @@ import io.quarkus.scheduler.common.runtime.MutableScheduledMethod; import io.quarkus.scheduler.common.runtime.SchedulerContext; import io.quarkus.scheduler.common.runtime.util.SchedulerUtils; +import io.quarkus.scheduler.runtime.CompositeScheduler; +import io.quarkus.scheduler.runtime.Constituent; import io.quarkus.scheduler.runtime.SchedulerConfig; import io.quarkus.scheduler.runtime.SchedulerRecorder; import io.quarkus.scheduler.runtime.SimpleScheduler; +import io.smallrye.common.annotation.Identifier; public class SchedulerProcessor { @@ -97,9 +103,58 @@ public class SchedulerProcessor { static final String NESTED_SEPARATOR = "$_"; @BuildStep - void beans(Capabilities capabilities, BuildProducer additionalBeans) { + SchedulerImplementationBuildItem implementation() { + return new SchedulerImplementationBuildItem(Scheduled.SIMPLE, DotName.createSimple(SimpleScheduler.class), 0); + } + + @BuildStep + void compositeScheduler(SchedulerConfig config, List implementations, + BuildProducer additionalBeans, + BuildProducer discoveredImplementations) { + List sorted = implementations.stream() + .sorted(Comparator.comparingInt(SchedulerImplementationBuildItem::getPriority).reversed()).toList(); + Set found = sorted.stream().map(SchedulerImplementationBuildItem::getImplementation) + .collect(Collectors.toUnmodifiableSet()); + if (found.size() != implementations.size()) { + throw new IllegalStateException("Invalid scheduler implementations detected: " + implementations); + } + DiscoveredImplementationsBuildItem discovered = new DiscoveredImplementationsBuildItem( + sorted.get(0).getImplementation(), found, + config.useCompositeScheduler); + discoveredImplementations.produce(discovered); + if (implementations.size() > 1 && config.useCompositeScheduler) { + // If multiple implementations are needed we have to register the CompositeScheduler, and + // instruct the extensions that provide an implementation to modify the bean metadata, i.e. add the marker qualifier + additionalBeans.produce(AdditionalBeanBuildItem.builder() + .addBeanClasses(Constituent.class, CompositeScheduler.class).setUnremovable().build()); + } + } + + @BuildStep + void transformSchedulerBeans(DiscoveredImplementationsBuildItem discoveredImplementations, + List implementations, + BuildProducer transformer) { + if (discoveredImplementations.isCompositeSchedulerUsed()) { + Map implsToBeanClass = implementations.stream() + .collect(Collectors.toMap(SchedulerImplementationBuildItem::getSchedulerBeanClass, + SchedulerImplementationBuildItem::getImplementation)); + transformer.produce(new AnnotationsTransformerBuildItem(AnnotationTransformation.forClasses() + .whenClass(c -> implsToBeanClass.containsKey(c.name())) + .transform(c -> { + c.add(AnnotationInstance.builder(Constituent.class).build()); + c.add(AnnotationInstance.builder(Identifier.class) + .add("value", implsToBeanClass.get(c.declaration().asClass().name())).build()); + }))); + } + } + + @BuildStep + void beans(DiscoveredImplementationsBuildItem discoveredImplementations, + BuildProducer additionalBeans) { additionalBeans.produce(new AdditionalBeanBuildItem(Scheduled.ApplicationNotRunning.class)); - if (capabilities.isMissing(Capability.QUARTZ)) { + if (discoveredImplementations.getImplementations().size() == 1 + || discoveredImplementations.isCompositeSchedulerUsed()) { + // Quartz extension is not present or composite scheduler is used additionalBeans.produce(new AdditionalBeanBuildItem(SimpleScheduler.class)); } } @@ -195,7 +250,8 @@ private void collectScheduledMethods(IndexView index, TransformedAnnotationsBuil @BuildStep void validateScheduledBusinessMethods(SchedulerConfig config, List scheduledMethods, ValidationPhaseBuildItem validationPhase, BuildProducer validationErrors, - Capabilities capabilities, BeanArchiveIndexBuildItem beanArchiveIndex) { + Capabilities capabilities, BeanArchiveIndexBuildItem beanArchiveIndex, + DiscoveredImplementationsBuildItem discoveredImplementations) { List errors = new ArrayList<>(); Map encounteredIdentities = new HashMap<>(); Set methodDescriptions = new HashSet<>(); @@ -252,7 +308,7 @@ void validateScheduledBusinessMethods(SchedulerConfig config, List unremovableBeans() { public FeatureBuildItem build(SchedulerConfig config, BuildProducer syntheticBeans, SchedulerRecorder recorder, List scheduledMethods, BuildProducer generatedClasses, BuildProducer reflectiveClass, - AnnotationProxyBuildItem annotationProxy, List schedulerForcedStartItems) { + AnnotationProxyBuildItem annotationProxy, List schedulerForcedStartItems, + DiscoveredImplementationsBuildItem discoveredImplementations) { List scheduledMetadata = new ArrayList<>(); ClassOutput classOutput = new GeneratedClassGizmoAdaptor(generatedClasses, new Function() { @@ -330,7 +387,8 @@ public String apply(String name) { } syntheticBeans.produce(SyntheticBeanBuildItem.configure(SchedulerContext.class).setRuntimeInit() - .supplier(recorder.createContext(config, scheduledMetadata, !schedulerForcedStartItems.isEmpty())) + .supplier(recorder.createContext(config, scheduledMetadata, !schedulerForcedStartItems.isEmpty(), + discoveredImplementations.getAutoImplementation())) .done()); return new FeatureBuildItem(Feature.SCHEDULER); @@ -530,8 +588,8 @@ private String generateInvoker(ScheduledBusinessMethodItem scheduledMethod, Clas } private Throwable validateScheduled(CronParser parser, AnnotationInstance schedule, - Map encounteredIdentities, - BeanDeploymentValidator.ValidationContext validationContext, long checkPeriod, IndexView index) { + Map encounteredIdentities, BeanDeploymentValidator.ValidationContext validationContext, + long checkPeriod, IndexView index, DiscoveredImplementationsBuildItem discoveredImplementations) { MethodInfo method = schedule.target().asMethod(); AnnotationValue cronValue = schedule.value("cron"); AnnotationValue everyValue = schedule.value("every"); @@ -645,6 +703,22 @@ private Throwable validateScheduled(CronParser parser, AnnotationInstance schedu } } } + + AnnotationValue executeWithValue = schedule.value("executeWith"); + if (executeWithValue != null) { + String implementation = executeWithValue.asString(); + if (!Scheduled.AUTO.equals(implementation)) { + if (!discoveredImplementations.getImplementations().contains(implementation)) { + return new IllegalStateException( + "The required scheduler implementation was not discovered in application: " + implementation); + } else if (!discoveredImplementations.isCompositeSchedulerUsed() + && !discoveredImplementations.isAutoImplementation(implementation)) { + return new IllegalStateException( + "The required scheduler implementation is not available because the composite scheduler is not used: " + + implementation); + } + } + } return null; } diff --git a/extensions/scheduler/deployment/src/test/java/io/quarkus/scheduler/test/composite/SchedulerImplementationNotDiscoveredTest.java b/extensions/scheduler/deployment/src/test/java/io/quarkus/scheduler/test/composite/SchedulerImplementationNotDiscoveredTest.java new file mode 100644 index 00000000000000..045bfc8553cb44 --- /dev/null +++ b/extensions/scheduler/deployment/src/test/java/io/quarkus/scheduler/test/composite/SchedulerImplementationNotDiscoveredTest.java @@ -0,0 +1,36 @@ +package io.quarkus.scheduler.test.composite; + +import static org.assertj.core.api.Assertions.assertThat; +import static org.junit.jupiter.api.Assertions.fail; + +import org.junit.jupiter.api.Test; +import org.junit.jupiter.api.extension.RegisterExtension; + +import io.quarkus.scheduler.Scheduled; +import io.quarkus.test.QuarkusUnitTest; + +public class SchedulerImplementationNotDiscoveredTest { + + @RegisterExtension + static final QuarkusUnitTest test = new QuarkusUnitTest() + .withApplicationRoot(root -> root + .addClasses(Jobs.class)) + .assertException(t -> { + assertThat(t).cause().isInstanceOf(IllegalStateException.class) + .hasMessageContaining( + "The required scheduler implementation was not discovered in application: QUARTZ"); + }); + + @Test + public void test() { + fail(); + } + + static class Jobs { + + @Scheduled(every = "1s", executeWith = Scheduled.QUARTZ) + void quartz() { + } + + } +} diff --git a/extensions/scheduler/runtime/src/main/java/io/quarkus/scheduler/runtime/CompositeScheduler.java b/extensions/scheduler/runtime/src/main/java/io/quarkus/scheduler/runtime/CompositeScheduler.java new file mode 100644 index 00000000000000..52beeeb54ae5ee --- /dev/null +++ b/extensions/scheduler/runtime/src/main/java/io/quarkus/scheduler/runtime/CompositeScheduler.java @@ -0,0 +1,175 @@ +package io.quarkus.scheduler.runtime; + +import java.util.ArrayList; +import java.util.List; + +import jakarta.enterprise.inject.Typed; +import jakarta.inject.Singleton; + +import io.quarkus.arc.All; +import io.quarkus.scheduler.Scheduled; +import io.quarkus.scheduler.Scheduler; +import io.quarkus.scheduler.Trigger; +import io.quarkus.scheduler.common.runtime.AbstractJobDefinition; + +/** + * The composite scheduler is only used in case of multiple {@link Scheduler} implementations are required. + * + * @see Scheduled#executeWith() + */ +@Typed(Scheduler.class) +@Singleton +public class CompositeScheduler implements Scheduler { + + private final List schedulers; + + CompositeScheduler(@All @Constituent List schedulers) { + this.schedulers = schedulers; + } + + @Override + public void pause() { + for (Scheduler scheduler : schedulers) { + scheduler.pause(); + } + } + + @Override + public void pause(String identity) { + for (Scheduler scheduler : schedulers) { + scheduler.pause(identity); + } + } + + @Override + public void resume() { + for (Scheduler scheduler : schedulers) { + scheduler.resume(); + } + } + + @Override + public void resume(String identity) { + for (Scheduler scheduler : schedulers) { + scheduler.resume(identity); + } + } + + @Override + public boolean isPaused(String identity) { + for (Scheduler scheduler : schedulers) { + if (scheduler.isPaused(identity)) { + return true; + } + } + return false; + } + + @Override + public boolean isRunning() { + // IMPL NOTE: we return true if at least one of the schedulers is running + for (Scheduler scheduler : schedulers) { + if (scheduler.isRunning()) { + return true; + } + } + return false; + } + + @Override + public List getScheduledJobs() { + List triggers = new ArrayList<>(); + for (Scheduler scheduler : schedulers) { + triggers.addAll(scheduler.getScheduledJobs()); + } + return triggers; + } + + @Override + public Trigger getScheduledJob(String identity) { + for (Scheduler scheduler : schedulers) { + Trigger trigger = scheduler.getScheduledJob(identity); + if (trigger != null) { + return trigger; + } + } + return null; + } + + @Override + public JobDefinition newJob(String identity) { + return new CompositeJobDefinition(identity); + } + + @Override + public Trigger unscheduleJob(String identity) { + for (Scheduler scheduler : schedulers) { + Trigger trigger = scheduler.unscheduleJob(identity); + if (trigger != null) { + return trigger; + } + } + return null; + } + + @Override + public String implementation() { + return Scheduled.AUTO; + } + + class CompositeJobDefinition extends AbstractJobDefinition { + + public CompositeJobDefinition(String identity) { + super(identity); + } + + @Override + public Trigger schedule() { + for (Scheduler scheduler : schedulers) { + if (scheduler.implementation() == implementation) { + copy(scheduler.newJob(identity)).schedule(); + } + } + throw new IllegalStateException("Matching scheduler implementation not found: " + implementation); + } + + private JobDefinition copy(JobDefinition to) { + to.setCron(cron); + to.setInterval(every); + to.setDelayed(delayed); + to.setOverdueGracePeriod(overdueGracePeriod); + to.setConcurrentExecution(concurrentExecution); + to.setTimeZone(timeZone); + to.setExecuteWith(implementation); + if (skipPredicate != null) { + to.setSkipPredicate(skipPredicate); + } + if (skipPredicateClass != null) { + to.setSkipPredicate(skipPredicateClass); + } + if (task != null) { + if (runOnVirtualThread) { + to.setTask(task, runOnVirtualThread); + } else { + to.setTask(task); + } + } + if (taskClass != null) { + if (runOnVirtualThread) { + to.setTask(taskClass, runOnVirtualThread); + } else { + to.setTask(taskClass); + } + } + if (asyncTask != null) { + to.setAsyncTask(asyncTask); + } + if (asyncTaskClass != null) { + to.setAsyncTask(asyncTaskClass); + } + return to; + } + + } + +} diff --git a/extensions/scheduler/runtime/src/main/java/io/quarkus/scheduler/runtime/Constituent.java b/extensions/scheduler/runtime/src/main/java/io/quarkus/scheduler/runtime/Constituent.java new file mode 100644 index 00000000000000..4fbf7c8a8d5dd1 --- /dev/null +++ b/extensions/scheduler/runtime/src/main/java/io/quarkus/scheduler/runtime/Constituent.java @@ -0,0 +1,26 @@ +package io.quarkus.scheduler.runtime; + +import static java.lang.annotation.ElementType.FIELD; +import static java.lang.annotation.ElementType.PARAMETER; +import static java.lang.annotation.ElementType.TYPE; +import static java.lang.annotation.RetentionPolicy.RUNTIME; + +import java.lang.annotation.Documented; +import java.lang.annotation.Retention; +import java.lang.annotation.Target; + +import jakarta.inject.Qualifier; + +import io.quarkus.scheduler.Scheduler; + +/** + * This qualifier is used to mark a constituent of a composite {@link Scheduler}, i.e. to distinguish various scheduler + * implementations. + */ +@Qualifier +@Documented +@Retention(RUNTIME) +@Target({ TYPE, PARAMETER, FIELD }) +public @interface Constituent { + +} diff --git a/extensions/scheduler/runtime/src/main/java/io/quarkus/scheduler/runtime/SchedulerConfig.java b/extensions/scheduler/runtime/src/main/java/io/quarkus/scheduler/runtime/SchedulerConfig.java index 23ce44a235145a..970ae9923e34b2 100644 --- a/extensions/scheduler/runtime/src/main/java/io/quarkus/scheduler/runtime/SchedulerConfig.java +++ b/extensions/scheduler/runtime/src/main/java/io/quarkus/scheduler/runtime/SchedulerConfig.java @@ -6,6 +6,7 @@ import io.quarkus.runtime.annotations.ConfigPhase; import io.quarkus.runtime.annotations.ConfigRoot; import io.quarkus.scheduler.Scheduled; +import io.quarkus.scheduler.Scheduler; @ConfigRoot(phase = ConfigPhase.BUILD_AND_RUN_TIME_FIXED) public class SchedulerConfig { @@ -30,4 +31,15 @@ public class SchedulerConfig { */ @ConfigItem(name = "tracing.enabled") public boolean tracingEnabled; + + /** + * By default, only one {@link Scheduler} implementation is used. If set to {@code true} then a composite {@link Scheduler} + * that delegates to all running implementations is used. + *

+ * Scheduler implementations will be started depending on the value of {@code quarkus.scheduler.start-mode}, i.e. the + * scheduler is not started unless a relevant {@link io.quarkus.scheduler.Scheduled} business method is found. + */ + @ConfigItem(defaultValue = "false") + public boolean useCompositeScheduler; + } diff --git a/extensions/scheduler/runtime/src/main/java/io/quarkus/scheduler/runtime/SchedulerRecorder.java b/extensions/scheduler/runtime/src/main/java/io/quarkus/scheduler/runtime/SchedulerRecorder.java index d0cc4037c9bc17..cb7cf17e329263 100644 --- a/extensions/scheduler/runtime/src/main/java/io/quarkus/scheduler/runtime/SchedulerRecorder.java +++ b/extensions/scheduler/runtime/src/main/java/io/quarkus/scheduler/runtime/SchedulerRecorder.java @@ -7,6 +7,7 @@ import com.cronutils.model.CronType; import io.quarkus.runtime.annotations.Recorder; +import io.quarkus.scheduler.Scheduled; import io.quarkus.scheduler.common.runtime.ImmutableScheduledMethod; import io.quarkus.scheduler.common.runtime.MutableScheduledMethod; import io.quarkus.scheduler.common.runtime.ScheduledMethod; @@ -16,7 +17,7 @@ public class SchedulerRecorder { public Supplier createContext(SchedulerConfig config, - List scheduledMethods, boolean forceSchedulerStart) { + List scheduledMethods, boolean forceSchedulerStart, String autoImplementation) { // Defensive design - make an immutable copy of the scheduled method metadata List metadata = immutableCopy(scheduledMethods); return new Supplier() { @@ -38,6 +39,26 @@ public List getScheduledMethods() { public boolean forceSchedulerStart() { return forceSchedulerStart; } + + @Override + public List getScheduledMethods(String implementation) { + List ret = new ArrayList<>(metadata.size()); + for (ScheduledMethod method : metadata) { + for (Scheduled scheduled : method.getSchedules()) { + if (matchesImplementation(scheduled, implementation)) { + ret.add(method); + } + } + } + return ret; + } + + @Override + public boolean matchesImplementation(Scheduled scheduled, String implementation) { + return scheduled.executeWith().equals(implementation) || ((autoImplementation.equals(implementation)) + && scheduled.executeWith().equals(Scheduled.AUTO)); + } + }; } }; diff --git a/extensions/scheduler/runtime/src/main/java/io/quarkus/scheduler/runtime/SimpleScheduler.java b/extensions/scheduler/runtime/src/main/java/io/quarkus/scheduler/runtime/SimpleScheduler.java index 8a3d77797146e8..0a176bcaa2610b 100644 --- a/extensions/scheduler/runtime/src/main/java/io/quarkus/scheduler/runtime/SimpleScheduler.java +++ b/extensions/scheduler/runtime/src/main/java/io/quarkus/scheduler/runtime/SimpleScheduler.java @@ -131,7 +131,8 @@ public SimpleScheduler(SchedulerContext context, SchedulerRuntimeConfig schedule } StartMode startMode = schedulerRuntimeConfig.startMode.orElse(StartMode.NORMAL); - if (startMode == StartMode.NORMAL && context.getScheduledMethods().isEmpty() && !context.forceSchedulerStart()) { + if (startMode == StartMode.NORMAL && context.getScheduledMethods(Scheduled.SIMPLE).isEmpty() + && !context.forceSchedulerStart()) { this.scheduledExecutor = null; LOG.info("No scheduled business methods found - Simple scheduler will not be started"); return; @@ -168,9 +169,12 @@ public void run() { } // Create triggers and invokers for @Scheduled methods - for (ScheduledMethod method : context.getScheduledMethods()) { + for (ScheduledMethod method : context.getScheduledMethods(Scheduled.SIMPLE)) { int nameSequence = 0; for (Scheduled scheduled : method.getSchedules()) { + if (!context.matchesImplementation(scheduled, Scheduled.SIMPLE)) { + continue; + } nameSequence++; String id = SchedulerUtils.lookUpPropertyValue(scheduled.identity()); if (id.isEmpty()) { @@ -192,6 +196,11 @@ public void run() { } } + @Override + public String implementation() { + return Scheduled.SIMPLE; + } + @Override public JobDefinition newJob(String identity) { Objects.requireNonNull(identity); @@ -724,7 +733,7 @@ public boolean isBlocking() { }; } Scheduled scheduled = new SyntheticScheduled(identity, cron, every, 0, TimeUnit.MINUTES, delayed, - overdueGracePeriod, concurrentExecution, skipPredicate, timeZone); + overdueGracePeriod, concurrentExecution, skipPredicate, timeZone, implementation); Optional trigger = createTrigger(identity, null, cronParser, scheduled, defaultOverdueGracePeriod); if (trigger.isPresent()) {