MongoDB and Quartz

<dependencies>
  <dependency>
    <groupId>org.springframework.boot</groupId>
    <artifactId>spring-boot-starter-quartz</artifactId>
  </dependency>
  <dependency>
    <groupId>org.springframework.boot</groupId>
    <artifactId>spring-boot-starter-jdbc</artifactId>
  </dependency>
  <dependency>
    <groupId>org.springframework.boot</groupId>
    <artifactId>spring-boot-starter-data-mongodb</artifactId>
  </dependency>
  <dependency>
    <groupId>org.springframework.boot</groupId>
    <artifactId>spring-boot-starter-flyway</artifactId>
  </dependency>
  <dependency>
    <groupId>org.flywaydb</groupId>
    <artifactId>flyway-database-postgresql</artifactId>
  </dependency>
  <dependency>
    <groupId>org.postgresql</groupId>
    <artifactId>postgresql</artifactId>
  </dependency>
</dependencies>

package bookshelf;

import com.zaxxer.hikari.HikariDataSource;
import org.springframework.boot.context.properties.ConfigurationProperties;
import org.springframework.boot.jdbc.autoconfigure.DataSourceProperties;
import org.springframework.boot.quartz.autoconfigure.QuartzDataSource;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;

import javax.sql.DataSource;

/**
 * Quartz keeps its state in a small PostgreSQL database; the business data lives in MongoDB.
 * The pool is dedicated to Quartz and sized independently of everything else.
 */
@Configuration(proxyBeanMethods = false)
class QuartzDatabaseConfig {

    @Bean
    @ConfigurationProperties("bookshelf.quartz.datasource")
    DataSourceProperties quartzDataSourceProperties() {
        return new DataSourceProperties();
    }

    @Bean
    @QuartzDataSource
    @ConfigurationProperties("bookshelf.quartz.datasource.hikari")
    HikariDataSource quartzDataSource(DataSourceProperties quartzDataSourceProperties) {
        return quartzDataSourceProperties.initializeDataSourceBuilder().type(HikariDataSource.class).build();
    }
}

spring:
  application:
    name: bookshelf-service
  mongodb:
    uri: mongodb://${MONGODB_HOST:localhost}:27017/bookshelf
  flyway:
    locations: classpath:db/migration
  quartz:
    job-store-type: jdbc
    scheduler-name: BookshelfScheduler
    jdbc:
      initialize-schema: never            # the schema is owned by Flyway
    properties:
      org.quartz.scheduler.instanceId: AUTO
      org.quartz.jobStore.driverDelegateClass: org.quartz.impl.jdbcjobstore.PostgreSQLDelegate
      org.quartz.jobStore.isClustered: "true"
      org.quartz.threadPool.threadCount: "5"

bookshelf:
  quartz:
    datasource:
      url: jdbc:postgresql://${QUARTZ_DB_HOST:localhost}:5432/${QUARTZ_DB_NAME:quartz}
      username: ${QUARTZ_DB_USER:quartz}
      password: ${QUARTZ_DB_PASSWORD}
      hikari:
        maximum-pool-size: 8              # threadCount + 3
  reminders:
    lead: P1D
  reconciliation:
    interval: PT10M

-- Thanks to Patrick Lightbody for submitting this...
--
-- In your Quartz properties file, you'll need to set
-- org.quartz.jobStore.driverDelegateClass = org.quartz.impl.jdbcjobstore.PostgreSQLDelegate


CREATE TABLE QRTZ_JOB_DETAILS
(
  SCHED_NAME        VARCHAR(120) NOT NULL,
  JOB_NAME          VARCHAR(200) NOT NULL,
  JOB_GROUP         VARCHAR(200) NOT NULL,
  DESCRIPTION       VARCHAR(250) NULL,
  JOB_CLASS_NAME    VARCHAR(250) NOT NULL,
  IS_DURABLE        BOOL         NOT NULL,
  IS_NONCONCURRENT  BOOL         NOT NULL,
  IS_UPDATE_DATA    BOOL         NOT NULL,
  REQUESTS_RECOVERY BOOL         NOT NULL,
  JOB_DATA          BYTEA        NULL,
  PRIMARY KEY (SCHED_NAME, JOB_NAME, JOB_GROUP)
);

