Adding the ability to prioritize tasks

Author: simeq

README.md

@@ -252,6 +252,25 @@ in queue (applicable if `upperLimitFractionOfThreads > 1.0`). If they stay there
 unlocked again (determined by `DeadExecutionHandler`).  Currently supported by **postgres**. **sql-server** also supports
 this, but testing has shown this is prone to deadlocks and thus not recommended until understood/resolved.
 
+### Prioritization
+
+If you have a mix of tasks with different priorities, you might want to enable prioritization.
+
+:gear: `.enablePrioritization()`<br/>
+An executor will always run the tasks with the highest priority first even if the `execution_time` is greater than other tasks. <br />
+You can set the priority of a task using the `setPriority(int)` method on the task instance, by default the priority of task is set to `0`.
+
+```java
+scheduler.schedule(
+  onetimeTask.instanceBuilder("1").setPriority(100),
+  Instant.now()
+);
+
+scheduler.schedule(
+  onetimeTask.instanceBuilder("2").setPriority(200),
+  Instant.now()
+);
+```
 
 #### Less commonly tuned
 
@@ -406,6 +425,7 @@ db-scheduler.table-name=scheduled_tasks
 db-scheduler.immediate-execution-enabled=false
 db-scheduler.scheduler-name=
 db-scheduler.threads=10
+db-scheduler.prioritization-enabled=false
 
 # Ignored if a custom DbSchedulerStarter bean is defined
 db-scheduler.delay-startup-until-context-ready=false
@@ -577,6 +597,9 @@ There are a number of users that are using db-scheduler for high throughput use-
 
 See [releases](https://github.com/kagkarlsson/db-scheduler/releases) for release-notes.
 
+**Upgrading to 15.x**
+* Add column `priority` and `priority_execution_time_idx` index to the database schema. See table definitions for [postgresql](./b-scheduler/src/test/resources/postgresql_tables.sql), [oracle](./db-scheduler/src/test/resources/oracle_tables.sql) or [mysql](./db-scheduler/src/test/resources/mysql_tables.sql). Note that when `enablePrioritization()` is used, the `null` value in order of prioritization is handled differently depending on the database used.
+
 **Upgrading to 8.x**
 * Custom Schedules must implement a method `boolean isDeterministic()` to indicate whether they will always produce the same instants or not.
 

db-scheduler-boot-starter/src/main/java/com/github/kagkarlsson/scheduler/boot/autoconfigure/DbSchedulerAutoConfiguration.java

@@ -164,6 +164,10 @@ public Scheduler scheduler(DbSchedulerCustomizer customizer, StatsRegistry regis
       builder.enableImmediateExecution();
     }
 
+    if (config.isPrioritizationEnabled()) {
+      builder.enablePrioritization();
+    }
+
     // Use custom executor service if provided
     customizer.executorService().ifPresent(builder::executorService);
 

db-scheduler-boot-starter/src/main/java/com/github/kagkarlsson/scheduler/boot/config/DbSchedulerProperties.java

@@ -114,6 +114,9 @@ public class DbSchedulerProperties {
   /** Whether or not to log the {@link Throwable} that caused a task to fail. */
   private boolean failureLoggerLogStackTrace = SchedulerBuilder.LOG_STACK_TRACE_ON_FAILURE;
 
+  /** Whether or not to prioritization of tasks is enabled. */
+  private boolean prioritizationEnabled = false;
+
   public boolean isEnabled() {
     return enabled;
   }
@@ -243,4 +246,12 @@ public boolean isAlwaysPersistTimestampInUtc() {
   public void setAlwaysPersistTimestampInUtc(boolean alwaysPersistTimestampInUTC) {
     this.alwaysPersistTimestampInUtc = alwaysPersistTimestampInUTC;
   }
+
+  public boolean isPrioritizationEnabled() {
+    return prioritizationEnabled;
+  }
+
+  public void setPrioritizationEnabled(boolean prioritizationEnabled) {
+    this.prioritizationEnabled = prioritizationEnabled;
+  }
 }

db-scheduler-boot-starter/src/test/java/com/github/kagkarlsson/scheduler/boot/autoconfigure/DbSchedulerAutoConfigurationTest.java

@@ -182,6 +182,20 @@ public void it_should_start_when_the_context_is_ready() {
             });
   }
 
