PR: Introduce pluggable TimeBroker as replacement for system clock

quartz/src/main/java/org/quartz/core/QuartzScheduler.java

@@ -70,11 +70,7 @@
 import org.quartz.impl.matchers.GroupMatcher;
 import org.quartz.listeners.SchedulerListenerSupport;
 import org.quartz.simpl.PropertySettingJobFactory;
-import org.quartz.spi.JobFactory;
-import org.quartz.spi.OperableTrigger;
-import org.quartz.spi.SchedulerPlugin;
-import org.quartz.spi.SchedulerSignaler;
-import org.quartz.spi.ThreadExecutor;
+import org.quartz.spi.*;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
@@ -179,6 +175,8 @@ public class QuartzScheduler implements RemotableQuartzScheduler {
     private Date initialStart = null;
 
     private final Logger log = LoggerFactory.getLogger(getClass());
+
+    private final TimeBroker timeBroker;
 
     // private static final Map<String, ManagementServer> MGMT_SVR_BY_BIND = new
     // HashMap<String, ManagementServer>();
@@ -202,6 +200,8 @@ public class QuartzScheduler implements RemotableQuartzScheduler {
      */
     public QuartzScheduler(QuartzSchedulerResources resources, long idleWaitTime, @Deprecated long dbRetryInterval)
         throws SchedulerException {
+        this.timeBroker = resources.getTimeBroker();
+        this.timeBroker.initialize();
         this.resources = resources;
         if (resources.getJobStore() instanceof JobListener) {
             addInternalJobListener((JobListener)resources.getJobStore());
@@ -719,6 +719,12 @@ public void shutdown(boolean waitForJobsToComplete) {
 
         resources.getJobStore().shutdown();
 
+        try {
+            resources.getTimeBroker().shutdown();
+        } catch (Throwable t) {
+            getLog().warn("Error shutting down TimeBroker: {}", t.getMessage(), t);
+        }
+
         notifySchedulerListenersShutdown();
 
         SchedulerRepository.getInstance().remove(resources.getName());

quartz/src/main/java/org/quartz/core/QuartzSchedulerResources.java

@@ -23,10 +23,8 @@
 import java.util.List;
 
 import org.quartz.management.ManagementRESTServiceConfiguration;
-import org.quartz.spi.JobStore;
-import org.quartz.spi.SchedulerPlugin;
-import org.quartz.spi.ThreadExecutor;
-import org.quartz.spi.ThreadPool;
+import org.quartz.simpl.SimpleTimeBroker;
+import org.quartz.spi.*;
 
 /**
  * <p>
@@ -96,6 +94,8 @@ public class QuartzSchedulerResources {
 
     private boolean interruptJobsOnShutdown = false;
     private boolean interruptJobsOnShutdownWithWait = false;
+
+    private TimeBroker timeBroker = new SimpleTimeBroker();
 
     /*
      * ~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
@@ -583,4 +583,11 @@ public void setManagementRESTServiceConfiguration(ManagementRESTServiceConfigura
         this.managementRESTServiceConfiguration = managementRESTServiceConfiguration;
     }
 
+    public TimeBroker getTimeBroker() {
+        return timeBroker;
+    }
+
+    public void setTimeBroker(TimeBroker timeBroker) {
+        this.timeBroker = timeBroker;
+    }
 }

quartz/src/main/java/org/quartz/core/QuartzSchedulerThread.java

@@ -287,7 +287,7 @@ public void run() {
 
                     List<OperableTrigger> triggers;
 
-                    long now = System.currentTimeMillis();
+                    long now = currentTimeMillis();
 
                     clearSignaledSchedulingChange();
                     try {
@@ -316,7 +316,7 @@ public void run() {
 
                     if (triggers != null && !triggers.isEmpty()) {
 
-                        now = System.currentTimeMillis();
+                        now = currentTimeMillis();
                         long triggerTime = triggers.get(0).getNextFireTime().getTime();
                         long timeUntilTrigger = triggerTime - now;
                         while(timeUntilTrigger > 2) {
@@ -328,7 +328,7 @@ public void run() {
                                     try {
                                         // we could have blocked a long while
                                         // on 'synchronize', so we must recompute
-                                        now = System.currentTimeMillis();
+                                        now = currentTimeMillis();
                                         timeUntilTrigger = triggerTime - now;
                                         if(timeUntilTrigger >= 1)
                                             sigLock.wait(timeUntilTrigger);
@@ -344,7 +344,7 @@ public void run() {
                             if(releaseIfScheduleChangedSignificantly(triggers, triggerTime)) {
                                 break;
                             }
-                            now = System.currentTimeMillis();
+                            now = currentTimeMillis();
                             timeUntilTrigger = triggerTime - now;
                         }
 
@@ -424,7 +424,7 @@ public void run() {
                     continue; // while (!halted)
                 }
 
-                long now = System.currentTimeMillis();
+                long now = currentTimeMillis();
                 long waitTime = now + getRandomizedIdleWaitTime();
                 long timeUntilContinue = waitTime - now;
                 synchronized(sigLock) {
@@ -522,7 +522,7 @@ else if(getSignaledNextFireTime() < oldTime )
 
             if(earlier) {
                 // so the new time is considered earlier, but is it enough earlier?
-                long diff = oldTime - System.currentTimeMillis();
+                long diff = oldTime - currentTimeMillis();
                 if(diff < (qsRsrcs.getJobStore().supportsPersistence() ? 70L : 7L))
                     earlier = false;
             }
@@ -535,6 +535,19 @@ else if(getSignaledNextFireTime() < oldTime )
         }
     }
 
+    /**
+     * Get current time from the configured TimeBroker, falling back to
+     * System.currentTimeMillis() if it fails.
+     */
+    private long currentTimeMillis() {
+        try {
+            return qsRsrcs.getTimeBroker().getCurrentTime().getTime();
+        } catch (SchedulerException e) {
+            log.error("Error getting current time from TimeBroker, falling back to System.currentTimeMillis()", e);
+            return System.currentTimeMillis();
+        }
+    }
+
     public Logger getLog() {
         return log;
     }

quartz/src/main/java/org/quartz/impl/StdSchedulerFactory.java

@@ -37,13 +37,7 @@
 import org.quartz.management.ManagementRESTServiceConfiguration;
 import org.quartz.simpl.RAMJobStore;
 import org.quartz.simpl.SimpleThreadPool;
-import org.quartz.spi.ClassLoadHelper;
-import org.quartz.spi.InstanceIdGenerator;
-import org.quartz.spi.JobFactory;
-import org.quartz.spi.JobStore;
-import org.quartz.spi.SchedulerPlugin;
-import org.quartz.spi.ThreadExecutor;
-import org.quartz.spi.ThreadPool;
+import org.quartz.spi.*;
 import org.quartz.utils.ConnectionProvider;
 import org.quartz.utils.DBConnectionManager;
 import org.quartz.utils.JNDIConnectionProvider;
@@ -287,6 +281,10 @@ public class StdSchedulerFactory implements SchedulerFactory {
 
     public static final String MANAGEMENT_REST_SERVICE_HOST_PORT = "org.quartz.managementRESTService.bind";
 
+    public static final String TIMEBROKER_CLASS = "org.quartz.scheduler.timeBroker.class";
+
+    public static final String SIMPLE_TIMEBROKER_CLASS = "org.quartz.simpl.SimpleTimeBroker";
+
     /*
      * ~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
      *
@@ -1296,6 +1294,13 @@ else if (schedInstId.equals(SYSTEM_PROPERTY_AS_INSTANCE_ID)) {
             rsrcs.setInterruptJobsOnShutdownWithWait(interruptJobsOnShutdownWithWait);
             rsrcs.setJMXExport(jmxExport);
             rsrcs.setJMXObjectName(jmxObjectName);
+            String timeBrokerClassName = cfg.getStringProperty(
+                    TIMEBROKER_CLASS,
+                    SIMPLE_TIMEBROKER_CLASS);
+
+            TimeBroker timeBroker;
+            timeBroker = instantiateTimeBroker(loadHelper, timeBrokerClassName);
+            rsrcs.setTimeBroker(timeBroker);
 
             if (managementRESTServiceEnabled) {
                 ManagementRESTServiceConfiguration managementRESTServiceConfiguration = new ManagementRESTServiceConfiguration();
@@ -1393,6 +1398,21 @@ else if (schedInstId.equals(SYSTEM_PROPERTY_AS_INSTANCE_ID)) {
         }
     }
 
+    private TimeBroker instantiateTimeBroker(ClassLoadHelper loadHelper, String timeBrokerClassName) throws SchedulerException {
+        TimeBroker timeBroker;
+        try {
+            timeBroker = (TimeBroker) loadHelper
+                    .loadClass(timeBrokerClassName)
+                    .getDeclaredConstructor()
+                    .newInstance();
+        } catch (Exception e) {
+            initException = new SchedulerException(
+                    "TimeBroker class '" + timeBrokerClassName + "' could not be instantiated.", e);
+            throw initException;
+        }
+        return timeBroker;
+    }
+
     private void populateProviderWithExtraProps(PoolingConnectionProvider cp, Properties props) throws Exception {
         Properties copyProps = new Properties();
         copyProps.putAll(props);

quartz/src/main/java/org/quartz/simpl/SimpleTimeBroker.java

@@ -33,7 +33,7 @@
  * <p>
  * In general, the default implementation of this interface (<code>{@link org.quartz.simpl.SimpleTimeBroker}</code>-
  * which simply uses <code>System.getCurrentTimeMillis()</code> )is
- * sufficient. However situations may exist where this default scheme is
+ * sufficient. However, situations may exist where this default scheme is
  * lacking in its robustness - especially when Quartz is used in a clustered
  * configuration. For example, if one or more of the machines in the cluster
  * has a system time that varies by more than a few seconds from the clocks on
@@ -44,7 +44,6 @@
  * 
  * @author James House
  */
-@SuppressWarnings("deprecation")
 public class SimpleTimeBroker implements TimeBroker {
 
     /*

quartz/src/main/java/org/quartz/spi/TimeBroker.java

@@ -43,10 +43,8 @@
  * </p>
  * 
  * @see org.quartz.core.QuartzScheduler
- * @deprecated TimeBroker is not currently used in the Quartz code base.
  * @author James House
  */
-@Deprecated
 public interface TimeBroker {
 
     /*

quartz/src/test/java/org/quartz/core/FakeTimeBroker.java

@@ -0,0 +1,59 @@
+/*
+ * All content copyright Terracotta, Inc., unless otherwise indicated. All rights reserved.
+ * Copyright IBM Corp. 2024, 2025
+ *
+ * Licensed 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.quartz.core;
+
+import java.util.Date;
+import java.util.concurrent.atomic.AtomicLong;
+
+import org.quartz.SchedulerConfigException;
+import org.quartz.SchedulerException;
+import org.quartz.spi.TimeBroker;
+
+/*
+ * Test helper TimeBroker implementation used to verify that Quartz
+ * uses the configured time source instead of System.currentTimeMillis().
+ * author: Thanos Tsiamis https://github.com/ThanosTsiamis
+ */
+public class FakeTimeBroker implements TimeBroker {
+
+    private static final AtomicLong NOW = new AtomicLong();
+
+    public static void setNow(long millis) {
+        NOW.set(millis);
+    }
+
+    public static long currentTimeMillis() {
+        return NOW.get();
+    }
+
+    @Override
+    public Date getCurrentTime() throws SchedulerException {
+        return new Date(NOW.get());
+    }
+
+    @Override
+    public void initialize() throws SchedulerConfigException {
+        // no-op for tests
+    }
+
+    @Override
+    public void shutdown() {
+        // no-op for tests
+    }
+}