CAMEL-24079: Fix QuartzScheduledPollConsumerScheduler ignoring startScheduler=false

Move quartzScheduler.scheduleJob()/rescheduleJob() from doStart() to
startScheduler() so the startScheduler=false flag is respected. Same
structural fix as CAMEL-24066 for the Spring scheduler. doStart() now
only prepares resources; startScheduler() begins actual scheduling.

Closes #24715

Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>

Author: davsclaus

components/camel-quartz/src/main/java/org/apache/camel/pollconsumer/quartz/QuartzScheduledPollConsumerScheduler.java

@@ -71,6 +71,8 @@ public class QuartzScheduledPollConsumerScheduler extends ServiceSupport
     private boolean deleteJob = true;
     private volatile CronTrigger trigger;
     private volatile JobDetail job;
+    private volatile boolean scheduled;
+    private volatile TriggerKey rescheduleKey;
 
     @Override
     public void onInit(Consumer consumer) {
@@ -100,16 +102,43 @@ public void unscheduleTask() {
 
     @Override
     public void startScheduler() {
-        // the quartz component starts the scheduler
+        if (!scheduled) {
+            try {
+                if (rescheduleKey != null) {
+                    LOG.debug("Re-scheduling job: {} with trigger: {}", job, trigger.getKey());
+                    quartzScheduler.rescheduleJob(rescheduleKey, trigger);
+                } else {
+                    LOG.debug("Scheduling job: {} with trigger: {}", job, trigger.getKey());
+                    try {
+                        quartzScheduler.scheduleJob(job, trigger);
+                    } catch (ObjectAlreadyExistsException ex) {
+                        QuartzComponent quartz = getCamelContext().getComponent("quartz", QuartzComponent.class);
+                        if (!(quartz.isClustered())) {
+                            throw ex;
+                        } else {
+                            TriggerKey triggerKey = trigger.getKey();
+                            trigger = (CronTrigger) quartzScheduler.getTrigger(triggerKey);
+                            if (trigger == null) {
+                                throw new SchedulerException("Trigger could not be found in quartz scheduler.");
+                            }
+                        }
+                    }
+                }
+                scheduled = true;
+                if (LOG.isInfoEnabled()) {
+                    LOG.info("Job {} (triggerType={}, jobClass={}) is scheduled. Next fire date is {}",
+                            trigger.getKey(), trigger.getClass().getSimpleName(),
+                            job.getJobClass().getSimpleName(), trigger.getNextFireTime());
+                }
+            } catch (SchedulerException e) {
+                throw RuntimeCamelException.wrapRuntimeCamelException(e);
+            }
+        }
     }
 
     @Override
     public boolean isSchedulerStarted() {
-        try {
-            return quartzScheduler != null && quartzScheduler.isStarted();
-        } catch (SchedulerException e) {
-            return false;
-        }
+        return scheduled;
     }
 
     @Override
@@ -244,9 +273,6 @@ protected void doStart() throws Exception {
                 LOG.debug("Setting user extra triggerParameters {}", copy);
                 PropertyBindingSupport.bindProperties(camelContext, trigger, copy);
             }
-
-            LOG.debug("Scheduling job: {} with trigger: {}", job, trigger.getKey());
-            quartzScheduler.scheduleJob(job, trigger);
         } else {
             checkTriggerIsNonConflicting(existingTrigger);
 
@@ -264,36 +290,10 @@ protected void doStart() throws Exception {
                     .withSchedule(CronScheduleBuilder.cronSchedule(getCron()).inTimeZone(getTimeZone()))
                     .build();
 
-            // Reschedule job if trigger settings were changed
             if (hasTriggerChanged(existingTrigger, trigger)) {
-                LOG.debug("Re-scheduling job: {} with trigger: {}", job, trigger.getKey());
-                quartzScheduler.rescheduleJob(triggerKey, trigger);
-            } else {
-                // Schedule it now. Remember that scheduler might not be started it, but we can schedule now.
-                LOG.debug("Scheduling job: {} with trigger: {}", job, trigger.getKey());
-                try {
-                    // Schedule it now. Remember that scheduler might not be started it, but we can schedule now.
-                    quartzScheduler.scheduleJob(job, trigger);
-                } catch (ObjectAlreadyExistsException ex) {
-                    // some other VM might may have stored the job & trigger in DB in clustered mode, in the mean time
-                    QuartzComponent quartz = getCamelContext().getComponent("quartz", QuartzComponent.class);
-                    if (!(quartz.isClustered())) {
-                        throw ex;
-                    } else {
-                        trigger = (CronTrigger) quartzScheduler.getTrigger(triggerKey);
-                        if (trigger == null) {
-                            throw new SchedulerException("Trigger could not be found in quartz scheduler.");
-                        }
-                    }
-                }
+                rescheduleKey = triggerKey;
             }
         }
-
-        if (LOG.isInfoEnabled()) {
-            LOG.info("Job {} (triggerType={}, jobClass={}) is scheduled. Next fire date is {}",
-                    trigger.getKey(), trigger.getClass().getSimpleName(),
-                    job.getJobClass().getSimpleName(), trigger.getNextFireTime());
-        }
     }
 
     @Override
@@ -313,6 +313,8 @@ private void unscheduleJob() throws SchedulerException {
                 quartzScheduler.unscheduleJob(trigger.getKey());
             }
         }
+        scheduled = false;
+        rescheduleKey = null;
     }
 
     private void checkTriggerIsNonConflicting(Trigger trigger) {

components/camel-quartz/src/test/java/org/apache/camel/pollconsumer/quartz/FileConsumerQuartzSchedulerStartSchedulerTest.java

@@ -0,0 +1,56 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *      http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.camel.pollconsumer.quartz;
+
+import java.nio.file.Path;
+import java.util.concurrent.TimeUnit;
+
+import org.apache.camel.Exchange;
+import org.apache.camel.builder.RouteBuilder;
+import org.apache.camel.test.junit6.CamelTestSupport;
+import org.apache.camel.test.junit6.TestSupport;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.io.TempDir;
+
+import static org.awaitility.Awaitility.await;
+
+public class FileConsumerQuartzSchedulerStartSchedulerTest extends CamelTestSupport {
+    @TempDir
+    Path testDirectory;
+
+    @Test
+    public void testStartSchedulerFalseMustNotPoll() throws Exception {
+        template.sendBodyAndHeader(TestSupport.fileUri(testDirectory), "Hello World", Exchange.FILE_NAME, "hello.txt");
+
+        getMockEndpoint("mock:result").expectedMessageCount(0);
+
+        // startScheduler=false should prevent polling even with scheduler=quartz
+        await().during(3, TimeUnit.SECONDS).atMost(4, TimeUnit.SECONDS)
+                .untilAsserted(() -> getMockEndpoint("mock:result").assertIsSatisfied());
+    }
+
+    @Override
+    protected RouteBuilder createRouteBuilder() {
+        return new RouteBuilder() {
+            @Override
+            public void configure() {
+                from(TestSupport.fileUri(testDirectory, "?scheduler=quartz&scheduler.cron=0/2+*+*+*+*+?&startScheduler=false"))
+                        .to("mock:result");
+            }
+        };
+    }
+}