Quartz on PostgreSQL

CREATE ROLE quartz_app LOGIN PASSWORD 'change-me-as-well';

GRANT USAGE ON SCHEMA quartz TO quartz_app;

-- Tables created later by role quartz (Flyway) become usable by quartz_app automatically.
ALTER DEFAULT PRIVILEGES FOR ROLE quartz IN SCHEMA quartz
    GRANT SELECT, INSERT, UPDATE, DELETE ON TABLES TO quartz_app;

ALTER ROLE quartz_app IN DATABASE bookshelf SET search_path = quartz;

spring:
  datasource:
    url: jdbc:postgresql://${DB_HOST:localhost}:5432/bookshelf?currentSchema=quartz
    username: quartz_app
    password: ${DB_PASSWORD}
  flyway:
    user: quartz
    password: ${DB_MIGRATION_PASSWORD}

import org.springframework.beans.factory.annotation.Qualifier;
import org.springframework.boot.quartz.autoconfigure.QuartzDataSource;
import org.springframework.boot.quartz.autoconfigure.QuartzTransactionManager;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.context.annotation.Primary;
import org.springframework.jdbc.datasource.DataSourceTransactionManager;
import org.springframework.transaction.PlatformTransactionManager;

import javax.sql.DataSource;

// A separate class, not a subclass of DataSourcesConfiguration: it only consumes the two DataSource beans.
@Configuration(proxyBeanMethods = false)
class QuartzTransactionConfiguration {

    @Bean @Primary
    PlatformTransactionManager appTransactionManager(@Qualifier("appDataSource") DataSource appDataSource) {
        return new DataSourceTransactionManager(appDataSource);
    }

    @Bean @QuartzTransactionManager
    PlatformTransactionManager quartzTransactionManager(@QuartzDataSource DataSource quartzDataSource) {
        return new DataSourceTransactionManager(quartzDataSource);
    }
}

import org.quartz.JobDetail;
import org.quartz.Scheduler;
import org.quartz.Trigger;
import org.springframework.boot.quartz.autoconfigure.QuartzTransactionManager;
import org.springframework.stereotype.Service;
import org.springframework.transaction.PlatformTransactionManager;
import org.springframework.transaction.support.TransactionTemplate;

@Service
class ReminderScheduling {

    private final Scheduler scheduler;
    private final TransactionTemplate quartzTransaction;

    ReminderScheduling(Scheduler scheduler,
                       @QuartzTransactionManager PlatformTransactionManager quartzTransactionManager) {
        this.scheduler = scheduler;
        this.quartzTransaction = new TransactionTemplate(quartzTransactionManager);
    }

    void schedule(JobDetail job, Trigger trigger) {
        quartzTransaction.executeWithoutResult(status -> {
            try {
                scheduler.scheduleJob(job, trigger);    // one BEGIN ... COMMIT on the Quartz database
            } catch (org.quartz.SchedulerException e) {
                throw new IllegalStateException(e);     // rolls the Quartz transaction back
            }
        });
    }
}

-- connection 268: the scheduler thread acquires the trigger. No FOR UPDATE.
BEGIN
SELECT TRIGGER_NAME, TRIGGER_GROUP, NEXT_FIRE_TIME, PRIORITY FROM QRTZ_TRIGGERS WHERE SCHED_NAME = 'BookshelfCluster' AND TRIGGER_STATE = $1 AND NEXT_FIRE_TIME <= $2 AND (MISFIRE_INSTR = -1 OR (...)) ORDER BY NEXT_FIRE_TIME ASC, PRIORITY DESC
SELECT * FROM QRTZ_TRIGGERS WHERE SCHED_NAME = ... AND TRIGGER_NAME = $1 AND TRIGGER_GROUP = $2
SELECT * FROM QRTZ_SIMPLE_TRIGGERS WHERE ...
SELECT * FROM QRTZ_JOB_DETAILS WHERE ...
UPDATE QRTZ_TRIGGERS SET TRIGGER_STATE = $1 WHERE SCHED_NAME = ... AND TRIGGER_NAME = $2 AND TRIGGER_GROUP = $3 AND TRIGGER_STATE = $4
INSERT INTO QRTZ_FIRED_TRIGGERS (SCHED_NAME, ENTRY_ID, TRIGGER_NAME, TRIGGER_GROUP, INSTANCE_NAME, FIRED_TIME, SCHED_TIME, STATE, ...) VALUES (...)
COMMIT

