feat: Support for batch scheduling (#596)

## Brief, plain english overview of your changes here

This PR adds support for scheduling new tasks using batch inserts.

Known issues:
- Updates are not supported; if any of the tasks in a batch exists, the
batch will fail
- Hard-coded batch size of 100 when using `Stream` as parameter. Inserts
are committed after each batch.

## Fixes
#595 


## Reminders
- [x] Added/ran automated tests
- [x] Update README and/or examples
- [x] Ran `mvn spotless:apply`

---------

Co-authored-by: Gustav Karlsson <[email protected]>

.github/workflows/ci.yml

@@ -18,6 +18,6 @@ jobs:
         cache: 'maven'
 
     - name: Run all tests
-      run: mvn -B -Pall-tests,publication clean package --file pom.xml
+      run: mvn -B -Ptests-for-ci,publication clean package --file pom.xml
       env:
         TZ: UTC

README.md

@@ -153,6 +153,17 @@ scheduler.schedule(
         .scheduledTo(Instant.now().plusSeconds(5)));
 ```
 
+... or schedule in batches using:
+
+```java
+Stream<TaskInstance<?>> taskInstances = Stream.of(
+    MY_TASK.instance("my-task-1", 1),
+    MY_TASK.instance("my-task-2", 2),
+    MY_TASK.instance("my-task-3", 3));
+
+scheduler.scheduleBatch(taskInstances, Instant.now());
+```
+
 ### More examples
 
 #### Plain Java

db-scheduler/pom.xml

@@ -369,16 +369,10 @@
             </properties>
         </profile>
         <profile>
-            <id>compatibility</id>
+            <id>tests-for-ci</id>
             <properties>
                 <test.excludedTags>compatibility-cluster</test.excludedTags>
             </properties>
         </profile>
-        <profile>
-            <id>compatibility-cluster</id>
-            <properties>
-                <test.excludedTags>compatibility</test.excludedTags>
-            </properties>
-        </profile>
     </profiles>
 </project>

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

@@ -264,6 +264,16 @@ public <T> boolean scheduleIfNotExists(SchedulableInstance<T> schedulableInstanc
     return this.delegate.scheduleIfNotExists(schedulableInstance);
   }
 
+  @Override
+  public void scheduleBatch(List<TaskInstance<?>> taskInstances, Instant executionTime) {
+    this.delegate.scheduleBatch(taskInstances, executionTime);
+  }
+
+  @Override
+  public void scheduleBatch(List<SchedulableInstance<?>> schedulableInstances) {
+    this.delegate.scheduleBatch(schedulableInstances);
+  }
+
   @Override
   public <T> void schedule(TaskInstance<T> taskInstance, Instant executionTime) {
     this.delegate.schedule(taskInstance, executionTime);

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

@@ -14,6 +14,7 @@
 package com.github.kagkarlsson.scheduler;
 
 import static java.util.Optional.ofNullable;
+import static java.util.stream.Collectors.toList;
 
 import com.github.kagkarlsson.scheduler.SchedulerClient.ScheduleOptions.WhenExists;
 import com.github.kagkarlsson.scheduler.event.SchedulerListeners;
@@ -26,6 +27,7 @@
 import com.github.kagkarlsson.scheduler.stats.StatsRegistry;
 import com.github.kagkarlsson.scheduler.task.Execution;
 import com.github.kagkarlsson.scheduler.task.SchedulableInstance;
+import com.github.kagkarlsson.scheduler.task.ScheduledTaskInstance;
 import com.github.kagkarlsson.scheduler.task.Task;
 import com.github.kagkarlsson.scheduler.task.TaskInstance;
 import com.github.kagkarlsson.scheduler.task.TaskInstanceId;
@@ -35,12 +37,15 @@
 import java.util.List;
 import java.util.Optional;
 import java.util.function.Consumer;
+import java.util.stream.Stream;
 import javax.sql.DataSource;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
 public interface SchedulerClient {
 
+  int DEFAULT_BATCH_SIZE = 100;
+
   /**
    * Schedule a new execution for the given task instance.
    *
@@ -116,6 +121,51 @@ <T> boolean schedule(
    */
   <T> boolean scheduleIfNotExists(SchedulableInstance<T> schedulableInstance);
 
+  /**
+   * Schedule a batch of executions. If any of the executions already exists, the scheduling will
+   * fail and an exception will be thrown.
+   *
+   * @param taskInstances Task-instance to schedule, optionally with data
+   * @param executionTime Instant it should run
+   * @see java.time.Instant
+   * @see com.github.kagkarlsson.scheduler.task.TaskInstance
+   */
+  void scheduleBatch(List<TaskInstance<?>> taskInstances, Instant executionTime);
+
+  /**
+   * Schedule a batch of executions. If any of the executions already exists, the scheduling will
+   * fail and an exception will be thrown.
+   *
+   * @param schedulableInstances Task-instances with individual execution-times
+   * @see com.github.kagkarlsson.scheduler.task.SchedulableInstance
+   */
+  void scheduleBatch(List<SchedulableInstance<?>> schedulableInstances);
+
+  /**
+   * Schedule a batch of executions. If any of the executions already exists, the scheduling will
+   * fail and an exception will be thrown.
+   *
+   * @param taskInstances Task-instances to schedule, optionally with data
+   * @param executionTime Instant it should run
+   * @see java.time.Instant
+   * @see com.github.kagkarlsson.scheduler.task.TaskInstance
+   */
+  default void scheduleBatch(Stream<TaskInstance<?>> taskInstances, Instant executionTime) {
+    StreamUtils.chunkStream(taskInstances, DEFAULT_BATCH_SIZE)
+        .forEach(chunk -> scheduleBatch(chunk, executionTime));
+  }
+
+  /**
+   * Schedule a batch of executions. If any of the executions already exists, the scheduling will
+   * fail and an exception will be thrown.
+   *
+   * @param schedulableInstances Task-instances with individual execution-times
+   * @see com.github.kagkarlsson.scheduler.task.SchedulableInstance
+   */
+  default void scheduleBatch(Stream<SchedulableInstance<?>> schedulableInstances) {
+    StreamUtils.chunkStream(schedulableInstances, DEFAULT_BATCH_SIZE).forEach(this::scheduleBatch);
+  }
+
   /**
    * Update an existing execution to a new execution-time. If the execution does not exist or if it
    * is currently running, an exception is thrown.
@@ -389,6 +439,34 @@ public <T> boolean scheduleIfNotExists(SchedulableInstance<T> schedulableInstanc
           schedulableInstance.getNextExecutionTime(clock.now()));
     }
 
+    @Override
+    public void scheduleBatch(List<TaskInstance<?>> taskInstances, Instant executionTime) {
+      List<ScheduledTaskInstance> batchToSchedule =
+          taskInstances.stream()
+              .map(taskInstance -> new ScheduledTaskInstance(taskInstance, executionTime))
+              .collect(toList());
+
+      taskRepository.createBatch(batchToSchedule);
+      notifyListenersOfScheduledBatch(batchToSchedule);
+    }
+
+    @Override
+    public void scheduleBatch(List<SchedulableInstance<?>> schedulableInstances) {
+      List<ScheduledTaskInstance> batchToSchedule =
+          schedulableInstances.stream()
+              .map(schedulable -> ScheduledTaskInstance.fixExecutionTime(schedulable, clock))
+              .collect(toList());
+
+      taskRepository.createBatch(batchToSchedule);
+      notifyListenersOfScheduledBatch(batchToSchedule);
+    }
+
+    private void notifyListenersOfScheduledBatch(List<ScheduledTaskInstance> batchToSchedule) {
+      batchToSchedule.forEach(
+          instance ->
+              schedulerListeners.onExecutionScheduled(instance, instance.getExecutionTime()));
+    }
+
     @Override
     public <T> void schedule(SchedulableInstance<T> schedulableInstance) {
       schedule(

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

@@ -0,0 +1,39 @@
+/*
+ * Copyright (C) Gustav Karlsson
+ *
+ * <p>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
+ *
+ * <p>http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * <p>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 com.github.kagkarlsson.scheduler;
+
+import java.util.ArrayList;
+import java.util.Iterator;
+import java.util.List;
+import java.util.stream.Stream;
+
+public class StreamUtils {
+
+  public static <T> Stream<List<T>> chunkStream(Stream<T> stream, int chunkSize) {
+    if (chunkSize <= 0) {
+      throw new IllegalArgumentException("Chunk size must be greater than 0");
+    }
+
+    Iterator<T> iterator = stream.iterator();
+    return Stream.generate(
+            () -> {
+              List<T> chunk = new ArrayList<>();
+              for (int i = 0; i < chunkSize && iterator.hasNext(); i++) {
+                chunk.add(iterator.next());
+              }
+              return chunk;
+            })
+        .takeWhile(chunk -> !chunk.isEmpty());
+  }
+}

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

@@ -15,6 +15,7 @@
 
 import com.github.kagkarlsson.scheduler.task.Execution;
 import com.github.kagkarlsson.scheduler.task.SchedulableInstance;
+import com.github.kagkarlsson.scheduler.task.ScheduledTaskInstance;
 import com.github.kagkarlsson.scheduler.task.TaskInstanceId;
 import java.time.Duration;
 import java.time.Instant;
@@ -24,11 +25,21 @@
 
 public interface TaskRepository {
 
-  boolean createIfNotExists(SchedulableInstance execution);
+  /** Prefer ScheduledTaskInstance which has a fixed execution-time */
+  @Deprecated
+  boolean createIfNotExists(SchedulableInstance<?> execution);
+
+  boolean createIfNotExists(ScheduledTaskInstance execution);
 
   List<Execution> getDue(Instant now, int limit);
 
-  Instant replace(Execution toBeReplaced, SchedulableInstance newInstance);
+  void createBatch(List<ScheduledTaskInstance> executions);
+
+  Instant replace(Execution toBeReplaced, ScheduledTaskInstance newInstance);
+
+  /** Prefer ScheduledTaskInstance which has a fixed execution-time */
+  @Deprecated
+  Instant replace(Execution toBeReplaced, SchedulableInstance<?> newInstance);
 
   void getScheduledExecutions(ScheduledExecutionsFilter filter, Consumer<Execution> consumer);
 
@@ -71,7 +82,6 @@ boolean reschedule(
   default Optional<Execution> getExecution(TaskInstanceId taskInstance) {
     return getExecution(taskInstance.getTaskName(), taskInstance.getId());
   }
-  ;
 
   int removeExecutions(String taskName);
 

db-scheduler/src/main/java/com/github/kagkarlsson/scheduler/exceptions/DbSchedulerException.java

@@ -14,7 +14,7 @@
 package com.github.kagkarlsson.scheduler.exceptions;
 
 public abstract class DbSchedulerException extends RuntimeException {
-  static final long serialVersionUID = -2132850112553296790L;
+  private static final long serialVersionUID = -2132850112553296790L;
 
   public DbSchedulerException(String message) {
     super(message);

db-scheduler/src/main/java/com/github/kagkarlsson/scheduler/exceptions/FailedToScheduleBatchException.java

@@ -0,0 +1,26 @@
+/*
+ * Copyright (C) Gustav Karlsson
+ *
+ * <p>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
+ *
+ * <p>http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * <p>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 com.github.kagkarlsson.scheduler.exceptions;
+
+public class FailedToScheduleBatchException extends DbSchedulerException {
+  private static final long serialVersionUID = -2132850112553296792L;
+
+  public FailedToScheduleBatchException(String message) {
+    super(message);
+  }
+
+  public FailedToScheduleBatchException(String message, Throwable cause) {
+    super(message, cause);
+  }
+}

db-scheduler/src/main/java/com/github/kagkarlsson/scheduler/exceptions/TaskInstanceException.java

@@ -14,7 +14,7 @@
 package com.github.kagkarlsson.scheduler.exceptions;
 
 public class TaskInstanceException extends DbSchedulerException {
-  static final long serialVersionUID = -2132850112553296790L;
+  private static final long serialVersionUID = -2132850112553296791L;
   private static final String TASK_NAME_INSTANCE_MESSAGE_PART = " (task name: %s, instance id: %s)";
 
   private final String taskName;