CREATE TABLE QRTZ_TRIGGERS
...

databaseChangeLog:
  - include:
      file: org/quartz/impl/jdbcjobstore/liquibase.quartz.init.xml

spring:
  flyway:
    enabled: false
  liquibase:
    change-log: classpath:db/changelog/db.changelog-master.yaml

package bookshelf.loans;

import org.springframework.data.annotation.Id;
import org.springframework.data.mongodb.core.index.Indexed;
import org.springframework.data.mongodb.core.mapping.Document;

import java.time.Instant;

@Document("loans")
public record Loan(
        @Id String id,
        @Indexed String memberId,
        String isbn,
        Instant dueAt,
        boolean returned,
        Instant reminderSentAt) {

    public Loan(String memberId, String isbn, Instant dueAt) {
        this(null, memberId, isbn, dueAt, false, null);
    }

    public Loan markReturned() {
        return new Loan(id, memberId, isbn, dueAt, true, reminderSentAt);
    }
}

package bookshelf.loans;

import org.springframework.data.mongodb.repository.MongoRepository;

import java.util.List;

public interface LoanRepository extends MongoRepository<Loan, String> {

    /** Loans whose reminder still has to be sent. */
    List<Loan> findByReturnedFalseAndReminderSentAtIsNull();
}

package bookshelf.loans;

import bookshelf.ReminderProperties;
import org.quartz.JobBuilder;
import org.quartz.JobDetail;
import org.quartz.JobKey;
import org.quartz.Scheduler;
import org.quartz.SimpleScheduleBuilder;
import org.quartz.SchedulerException;
import org.quartz.Trigger;
import org.quartz.TriggerBuilder;
import org.quartz.impl.matchers.GroupMatcher;
import org.springframework.stereotype.Component;

import java.util.Date;
import java.util.List;
import java.util.Set;

/** The only class that talks to Quartz about member reminders. Key convention: job and trigger "loan-<loanId>" in group "reminders". */
@Component
public class ReminderScheduler {

    public static final String GROUP = "reminders";
    private static final String KEY_PREFIX = "loan-";

    private final Scheduler scheduler;
    private final ReminderProperties properties;

    ReminderScheduler(Scheduler scheduler, ReminderProperties properties) {
        this.scheduler = scheduler;
        this.properties = properties;
    }

    public static JobKey keyFor(String loanId) {
        return JobKey.jobKey(KEY_PREFIX + loanId, GROUP);
    }

    /** Creates or replaces the reminder for a loan. Calling it twice leaves exactly one job and one trigger. */
    public void schedule(Loan loan) throws SchedulerException {
        JobDetail job = JobBuilder.newJob(MemberReminderJob.class)
                .withIdentity(keyFor(loan.id()))
                .usingJobData("loanId", loan.id())
                .usingJobData("memberId", loan.memberId())
                .build();
        Trigger trigger = TriggerBuilder.newTrigger()
                .withIdentity(KEY_PREFIX + loan.id(), GROUP)
                .forJob(job)
                .startAt(Date.from(loan.dueAt().minus(properties.lead())))
                .withSchedule(SimpleScheduleBuilder.simpleSchedule()
                        .withMisfireHandlingInstructionFireNow())
                .build();
        try {
            scheduler.scheduleJob(job, Set.of(trigger), true);
        } catch (SchedulerException e) {
            // LoanService and the reconciliation job can create the same reminder at the same moment; the
            // loser gets a duplicate key error from the primary key of QRTZ_JOB_DETAILS. The reminder exists, so that is fine.
            if (!exists(loan.id())) {
                throw e;
            }
        }
    }

    /** Deleting a job that is already gone is fine: the result is false, not an exception. */
    public boolean cancel(String loanId) throws SchedulerException {
        return scheduler.deleteJob(keyFor(loanId));
    }

