Dynamically scheduling recurring tasks in a Multi-Instance environment

Previously, we used an on-premises infrastructure where recurring tasks were scheduled using Spring’s TaskScheduler. This interface allows scheduling tasks with fixed-rate, fixed-delay, or cron-like expressions. You can read more about it here.

The tasks were dynamically generated in response to events, and each task executed at its own time based on a cron schedule. This setup worked perfectly in a single-instance, on-prem environment.

However, moving to the cloud introduced new challenges:

• No duplicate task execution: Tasks must not run on multiple instances simultaneously.

• Failover support: If one instance shuts down, others should pick up its tasks.

• Recovery after downtime: If all instances go down, tasks should resume when the service recovers.

• Task management: Tasks should support updates, pauses, and deletions.

Meeting these requirements in a cloud environment is no small feat.

Presenting db-scheduler

After some research, I discovered a library called db-scheduler. According to its documentation, it is a task scheduler for Java inspired by the need for a clustered java.util.concurrent.ScheduledExecutorService.

Key Features

• Cluster-friendly: Guarantees execution by a single scheduler instance. (Nice!)

• Persistent tasks: Uses a single database table for persistence. (Wow!)

• Simple: Easy to use. (I’m in love.)

• And more! Check out the complete list of features here.

What Can It Do?

db-scheduler supports:

• Recurring static tasks: For example, sending customers daily emails with recent transactions.

• One-time tasks: Useful for tasks that need to run at application startup.

• Batch tasks: Schedule multiple tasks at once.

But how does it help with my specific problem?

How db-scheduler Solved My Problem

Addressing the Challenges

1. No duplicate execution: db-scheduler ensures that tasks are executed by only one instance at a time.

2. Failover support: Tasks are persisted in a database, so new instances can pick up where others left off.

3. Recovery after downtime: The database acts as a single source of truth, enabling recovery after all instances go down.

What remained was the ability to schedule, update, delete, and pause tasks dynamically.

Implementation

Let’s see now how I implemented it.

Adding the Dependency

First, I added the db-scheduler dependency to my project:

<dependency>
    <groupId>com.github.kagkarlsson</groupId>
    <artifactId>db-scheduler-spring-boot-starter</artifactId>
    <version>15.1.1</version>
</dependency>

Database Setup

I used the script in the documentation to create the necessary database table.

Scheduler Bean

The Scheduler bean is the core of the setup. It handles task scheduling and execution.

public static final String DYNAMIC_RECURRING_TASK_NAME = "dynamic-recurring-task";
public static final TaskDescriptor<ScheduleAndNoData> DYNAMIC_RECURRING_TASK =
        TaskDescriptor.of(DYNAMIC_RECURRING_TASK_NAME, ScheduleAndNoData.class);

@Bean
public Scheduler scheduler(DataSource dataSource, TaskExecutorService taskExecutorService, ScheduledTaskRepository scheduledTaskRepository) {
    final RecurringTaskWithPersistentSchedule<ScheduleAndNoData> dynamicRecurringTask =
            Tasks.recurringWithPersistentSchedule(DYNAMIC_RECURRING_TASK)
                    .execute((taskInstance, executionContext) -> taskExecutorService.startTaskExecution(taskInstance.getId()));

    Scheduler scheduler = Scheduler.create(dataSource, dynamicRecurringTask)
            .pollingInterval(Duration.ofSeconds(1))
            .addExecutionInterceptor(new TaskExecutionCustomInterceptor(scheduledTaskRepository))
            .registerShutdownHook()
            .build();

    scheduler.start();

    return scheduler;
}

Task Descriptor

The TaskDescriptor defines the characteristics of a task, such as its name and associated data type.

public static final String DYNAMIC_RECURRING_TASK_NAME = "dynamic-recurring-task";
public static final TaskDescriptor<ScheduleAndNoData> DYNAMIC_RECURRING_TASK =
        TaskDescriptor.of(DYNAMIC_RECURRING_TASK_NAME, ScheduleAndNoData.class);

Data Object

The ScheduleAndNoData class implements the ScheduleAndData interface. It includes the task's schedule and the isOnHold flag to determine whether the task should execute.

@Getter
@RequiredArgsConstructor
public static class ScheduleAndNoData implements ScheduleAndData, Serializable {
    @Serial
    private static final long serialVersionUID = 1L;
    private final CronSchedule schedule;
    private final boolean isOnHold;

    @Override
    public Object getData() {
        return null;
    }
}

Execution Interceptor

The TaskExecutionCustomInterceptor intercepts task execution to handle tasks that are on hold.

The ExecutionInterceptor is an interface that, when implemented, allows you to intercept task execution and modify the task's state, data, schedule or even cancel it. Here, I used different types of CompletitionHandler that the library offers.

In summary, you can use OnCompleteReschedule for recurring tasks that are on hold, so they are not executed at this time. Use OnCompleteReplace when you need to update the task's state or schedule.

@RequiredArgsConstructor
@Slf4j
static class TaskExecutionCustomInterceptor implements ExecutionInterceptor {
    private final ScheduledTaskRepository scheduledTaskRepository;