-- connection 268 again, when the fire time is reached: the trigger is fired. Lock first.
BEGIN
SELECT * FROM QRTZ_LOCKS WHERE SCHED_NAME = 'BookshelfCluster' AND LOCK_NAME = $1 FOR UPDATE
SELECT TRIGGER_STATE FROM QRTZ_TRIGGERS WHERE ...
SELECT * FROM QRTZ_JOB_DETAILS WHERE ...
UPDATE QRTZ_FIRED_TRIGGERS SET INSTANCE_NAME = $1, FIRED_TIME = $2, SCHED_TIME = $3, STATE = $4, JOB_NAME = $5, ... WHERE ...
UPDATE QRTZ_TRIGGERS SET JOB_NAME = $1, JOB_GROUP = $2, DESCRIPTION = $3, NEXT_FIRE_TIME = $4, PREV_FIRE_TIME = $5, TRIGGER_STATE = $6, ... WHERE ...
UPDATE QRTZ_SIMPLE_TRIGGERS SET REPEAT_COUNT = $1, REPEAT_INTERVAL = $2, TIMES_TRIGGERED = $3 WHERE ...
COMMIT

-- connection 267: a worker thread, after the job returned. Lock again.
BEGIN
SELECT * FROM QRTZ_LOCKS WHERE SCHED_NAME = 'BookshelfCluster' AND LOCK_NAME = $1 FOR UPDATE
DELETE FROM QRTZ_FIRED_TRIGGERS WHERE SCHED_NAME = 'BookshelfCluster' AND ENTRY_ID = $1
COMMIT

-- called from a plain method: no BEGIN, no COMMIT, every statement autocommits
SELECT * FROM QRTZ_LOCKS WHERE SCHED_NAME = 'TxProbe' AND LOCK_NAME = $1 FOR UPDATE
SELECT JOB_NAME FROM QRTZ_JOB_DETAILS WHERE ...
INSERT INTO QRTZ_JOB_DETAILS (...) VALUES (...)
INSERT INTO QRTZ_TRIGGERS (...) VALUES (...)
INSERT INTO QRTZ_SIMPLE_TRIGGERS (...) VALUES (...)

-- called inside TransactionTemplate.executeWithoutResult: one transaction
BEGIN
SELECT * FROM QRTZ_LOCKS WHERE SCHED_NAME = 'TxProbe' AND LOCK_NAME = $1 FOR UPDATE
SELECT JOB_NAME FROM QRTZ_JOB_DETAILS WHERE ...
INSERT INTO QRTZ_JOB_DETAILS (...) VALUES (...)
INSERT INTO QRTZ_TRIGGERS (...) VALUES (...)
INSERT INTO QRTZ_SIMPLE_TRIGGERS (...) VALUES (...)
COMMIT

SELECT pid, state, wait_event_type, wait_event, left(query, 70) AS query
FROM pg_stat_activity
WHERE datname = 'bookshelf' AND usename = 'quartz'
  AND wait_event_type IS NOT NULL AND state = 'active'
ORDER BY pid;

package bookshelf;

import static org.assertj.core.api.Assertions.assertThat;
import static org.awaitility.Awaitility.await;

import java.time.Duration;
import java.util.List;

import org.junit.jupiter.api.Test;
import org.quartz.Scheduler;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.test.context.SpringBootTest;
import org.springframework.boot.testcontainers.service.connection.ServiceConnection;
import org.springframework.jdbc.core.JdbcTemplate;
import org.testcontainers.junit.jupiter.Container;
import org.testcontainers.junit.jupiter.Testcontainers;
import org.testcontainers.postgresql.PostgreSQLContainer;

@SpringBootTest(properties = {
        "bookshelf.fast=true",
        "spring.quartz.job-store-type=jdbc",
        "spring.quartz.jdbc.initialize-schema=never",          // Flyway creates the tables from db/migration/V1__quartz_tables.sql
        "spring.quartz.scheduler-name=BookshelfScheduler",
        "spring.quartz.properties.org.quartz.jobStore.driverDelegateClass=org.quartz.impl.jdbcjobstore.PostgreSQLDelegate"
})
@Testcontainers
class BookshelfPostgresIT {