    public boolean exists(String loanId) throws SchedulerException {
        return scheduler.checkExists(keyFor(loanId));
    }

    public List<String> scheduledLoanIds() throws SchedulerException {
        return scheduler.getJobKeys(GroupMatcher.jobGroupEquals(GROUP)).stream()
                .filter(k -> k.getName().startsWith(KEY_PREFIX))
                .map(k -> k.getName().substring(KEY_PREFIX.length()))
                .toList();
    }
}

package bookshelf.loans;

import org.quartz.SchedulerException;
import org.springframework.stereotype.Service;

import java.time.Instant;

/**
 * There is no transaction spanning MongoDB and the Quartz tables. MongoDB is written first (it is the source of truth),
 * Quartz second; if the second step fails the reconciliation job repairs it.
 */
@Service
public class LoanService {

    private final LoanRepository loans;
    private final ReminderScheduler reminders;

    LoanService(LoanRepository loans, ReminderScheduler reminders) {
        this.loans = loans;
        this.reminders = reminders;
    }

    public Loan lend(String memberId, String isbn, Instant dueAt) throws SchedulerException {
        Loan loan = loans.save(new Loan(memberId, isbn, dueAt));
        reminders.schedule(loan);
        return loan;
    }

    public void giveBack(String loanId) throws SchedulerException {
        loans.findById(loanId).ifPresent(loan -> loans.save(loan.markReturned()));
        reminders.cancel(loanId);
    }
}

package bookshelf.loans;

import org.quartz.JobExecutionContext;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.data.mongodb.core.MongoTemplate;
import org.springframework.data.mongodb.core.query.Criteria;
import org.springframework.data.mongodb.core.query.Query;
import org.springframework.data.mongodb.core.query.Update;
import org.springframework.scheduling.quartz.QuartzJobBean;

import java.time.Instant;

/**
 * Idempotent: it can run twice for the same loan (a retry after a crash, a recovered execution, a duplicate created
 * by reconciliation) and the member still receives one reminder.
 */
public class MemberReminderJob extends QuartzJobBean {

    private static final Logger log = LoggerFactory.getLogger(MemberReminderJob.class);

    private final MongoTemplate mongo;
    private final ReminderSender sender;
    private String loanId;
    private String memberId;

    MemberReminderJob(MongoTemplate mongo, ReminderSender sender) {
        this.mongo = mongo;
        this.sender = sender;
    }

    public void setLoanId(String loanId) {
        this.loanId = loanId;
    }

    public void setMemberId(String memberId) {
        this.memberId = memberId;
    }

    @Override
    protected void executeInternal(JobExecutionContext context) {
        Loan loan = mongo.findById(loanId, Loan.class);
        if (loan == null || loan.returned() || loan.reminderSentAt() != null) {
            log.info("Reminder for loan {} skipped (missing, returned or already sent)", loanId);
            return;
        }
        sender.send(memberId, loanId);
        // Only the first writer marks the loan; a concurrent duplicate would update nothing.
        mongo.updateFirst(
                Query.query(Criteria.where("_id").is(loanId).and("reminderSentAt").is(null)),
                Update.update("reminderSentAt", Instant.now()),
                Loan.class);
    }
}

package bookshelf.loans;

import org.quartz.DisallowConcurrentExecution;
import org.quartz.JobExecutionContext;
import org.quartz.JobExecutionException;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.scheduling.quartz.QuartzJobBean;

import java.util.HashSet;
import java.util.Set;

/**
 * Makes Quartz agree with MongoDB. MongoDB wins: active loans without a Quartz job get one, Quartz jobs whose loan is
 * returned, already reminded or gone are deleted.
 */
@DisallowConcurrentExecution
public class ReminderReconciliationJob extends QuartzJobBean {

    private static final Logger log = LoggerFactory.getLogger(ReminderReconciliationJob.class);

    private final LoanRepository loans;
    private final ReminderScheduler reminders;