+  @Test
+  public void it_should_enable_prioritization() {
+    ctxRunner
+        .withPropertyValues("db-scheduler.prioritization-enabled=true")
+        .run(
+            (AssertableApplicationContext ctx) -> {
+              assertThat(ctx).hasSingleBean(DataSource.class);
+              assertThat(ctx).hasSingleBean(Scheduler.class);
+
+              DbSchedulerProperties props = ctx.getBean(DbSchedulerProperties.class);
+              assertThat(props.isPrioritizationEnabled()).isTrue();
+            });
+  }
+
   @Test
   public void it_should_support_custom_starting_strategies() {
     ctxRunner

db-scheduler-boot-starter/src/test/resources/schema.sql

@@ -10,5 +10,6 @@ create table if not exists scheduled_tasks (
 	consecutive_failures INT,
 	last_heartbeat TIMESTAMP WITH TIME ZONE,
 	version BIGINT,
+	priority INT,
 	PRIMARY KEY (task_name, task_instance)
 );

db-scheduler/src/main/java/com/github/kagkarlsson/scheduler/FetchCandidates.java

@@ -44,6 +44,7 @@ public class FetchCandidates implements PollStrategy {
   AtomicInteger currentGenerationNumber = new AtomicInteger(0);
   private final int lowerLimit;
   private final int upperLimit;
+  private final boolean prioritization;
 
   public FetchCandidates(
       Executor executor,
@@ -58,7 +59,8 @@ public FetchCandidates(
       Clock clock,
       PollingStrategyConfig pollingStrategyConfig,
       Runnable triggerCheckForNewExecutions,
-      HeartbeatConfig heartbeatConfig) {
+      HeartbeatConfig heartbeatConfig,
+      boolean prioritization) {
     this.executor = executor;
     this.taskRepository = taskRepository;
     this.schedulerClient = schedulerClient;
@@ -71,6 +73,7 @@ public FetchCandidates(
     this.pollingStrategyConfig = pollingStrategyConfig;
     this.triggerCheckForNewExecutions = triggerCheckForNewExecutions;
     this.heartbeatConfig = heartbeatConfig;
+    this.prioritization = prioritization;
     lowerLimit = pollingStrategyConfig.getLowerLimit(threadpoolSize);
     // FIXLATER: this is not "upper limit", but rather nr of executions to get. those already in
     // queue will become stale
@@ -84,7 +87,8 @@ public void run() {
     // Fetch new candidates for execution. Old ones still in ExecutorService will become stale and
     // be discarded
     final int executionsToFetch = upperLimit;
-    List<Execution> fetchedDueExecutions = taskRepository.getDue(now, executionsToFetch);
+    List<Execution> fetchedDueExecutions =
+        taskRepository.getDue(now, executionsToFetch, prioritization);
     LOG.trace(
         "Fetched {} task instances due for execution at {}", fetchedDueExecutions.size(), now);
 

db-scheduler/src/main/java/com/github/kagkarlsson/scheduler/LockAndFetchCandidates.java

@@ -41,6 +41,7 @@ public class LockAndFetchCandidates implements PollStrategy {
   private final int lowerLimit;
   private final int upperLimit;
   private AtomicBoolean moreExecutionsInDatabase = new AtomicBoolean(false);
+  private final boolean prioritization;
 
   public LockAndFetchCandidates(
       Executor executor,
@@ -55,7 +56,8 @@ public LockAndFetchCandidates(
       Clock clock,
       PollingStrategyConfig pollingStrategyConfig,
       Runnable triggerCheckForNewExecutions,
-      HeartbeatConfig maxAgeBeforeConsideredDead) {
+      HeartbeatConfig maxAgeBeforeConsideredDead,
+      boolean prioritization) {
     this.executor = executor;
     this.taskRepository = taskRepository;
     this.schedulerClient = schedulerClient;
@@ -68,6 +70,7 @@ public LockAndFetchCandidates(
     this.pollingStrategyConfig = pollingStrategyConfig;
     this.triggerCheckForNewExecutions = triggerCheckForNewExecutions;
     this.maxAgeBeforeConsideredDead = maxAgeBeforeConsideredDead;
+    this.prioritization = prioritization;
     lowerLimit = pollingStrategyConfig.getLowerLimit(threadpoolSize);
     upperLimit = pollingStrategyConfig.getUpperLimit(threadpoolSize);
   }
@@ -85,7 +88,8 @@ public void run() {
     }
 
     // FIXLATER: should it fetch here if not under lowerLimit? probably
-    List<Execution> pickedExecutions = taskRepository.lockAndGetDue(now, executionsToFetch);
+    List<Execution> pickedExecutions =
+        taskRepository.lockAndGetDue(now, executionsToFetch, prioritization);
     LOG.trace("Picked {} taskinstances due for execution", pickedExecutions.size());
 
     // Shared indicator for if there are more due executions in the database.

db-scheduler/src/main/java/com/github/kagkarlsson/scheduler/Scheduler.java

@@ -69,6 +69,7 @@ public class Scheduler implements SchedulerClient {
   private final Waiter heartbeatWaiter;
   final SettableSchedulerState schedulerState = new SettableSchedulerState();
   final ConfigurableLogger failureLogger;
+  final boolean prioritization;
 
   protected Scheduler(
       Clock clock,
@@ -90,7 +91,8 @@ protected Scheduler(
       boolean logStackTrace,
       List<OnStartup> onStartup,
       ExecutorService dueExecutor,
-      ScheduledExecutorService housekeeperExecutor) {
+      ScheduledExecutorService housekeeperExecutor,
+      boolean prioritization) {
     this.clock = clock;
     this.schedulerTaskRepository = schedulerTaskRepository;
     this.taskResolver = taskResolver;
@@ -112,6 +114,7 @@ protected Scheduler(
     this.housekeeperExecutor = housekeeperExecutor;
     delegate = new StandardSchedulerClient(clientTaskRepository, this.schedulerListeners, clock);
     this.failureLogger = ConfigurableLogger.create(LOG, logLevel, logStackTrace);
+    this.prioritization = prioritization;
 
     if (pollingStrategyConfig.type == PollingStrategyConfig.Type.LOCK_AND_FETCH) {
       schedulerTaskRepository.verifySupportsLockAndFetch();
@@ -129,7 +132,8 @@ protected Scheduler(
               clock,
               pollingStrategyConfig,
               this::triggerCheckForDueExecutions,
-              heartbeatConfig);
+              heartbeatConfig,
+              prioritization);
     } else if (pollingStrategyConfig.type == PollingStrategyConfig.Type.FETCH) {
       executeDueStrategy =
           new FetchCandidates(
@@ -145,7 +149,8 @@ protected Scheduler(
               clock,
               pollingStrategyConfig,
               this::triggerCheckForDueExecutions,
-              heartbeatConfig);
+              heartbeatConfig,
+              prioritization);
     } else {
       throw new IllegalArgumentException(
           "Unknown polling-strategy type: " + pollingStrategyConfig.type);

db-scheduler/src/main/java/com/github/kagkarlsson/scheduler/SchedulerBuilder.java

@@ -74,6 +74,7 @@ public class SchedulerBuilder {
   protected PollingStrategyConfig pollingStrategyConfig = DEFAULT_POLLING_STRATEGY;
   protected LogLevel logLevel = DEFAULT_FAILURE_LOG_LEVEL;
   protected boolean logStackTrace = LOG_STACK_TRACE_ON_FAILURE;
+  protected boolean prioritization = false;
   private boolean registerShutdownHook = false;
   private int numberOfMissedHeartbeatsBeforeDead = DEFAULT_MISSED_HEARTBEATS_LIMIT;
   private boolean alwaysPersistTimestampInUTC = false;
@@ -230,6 +231,11 @@ public SchedulerBuilder registerShutdownHook() {
     return this;
   }
 
+  public SchedulerBuilder enablePrioritization() {
+    this.prioritization = true;
+    return this;
+  }
+
   public Scheduler build() {
     if (schedulerName == null) {
       schedulerName = new SchedulerName.Hostname();
@@ -249,6 +255,7 @@ public Scheduler build() {
             taskResolver,
             schedulerName,
             serializer,
+            prioritization,
             clock);
     final JdbcTaskRepository clientTaskRepository =
         new JdbcTaskRepository(
@@ -259,6 +266,7 @@ public Scheduler build() {
             taskResolver,
             schedulerName,
             serializer,
+            prioritization,
             clock);
 
     ExecutorService candidateExecutorService = executorService;
@@ -316,7 +324,8 @@ public Scheduler build() {
             logStackTrace,
             startTasks,
             candidateDueExecutor,
-            candidateHousekeeperExecutor);
+            candidateHousekeeperExecutor,
+            prioritization);
 
     if (enableImmediateExecution) {
       scheduler.registerSchedulerListener(new ImmediateCheckForDueExecutions(scheduler, clock));

db-scheduler/src/main/java/com/github/kagkarlsson/scheduler/SchedulerClient.java

@@ -213,6 +213,7 @@ class Builder {
     private Serializer serializer = Serializer.DEFAULT_JAVA_SERIALIZER;
     private String tableName = JdbcTaskRepository.DEFAULT_TABLE_NAME;
     private JdbcCustomization jdbcCustomization;
+    private boolean prioritization = false;
 
     private Builder(DataSource dataSource, List<Task<?>> knownTasks) {
       this.dataSource = dataSource;
@@ -242,6 +243,11 @@ public Builder jdbcCustomization(JdbcCustomization jdbcCustomization) {
       return this;
     }
 
+    public Builder enablePrioritization() {
+      this.prioritization = true;
+      return this;
+    }
+
     public SchedulerClient build() {
       TaskResolver taskResolver = new TaskResolver(StatsRegistry.NOOP, knownTasks);
       final SystemClock clock = new SystemClock();
@@ -259,6 +265,7 @@ public SchedulerClient build() {
               taskResolver,
               new SchedulerClientName(),
               serializer,
+              prioritization,
               clock);
 
       return new StandardSchedulerClient(taskRepository, clock);

db-scheduler/src/main/java/com/github/kagkarlsson/scheduler/TaskRepository.java

@@ -26,7 +26,7 @@ public interface TaskRepository {
 
   boolean createIfNotExists(SchedulableInstance execution);
 
-  List<Execution> getDue(Instant now, int limit);
+  List<Execution> getDue(Instant now, int limit, boolean prioritization);
 
   Instant replace(Execution toBeReplaced, SchedulableInstance newInstance);
 
@@ -37,7 +37,7 @@ void getScheduledExecutions(
 
   List<Execution> lockAndFetchGeneric(Instant now, int limit);
 
-  List<Execution> lockAndGetDue(Instant now, int limit);
+  List<Execution> lockAndGetDue(Instant now, int limit, boolean prioritization);
 
   void remove(Execution execution);
 

db-scheduler/src/main/java/com/github/kagkarlsson/scheduler/jdbc/AutodetectJdbcCustomization.java

@@ -125,15 +125,20 @@ public String getQueryLimitPart(int limit) {
     return jdbcCustomization.getQueryLimitPart(limit);
   }
 
+  @Override
+  public String getQueryOrderPart(boolean prioritization) {
+    return jdbcCustomization.getQueryOrderPart(prioritization);
+  }
+
   @Override
   public boolean supportsSingleStatementLockAndFetch() {
     return jdbcCustomization.supportsSingleStatementLockAndFetch();
   }
 
   @Override
   public List<Execution> lockAndFetchSingleStatement(
-      JdbcTaskRepositoryContext ctx, Instant now, int limit) {
-    return jdbcCustomization.lockAndFetchSingleStatement(ctx, now, limit);
+      JdbcTaskRepositoryContext ctx, Instant now, int limit, boolean prioritization) {
+    return jdbcCustomization.lockAndFetchSingleStatement(ctx, now, limit, prioritization);
   }
 
   @Override
@@ -143,14 +148,15 @@ public boolean supportsGenericLockAndFetch() {
 
   @Override
   public String createGenericSelectForUpdateQuery(
-      String tableName, int limit, String requiredAndCondition) {
+      String tableName, int limit, String requiredAndCondition, boolean prioritization) {
     return jdbcCustomization.createGenericSelectForUpdateQuery(
-        tableName, limit, requiredAndCondition);
+        tableName, limit, requiredAndCondition, prioritization);
   }
 
   @Override
-  public String createSelectDueQuery(String tableName, int limit, String andCondition) {
-    return jdbcCustomization.createSelectDueQuery(tableName, limit, andCondition);
+  public String createSelectDueQuery(
+      String tableName, int limit, String andCondition, boolean prioritization) {
+    return jdbcCustomization.createSelectDueQuery(tableName, limit, andCondition, prioritization);
   }
 
   @Override