    @Container
    @ServiceConnection
    static final PostgreSQLContainer POSTGRES = new PostgreSQLContainer("postgres:17");

    @Autowired
    Scheduler scheduler;

    @Autowired
    JdbcTemplate jdbc;

    @Test
    void bookshelfJobsAreStoredInPostgresAndFire() throws Exception {
        assertThat(scheduler.getMetaData().getJobStoreClass().getSimpleName()).isEqualTo("LocalDataSourceJobStore");
        assertThat(scheduler.getMetaData().isJobStoreSupportsPersistence()).isTrue();

        // the four Bookshelf beans were stored on startup, in the bytea-backed tables created by Flyway
        assertThat(jdbc.queryForObject("SELECT COUNT(*) FROM QRTZ_JOB_DETAILS", Integer.class)).isEqualTo(4);
        assertThat(jdbc.queryForObject("SELECT COUNT(*) FROM QRTZ_TRIGGERS", Integer.class)).isEqualTo(4);

        // the one-second cache refresh fires and the store records it
        await().atMost(Duration.ofSeconds(30)).untilAsserted(() -> assertThat(jdbc.queryForObject(
                "SELECT COUNT(*) FROM QRTZ_TRIGGERS WHERE TRIGGER_NAME = 'cacheRefreshTrigger' AND PREV_FIRE_TIME > 0",
                Integer.class)).isEqualTo(1));

        // Quartz created its lock rows lazily the first time it needed them
        List<String> locks = jdbc.queryForList("SELECT LOCK_NAME FROM QRTZ_LOCKS", String.class);
        assertThat(locks).contains("TRIGGER_ACCESS");
    }
}

-- Executed by the Testcontainers PostgreSQL container as its superuser, in the database "bookshelf".
CREATE ROLE quartz LOGIN PASSWORD 'change-me-from-a-secret-store';
CREATE SCHEMA quartz AUTHORIZATION quartz;
ALTER ROLE quartz IN DATABASE bookshelf SET search_path = quartz;
REVOKE ALL ON SCHEMA quartz FROM PUBLIC;

package bookshelf;

import org.quartz.JobBuilder;
import org.quartz.Scheduler;
import org.quartz.TriggerBuilder;
import org.springframework.boot.WebApplicationType;
import org.springframework.boot.builder.SpringApplicationBuilder;
import org.springframework.context.ConfigurableApplicationContext;

import java.util.Arrays;
import java.util.Date;

/** One cluster node as its own JVM: {@code java -cp ... bookshelf.BookshelfNode --spring.datasource.url=...}. */
public class BookshelfNode {

    public static void main(String[] args) throws Exception {
        ConfigurableApplicationContext context = new SpringApplicationBuilder(BookshelfApplication.class)
                .web(WebApplicationType.NONE)
                .run(args);
        Scheduler scheduler = context.getBean(Scheduler.class);
        System.out.println("NODE-READY " + scheduler.getSchedulerInstanceId());

        if (Arrays.asList(args).contains("--bookshelf.schedule-weekly-report-now=true")) {
            // the weekly report: long, must not overlap, must be re-run if its node dies
            scheduler.scheduleJob(
                    JobBuilder.newJob(SlowRecoverableJob.class).withIdentity("weeklyReport", "bookshelf")
                            .requestRecovery(true).build(),
                    TriggerBuilder.newTrigger().withIdentity("weeklyReportNow", "bookshelf").startAt(new Date()).build());
        }
        Thread.currentThread().join();
    }
}

public abstract class RecordingJob extends QuartzJobBean {
    @Override
    protected void executeInternal(JobExecutionContext ctx) throws JobExecutionException {
        String node = instanceId(ctx);
        long scheduled = ctx.getScheduledFireTime() == null ? 0 : ctx.getScheduledFireTime().getTime();
        System.out.println("FIRE " + ctx.getJobDetail().getKey().getName() + " node=" + node
                + " scheduled=" + scheduled + " recovering=" + ctx.isRecovering());
        work(ctx);
    }