    ReminderReconciliationJob(LoanRepository loans, ReminderScheduler reminders) {
        this.loans = loans;
        this.reminders = reminders;
    }

    @Override
    protected void executeInternal(JobExecutionContext context) throws JobExecutionException {
        try {
            Set<String> pending = new HashSet<>();
            for (Loan loan : loans.findByReturnedFalseAndReminderSentAtIsNull()) {
                pending.add(loan.id());
                if (!reminders.exists(loan.id())) {
                    log.warn("Loan {} had no reminder, scheduling it", loan.id());
                    reminders.schedule(loan);
                }
            }
            for (String scheduledId : reminders.scheduledLoanIds()) {
                if (!pending.contains(scheduledId)) {
                    log.warn("Reminder for loan {} has no pending loan, deleting it", scheduledId);
                    reminders.cancel(scheduledId);
                }
            }
        } catch (Exception e) {
            // refireImmediately=false: try again on the next tick
            throw new JobExecutionException(e, false);
        }
    }
}

package bookshelf.loans;

import org.quartz.JobBuilder;
import org.quartz.JobDetail;
import org.quartz.SimpleScheduleBuilder;
import org.quartz.Trigger;
import org.quartz.TriggerBuilder;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;

import java.time.Duration;

/** The one static job of this page: the reconciliation. The per-member reminders are created at runtime. */
@Configuration(proxyBeanMethods = false)
class ReminderScheduleConfig {

    @Bean
    JobDetail reminderReconciliationJobDetail() {
        return JobBuilder.newJob(ReminderReconciliationJob.class)
                .withIdentity("reminderReconciliation", "bookshelf")
                .storeDurably()
                .build();
    }

    @Bean
    Trigger reminderReconciliationTrigger(JobDetail reminderReconciliationJobDetail,
                                          @Value("${bookshelf.reconciliation.interval:PT10M}") Duration interval) {
        return TriggerBuilder.newTrigger()
                .forJob(reminderReconciliationJobDetail)
                .withIdentity("reminderReconciliationTrigger", "bookshelf")
                .withSchedule(SimpleScheduleBuilder.simpleSchedule()
                        .withIntervalInMilliseconds(interval.toMillis())
                        .repeatForever()
                        .withMisfireHandlingInstructionNextWithRemainingCount())
                .build();
    }
}

package bookshelf.loans;

import bookshelf.BookshelfApplication;
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.context.annotation.Bean;
import org.springframework.context.annotation.Import;
import org.springframework.jdbc.core.JdbcTemplate;
import org.springframework.test.context.DynamicPropertyRegistry;
import org.springframework.test.context.DynamicPropertySource;
import org.testcontainers.junit.jupiter.Container;
import org.testcontainers.junit.jupiter.Testcontainers;
import org.testcontainers.mongodb.MongoDBContainer;
import org.testcontainers.postgresql.PostgreSQLContainer;

import java.time.Duration;
import java.time.Instant;

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

@Testcontainers
@SpringBootTest(classes = {BookshelfApplication.class, BookshelfRemindersIT.TestConfig.class}, properties = {
        "bookshelf.reminders.lead=PT0S",
        "bookshelf.reconciliation.interval=PT2S",
        "spring.flyway.enabled=${flyway:true}",
        "spring.liquibase.change-log=classpath:db/changelog/db.changelog-master.yaml"
})
class BookshelfRemindersIT {

    @Container
    static final MongoDBContainer MONGO = new MongoDBContainer("mongo:8");

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

    @DynamicPropertySource
    static void properties(DynamicPropertyRegistry registry) {
        registry.add("spring.mongodb.uri", () -> MONGO.getReplicaSetUrl("bookshelf"));
        registry.add("bookshelf.quartz.datasource.url", POSTGRES::getJdbcUrl);
        registry.add("bookshelf.quartz.datasource.username", POSTGRES::getUsername);
        registry.add("bookshelf.quartz.datasource.password", POSTGRES::getPassword);
    }