    private static boolean isDynamicRecurringTask(TaskInstance<?> taskInstance) {
        return taskInstance.getTaskName().equals(DYNAMIC_RECURRING_TASK_NAME) &&
                taskInstance.getData() instanceof ScheduleAndNoData;
    }

    @Override
    public CompletionHandler<?> execute(TaskInstance<?> taskInstance, ExecutionContext executionContext, ExecutionChain executionChain) {
        log.info("Custom interceptor is being executed");
        if (isDynamicRecurringTask(taskInstance)) {
            TaskInstance<ScheduleAndNoData> taskInstanceWithSchedule = (TaskInstance<ScheduleAndNoData>) taskInstance;
            if (isOnHold(taskInstanceWithSchedule)) {
                return handleOnHold(taskInstanceWithSchedule);
            }
            if (!isOnHold(taskInstanceWithSchedule) && isOnHoldDb(taskInstanceWithSchedule)) {
                return handleOnHoldNotInSyncWithDb("Task is on hold on the DB. Will be replaced", true, taskInstanceWithSchedule);
            }
        }
        log.info("Proceeding with the task execution");
        return executionChain.proceed(taskInstance, executionContext);
    }

    private CompletionHandler.OnCompleteReplace<ScheduleAndNoData> handleOnHoldNotInSyncWithDb(String log, boolean isOnHold, TaskInstance<ScheduleAndNoData> taskInstanceWithSchedule) {
        TaskExecutionCustomInterceptor.log.info(log);
        ScheduleAndNoData newData = new ScheduleAndNoData(taskInstanceWithSchedule.getData().getSchedule(), isOnHold);
        return new CompletionHandler.OnCompleteReplace<>(taskInstanceWithSchedule.getTaskName(), newData);
    }

    private CompletionHandler<?> handleOnHold(TaskInstance<ScheduleAndNoData> taskInstanceWithSchedule) {
        if (isOnHoldDb(taskInstanceWithSchedule)) {
            return rescheduleTaskExecutor(taskInstanceWithSchedule);
        }
        return handleOnHoldNotInSyncWithDb("Task was on hold. Will be replaced", false, taskInstanceWithSchedule);
    }

    private CompletionHandler.OnCompleteReschedule<Object> rescheduleTaskExecutor(TaskInstance<ScheduleAndNoData> taskInstanceWithSchedule) {
        log.info("Task is on hold. Will be rescheduled");
        return new CompletionHandler.OnCompleteReschedule<>(taskInstanceWithSchedule.getData().getSchedule());
    }

    private boolean isOnHold(TaskInstance<ScheduleAndNoData> taskInstanceWithSchedule) {
        return taskInstanceWithSchedule.getData().isOnHold();
    }

    private boolean isOnHoldDb(TaskInstance<ScheduleAndNoData> taskInstanceWithSchedule) {
        return scheduledTaskRepository.findByIdAndOnHoldTrue(taskInstanceWithSchedule.getId()).isPresent();
    }
}

Dynamic Task Scheduling

The TaskSchedulerService handles dynamic scheduling, updating, and cancelling of tasks.

@Service
@Slf4j
@RequiredArgsConstructor
public class TaskSchedulerService {
    private final Scheduler scheduler;

    public void scheduleTaskExecution(String taskId, String trigger) {
        log.info("Schedule task with taskId: {}", taskId);
        this.scheduler.scheduleIfNotExists(DYNAMIC_RECURRING_TASK
                .instance(taskId)
                .data(new SchedulerConfiguration.ScheduleAndNoData(new CronSchedule(trigger), false))
                .scheduledAccordingToData());
    }

    public void cancelTaskExecution(String taskId) {
        log.info("Cancel task with taskId: {}", taskId);
        try {
            this.scheduler.cancel(new TaskInstance<>(DYNAMIC_RECURRING_TASK_NAME, taskId));
        } catch (TaskInstanceNotFoundException e) {
            log.warn("Task not found with taskId: {}", taskId);
        } catch (TaskInstanceCurrentlyExecutingException e) {
            log.warn("Task with taskId {} is currently executing. Will be cancelled after execution.", taskId);
        }
        // in both exceptions, you can choose how you handle according to the project needs.
        // let's assume that after this succeeds, the task data is deleted also on the repository
    }

    public void updateTaskExecution(String taskId, String trigger, boolean onHold) {
        log.info("Update task with taskId: {}", taskId);
        this.cancelTaskExecution(taskId);
        this.scheduler.scheduleIfNotExists(DYNAMIC_RECURRING_TASK
                .instance(taskId)
                .data(new SchedulerConfiguration.ScheduleAndNoData(new CronSchedule(trigger), onHold))
                .scheduledAccordingToData());
    }
}

Testing the service:

I created a simple controller to schedule, cancel, and update tasks as a test:

1. I created an integration test that spins up 2 instances and schedules 2 tasks that run every 5 seconds.

2. Then, I stop instance 1 and create instance 3.

3. Finally, I stop instance 2 and create instance 4.

The test ensures that the tasks are not executed by more than one instance at a time and that they run every 5 seconds.

You can check the test here.

Conclusion

Using db-scheduler, I successfully implemented dynamic task scheduling in a multi-instance environment. The solution ensures no duplicate execution, supports failover, and allows tasks to be updated, paused, or deleted.

You can find the complete code in this GitHub repository.