    private static String instanceId(JobExecutionContext ctx) {
        try { return ctx.getScheduler().getSchedulerInstanceId(); } catch (Exception e) { return "?"; }
    }

    protected void work(JobExecutionContext ctx) throws JobExecutionException { }
}

/** Stand-in for the long weekly statistics report: slow, never overlapping, re-run after a node crash. */
@DisallowConcurrentExecution
public class SlowRecoverableJob extends RecordingJob {
    @Override
    protected void work(JobExecutionContext c) {
        try { Thread.sleep(c.isRecovering() ? 500 : 25_000); } catch (InterruptedException e) { Thread.currentThread().interrupt(); }
    }
}

package bookshelf;

import org.junit.jupiter.api.Test;
import org.testcontainers.junit.jupiter.Container;
import org.testcontainers.junit.jupiter.Testcontainers;
import org.testcontainers.postgresql.PostgreSQLContainer;

import java.io.BufferedReader;
import java.io.InputStreamReader;
import java.sql.Connection;
import java.sql.DriverManager;
import java.sql.ResultSet;
import java.sql.Statement;
import java.util.ArrayList;
import java.util.List;
import java.util.Map;
import java.util.TreeMap;
import java.util.concurrent.CopyOnWriteArrayList;
import java.util.function.Predicate;
import java.util.stream.Collectors;

import static org.assertj.core.api.Assertions.assertThat;

@Testcontainers
class PostgresClusterTest {

    static final String QUARTZ_PASSWORD = "change-me-from-a-secret-store"; // same placeholder as db/init-quartz-schema.sql

    @Container
    static final PostgreSQLContainer POSTGRES = new PostgreSQLContainer("postgres:17")
            .withDatabaseName("bookshelf")
            .withInitScript("db/init-quartz-schema.sql");   // CREATE ROLE quartz, CREATE SCHEMA quartz

    /** A cluster node = a separate JVM running bookshelf.BookshelfNode, so that it can be killed with SIGKILL. */
    static final class Node implements AutoCloseable {
        final Process process;
        final List<String> output = new CopyOnWriteArrayList<>();

        Node(String schedulerName, String... extraArgs) throws Exception {
            List<String> command = new ArrayList<>(List.of(
                    System.getProperty("java.home") + "/bin/java", "-cp", System.getProperty("java.class.path"),
                    "bookshelf.BookshelfNode",
                    "--spring.main.banner-mode=off", "--logging.level.root=WARN",
                    "--logging.level.org.springframework.scheduling.quartz=INFO", // the job store logs under its own class name: shows ClusterManager messages
                    "--bookshelf.fast=true",
                    "--spring.datasource.url=jdbc:postgresql://%s:%d/bookshelf?currentSchema=quartz"
                            .formatted(POSTGRES.getHost(), POSTGRES.getFirstMappedPort()),
                    "--spring.datasource.username=quartz",
                    "--spring.datasource.password=" + QUARTZ_PASSWORD,
                    "--spring.flyway.enabled=true",                       // V1__quartz_tables.sql creates the tables
                    "--spring.quartz.job-store-type=jdbc",
                    "--spring.quartz.jdbc.initialize-schema=never",
                    "--spring.quartz.scheduler-name=" + schedulerName,   // the same name on every node of one cluster
                    "--spring.quartz.properties.org.quartz.scheduler.instanceId=AUTO",
                    "--spring.quartz.properties.org.quartz.jobStore.driverDelegateClass=org.quartz.impl.jdbcjobstore.PostgreSQLDelegate",
                    "--spring.quartz.properties.org.quartz.jobStore.isClustered=true",
                    "--spring.quartz.properties.org.quartz.jobStore.clusterCheckinInterval=2000",
                    "--spring.quartz.properties.org.quartz.threadPool.threadCount=3"));
            command.addAll(List.of(extraArgs));
            process = new ProcessBuilder(command).redirectErrorStream(true).start();
            Thread reader = new Thread(() -> {
                try (BufferedReader in = new BufferedReader(new InputStreamReader(process.getInputStream()))) {
                    String line;
                    while ((line = in.readLine()) != null) output.add(line);
                } catch (Exception ignored) { }
            });
            reader.setDaemon(true);
            reader.start();
            awaitLine(60_000, l -> l.startsWith("NODE-READY"));
        }