    static class TestConfig {
        @Bean
        @org.springframework.context.annotation.Primary
        RecordingSender recordingSender() {
            return new RecordingSender();
        }
    }

    @Autowired LoanService service;
    @Autowired LoanRepository loans;
    @Autowired ReminderScheduler reminders;
    @Autowired RecordingSender sender;
    @Autowired Scheduler scheduler;

    @Autowired(required = false)
    org.springframework.data.mongodb.core.MongoTemplate mongo;

    private int quartzRows(String table) {
        try (var c = new org.springframework.jdbc.datasource.SingleConnectionDataSource(
                POSTGRES.getJdbcUrl(), POSTGRES.getUsername(), POSTGRES.getPassword(), true)) {
            return new JdbcTemplate(c).queryForObject("select count(*) from " + table, Integer.class);
        }
    }

    @Test
    void quartzTablesLiveInPostgresAndLoansInMongo() throws Exception {
        assertThat(quartzRows("qrtz_job_details")).isGreaterThanOrEqualTo(1);       // the reconciliation job
        assertThat(mongo.getCollectionNames()).contains("loans").doesNotContain("qrtz_job_details");
        assertThat(scheduler.getMetaData().getJobStoreClass().getSimpleName()).isEqualTo("LocalDataSourceJobStore");
    }

    @Test
    void reminderFiresOnceAndTheLoanIsMarked() throws Exception {
        Loan loan = service.lend("m-1", "978-0-13-468599-1", Instant.now().plusSeconds(2));
        assertThat(reminders.exists(loan.id())).isTrue();
        assertThat(quartzRows("qrtz_triggers")).isGreaterThanOrEqualTo(2);
        await().atMost(Duration.ofSeconds(15)).until(() -> sender.sent.contains("m-1:" + loan.id()));
        await().atMost(Duration.ofSeconds(5)).until(() -> loans.findById(loan.id()).orElseThrow().reminderSentAt() != null);
        assertThat(sender.sent.stream().filter(s -> s.endsWith(loan.id()))).hasSize(1);
    }

    @Test
    void returningTheBookCancelsTheReminder() throws Exception {
        Loan loan = service.lend("m-2", "978-0-596-52068-7", Instant.now().plusSeconds(60));
        assertThat(reminders.exists(loan.id())).isTrue();
        service.giveBack(loan.id());
        assertThat(reminders.exists(loan.id())).isFalse();
        assertThat(loans.findById(loan.id()).orElseThrow().returned()).isTrue();
        assertThat(sender.sent).noneMatch(s -> s.endsWith(loan.id()));
    }

    @Test
    void schedulingTwiceLeavesOneJob() throws Exception {
        Loan loan = loans.save(new Loan("m-3", "978-1-4919-5025-7", Instant.now().plusSeconds(3600)));
        reminders.schedule(loan);
        reminders.schedule(loan);
        assertThat(reminders.scheduledLoanIds().stream().filter(loan.id()::equals)).hasSize(1);
        assertThat(scheduler.getTriggersOfJob(ReminderScheduler.keyFor(loan.id()))).hasSize(1);
    }

    @Test
    void reconciliationCreatesAMissingReminder() throws Exception {
        // MongoDB was written, the process died before Quartz was called
        Loan loan = loans.save(new Loan("m-4", "978-0-321-12521-7", Instant.now().plusSeconds(3)));
        assertThat(reminders.exists(loan.id())).isFalse();
        await().atMost(Duration.ofSeconds(20)).until(() -> sender.sent.contains("m-4:" + loan.id()));
    }

    @Test
    void reconciliationDeletesAnOrphanReminder() throws Exception {
        // Quartz was written, the loan was returned in MongoDB and the cancel never happened
        Loan loan = loans.save(new Loan("m-5", "978-0-201-63361-0", Instant.now().plusSeconds(3600)));
        reminders.schedule(loan);
        loans.save(loan.markReturned());
        await().atMost(Duration.ofSeconds(20)).until(() -> !reminders.exists(loan.id()));
        assertThat(sender.sent).noneMatch(s -> s.endsWith(loan.id()));
    }

    @Test
    void theJobIsIdempotent() throws Exception {
        Loan loan = loans.save(new Loan("m-6", "978-0-13-235088-4", Instant.now().plusSeconds(3600)));
        reminders.schedule(loan);
        var key = ReminderScheduler.keyFor(loan.id());
        scheduler.triggerJob(key);
        await().atMost(Duration.ofSeconds(10)).until(() -> sender.sent.contains("m-6:" + loan.id()));
        await().atMost(Duration.ofSeconds(5)).until(() -> loans.findById(loan.id()).orElseThrow().reminderSentAt() != null);
        scheduler.triggerJob(key);          // a duplicate run, as after a recovered execution
        Thread.sleep(2000);
        assertThat(sender.sent.stream().filter(s -> s.endsWith(loan.id()))).hasSize(1);
    }

    @Test
    void aFailedReminderIsRecreatedByReconciliation() throws Exception {
        sender.failNext.set(true);
        Loan loan = service.lend("m-7", "978-0-13-110362-7", Instant.now().plusSeconds(1));
        // the first run fails and the one-shot trigger is finished, so Quartz deletes the non-durable job
        await().atMost(Duration.ofSeconds(10)).until(() -> !reminders.exists(loan.id()) || !sender.failNext.get());
        // the loan is still pending, so reconciliation schedules it again and the second attempt succeeds
        await().atMost(Duration.ofSeconds(30)).until(() -> sender.sent.contains("m-7:" + loan.id()));
        assertThat(sender.sent.stream().filter(s -> s.endsWith(loan.id()))).hasSize(1);
    }
}

// qrtz_jobs
{ keyName: 'slow', keyGroup: 'bookshelf', jobDescription: null,
  jobClass: 'probe.ProbeApplication$SlowJob', durability: true, requestsRecovery: false }

// qrtz_triggers
{ state: 'waiting', class: 'org.quartz.impl.triggers.SimpleTriggerImpl', keyName: 'slowTrigger', keyGroup: 'bookshelf',
  jobId: ObjectId('...'), misfireInstruction: 0, nextFireTime: ISODate('2026-10-05T09:57:55.761Z'),
  previousFireTime: ISODate('2026-10-05T09:57:54.761Z'), priority: 5, repeatCount: -1, repeatInterval: Long('1000'),
  timesTriggered: 1 }

// qrtz_locks, seen while a @DisallowConcurrentExecution job was running
{ type: 't', keyGroup: 'bookshelf', keyName: 'slowTrigger', instanceId: 'host-<millis>' }
{ type: 'j', keyGroup: 'bookshelf', keyName: 'slow', instanceId: 'host-<millis>' }

// qrtz_schedulers
{ schedulerName: 'quartzScheduler', instanceId: 'host-<millis>', checkinInterval: Long('2000'), lastCheckinTime: Long('1791194289929') }

Exception in thread "quartzScheduler_QuartzSchedulerThread" java.lang.NoSuchMethodError: 'boolean org.quartz.JobDetail.isConcurrentExectionDisallowed()'
	at com.novemberain.quartz.mongodb.LockManager.lockJob(LockManager.java:31)
	at com.novemberain.quartz.mongodb.TriggerRunner.triggersFired(TriggerRunner.java:89)
	at com.novemberain.quartz.mongodb.MongoDBJobStore.triggersFired(MongoDBJobStore.java:488)
	at org.quartz.core.QuartzSchedulerThread.run(QuartzSchedulerThread.java:364)

<properties>
  <quartz.version>2.4.1</quartz.version>
</properties>

package jobstoresketch;

import com.mongodb.client.MongoCollection;
import com.mongodb.client.model.FindOneAndUpdateOptions;
import com.mongodb.client.model.Indexes;
import com.mongodb.client.model.ReturnDocument;
import com.mongodb.client.model.Updates;
import org.bson.Document;