        String awaitLine(long millis, Predicate<String> match) throws InterruptedException {
            long end = System.currentTimeMillis() + millis;
            while (System.currentTimeMillis() < end) {
                for (String l : output) if (match.test(l)) return l;
                Thread.sleep(100);
            }
            throw new AssertionError("no matching output within " + millis + " ms:\n" + String.join("\n", output));
        }

        List<String> fires(String job) {
            return output.stream().filter(l -> l.startsWith("FIRE " + job + " ")).toList();
        }

        @Override public void close() { process.destroyForcibly(); }
    }

    static List<Map<String, Object>> query(String sql) throws Exception {
        try (Connection c = DriverManager.getConnection(POSTGRES.getJdbcUrl(), POSTGRES.getUsername(), POSTGRES.getPassword());
             Statement st = c.createStatement(); ResultSet rs = st.executeQuery(sql)) {
            List<Map<String, Object>> rows = new ArrayList<>();
            while (rs.next()) {
                Map<String, Object> row = new TreeMap<>();
                for (int i = 1; i <= rs.getMetaData().getColumnCount(); i++) row.put(rs.getMetaData().getColumnLabel(i), rs.getObject(i));
                rows.add(row);
            }
            return rows;
        }
    }

    @Test
    void threeNodesShareTheWorkAndNeverRunTheSameFireTimeTwice() throws Exception {
        try (Node a = new Node("share"); Node b = new Node("share"); Node c = new Node("share")) {   // a runs Flyway, b and c find the tables
            Thread.sleep(15_000);

            assertThat(query("SELECT instance_name FROM quartz.qrtz_scheduler_state WHERE sched_name = 'share'")).hasSize(3);

            List<String> fires = new ArrayList<>();
            for (Node n : List.of(a, b, c)) { fires.addAll(n.fires("cacheRefresh")); fires.addAll(n.fires("overdueLoanReminders")); }
            assertThat(fires).isNotEmpty();

            // "FIRE <job> node=<id> scheduled=<epoch ms> recovering=<b>": the job plus its scheduled time must be unique
            Map<String, Long> perFireTime = fires.stream().collect(Collectors.groupingBy(
                    l -> l.split(" ")[1] + " " + l.split(" ")[3], Collectors.counting()));
            assertThat(perFireTime.values()).allMatch(count -> count == 1);

            for (String job : List.of("cacheRefresh", "overdueLoanReminders"))
                System.out.println(job + ": a=" + a.fires(job).size() + " b=" + b.fires(job).size() + " c=" + c.fires(job).size());
        }
    }

    @Test
    void aCrashedNodeIsDetectedAndItsRecoverableJobIsRunAgain() throws Exception {
        try (Node x = new Node("crash", "--bookshelf.schedule-weekly-report-now=true")) {
            x.awaitLine(40_000, l -> l.startsWith("FIRE weeklyReport "));   // x is now inside the 25 s report
            try (Node y = new Node("crash")) {                                // y starts after x took the job
                assertThat(query("SELECT state, requests_recovery FROM quartz.qrtz_fired_triggers WHERE sched_name = 'crash' AND job_name = 'weeklyReport'"))
                        .singleElement().satisfies(row -> {
                            assertThat(row.get("state")).isEqualTo("EXECUTING");
                            assertThat(row.get("requests_recovery")).isEqualTo(true);
                        });

                long killedAt = System.currentTimeMillis();
                x.process.destroyForcibly().waitFor();                       // kill -9: no shutdown hook, no clean-up

                String line = y.awaitLine(60_000, l -> l.startsWith("FIRE weeklyReport ") && l.endsWith("recovering=true"));
                System.out.println("recovered after " + (System.currentTimeMillis() - killedAt) + " ms: " + line);
                y.output.stream().filter(l -> l.contains("ClusterManager")).forEach(System.out::println);
                assertThat(query("SELECT instance_name FROM quartz.qrtz_scheduler_state WHERE sched_name = 'crash'")).hasSize(1);
            }
        }
    }
}