import java.time.Duration;
import java.time.Instant;
import java.util.ArrayList;
import java.util.Date;
import java.util.List;

import static com.mongodb.client.model.Filters.and;
import static com.mongodb.client.model.Filters.eq;
import static com.mongodb.client.model.Filters.lte;
import static com.mongodb.client.model.Filters.or;
import static com.mongodb.client.model.Sorts.ascending;

/**
 * The heart of a MongoDB JobStore, as a sketch: JobStore.acquireNextTriggers without the rest of the SPI.
 * One findOneAndUpdate is atomic per document, so two schedulers can never claim the same trigger.
 */
public class MongoTriggerAcquisition {

    private final MongoCollection<Document> triggers;
    private final String instanceId;

    public MongoTriggerAcquisition(MongoCollection<Document> triggers, String instanceId) {
        this.triggers = triggers;
        this.instanceId = instanceId;
    }

    public void createIndexes() {
        triggers.createIndex(Indexes.compoundIndex(Indexes.ascending("state"), Indexes.ascending("nextFireTime")));
    }

    /** Claims up to maxCount triggers due before noLaterThan; every claimed trigger is returned exactly once. */
    public List<Document> acquireNextTriggers(Instant noLaterThan, int maxCount, Duration claimTimeout) {
        List<Document> acquired = new ArrayList<>();
        Instant now = Instant.now();
        for (int i = 0; i < maxCount; i++) {
            Document claimed = triggers.findOneAndUpdate(
                    and(lte("nextFireTime", Date.from(noLaterThan)),
                            or(eq("state", "waiting"),
                                    // a node that died after claiming: its claim expired, take the trigger over
                                    and(eq("state", "acquired"), lte("claimExpiresAt", Date.from(now))))),
                    Updates.combine(
                            Updates.set("state", "acquired"),
                            Updates.set("claimedBy", instanceId),
                            Updates.set("claimExpiresAt", Date.from(now.plus(claimTimeout)))),
                    new FindOneAndUpdateOptions().sort(ascending("nextFireTime")).returnDocument(ReturnDocument.AFTER));
            if (claimed == null) {
                break;
            }
            acquired.add(claimed);
        }
        return acquired;
    }
}

package jobstoresketch;

import com.mongodb.ErrorCategory;
import com.mongodb.MongoWriteException;
import com.mongodb.client.MongoCollection;
import com.mongodb.client.model.IndexOptions;
import com.mongodb.client.model.Indexes;
import org.bson.Document;

import java.time.Duration;
import java.time.Instant;
import java.util.Date;
import java.util.concurrent.TimeUnit;

import static com.mongodb.client.model.Filters.and;
import static com.mongodb.client.model.Filters.eq;
import static com.mongodb.client.model.Filters.lte;

/**
 * Lock documents with an expiry, the Mongo answer to QRTZ_LOCKS. The unique _id decides who owns a lock,
 * the TTL index only cleans up after crashed owners.
 */
public class MongoLocks {

    private final MongoCollection<Document> locks;

    public MongoLocks(MongoCollection<Document> locks) {
        this.locks = locks;
    }

    public void createIndexes() {
        locks.createIndex(Indexes.ascending("expiresAt"), new IndexOptions().expireAfter(0L, TimeUnit.SECONDS));
    }

    public boolean tryLock(String name, String owner, Duration timeout) {
        Instant now = Instant.now();
        // the TTL monitor only runs about once a minute, so an expired lock is also removed here
        locks.deleteOne(and(eq("_id", name), lte("expiresAt", Date.from(now))));
        try {
            locks.insertOne(new Document("_id", name).append("owner", owner)
                    .append("expiresAt", Date.from(now.plus(timeout))));
            return true;
        } catch (MongoWriteException e) {
            if (e.getError().getCategory() == ErrorCategory.DUPLICATE_KEY) {
                return false;
            }
            throw e;
        }
    }

    public void unlock(String name, String owner) {
        locks.deleteOne(and(eq("_id", name), eq("owner", owner)));
    }
}