public Optional<JobTriggerDto> nextRunnableTrigger() {
        final DateTime now = clock.nowUTC();

        // exclude triggers which require a constraint that is not satisfied by this node
        final var constraintsQuery = not(
                elemMatch(FIELD_CONSTRAINTS, new Document("$nin", schedulerCapabilitiesService.getNodeCapabilities()))
        );

        final var filter = or(and(
                        // We cannot lock a trigger that is already locked by another node
                        eq(FIELD_LOCK_OWNER, null),
                        eq(FIELD_STATUS, JobTriggerStatus.RUNNABLE),
                        lte(FIELD_START_TIME, now),
                        constraintsQuery,

                        or( // Skip triggers that have an endTime which is due
                                not(exists(FIELD_END_TIME)),
                                eq(FIELD_END_TIME, null),
                                gt(FIELD_END_TIME, Optional.of(now))
                        ),
                        // TODO: Using the wall clock time here can be problematic if the node time is off
                        //       The scheduler should not lock any new triggers if it detects that its clock is wrong
                        lte(FIELD_NEXT_TIME, now)
                ), and(
                        ne(FIELD_LOCK_OWNER, null),
                        ne(FIELD_LOCK_OWNER, nodeId),
                        eq(FIELD_STATUS, JobTriggerStatus.RUNNING),
                        constraintsQuery,
                        lt(FIELD_LAST_LOCK_TIME, now.minus(lockExpirationDuration.toMilliseconds())))
        );
        // We want to lock the trigger with the oldest next time
        final var sort = ascending(FIELD_NEXT_TIME);

        final var lockUpdate = combine(
                set(FIELD_LOCK_OWNER, nodeId),
                set(FIELD_LAST_LOCK_OWNER, nodeId),
                set(FIELD_STATUS, JobTriggerStatus.RUNNING),
                set(FIELD_TRIGGERED_AT, Optional.of(now)),
                set(FIELD_LAST_LOCK_TIME, now)
        );

        // Atomically update, lock and return the next runnable trigger
        final JobTriggerDto trigger = collection.findOneAndUpdate(filter, lockUpdate,
                new FindOneAndUpdateOptions().sort(sort).returnDocument(ReturnDocument.AFTER)
        );

        return Optional.ofNullable(trigger);
    }
@Test
    public void nextRunnableTrigger() {
        // No triggers yet
        assertThat(dbJobTriggerService.nextRunnableTrigger()).isEmpty();

        final JobTriggerDto trigger1 = dbJobTriggerService.create(JobTriggerDto.Builder.create(clock)
                .jobDefinitionId("abc-123")
                .jobDefinitionType("event-processor-execution-v1")
                .nextTime(clock.nowUTC().plusSeconds(11))
                .schedule(IntervalJobSchedule.builder()
                        .interval(1)
                        .unit(TimeUnit.SECONDS)
                        .build())
                .build());

        final JobTriggerDto trigger2 = dbJobTriggerService.create(JobTriggerDto.Builder.create(clock)
                .jobDefinitionId("abc-123")
                .jobDefinitionType("event-processor-execution-v1")
                .nextTime(clock.nowUTC().plusSeconds(10))
                .schedule(IntervalJobSchedule.builder()
                        .interval(1)
                        .unit(TimeUnit.SECONDS)
                        .build())
                .build());

        final JobTriggerDto trigger3 = dbJobTriggerService.create(JobTriggerDto.Builder.create(clock)
                .jobDefinitionId("abc-123")
                .jobDefinitionType("event-processor-execution-v1")
                .nextTime(clock.nowUTC().plusSeconds(30))
                .schedule(IntervalJobSchedule.builder()
                        .interval(1)
                        .unit(TimeUnit.SECONDS)
                        .build())
                .build());

        final JobTriggerDto trigger4 = dbJobTriggerService.create(JobTriggerDto.Builder.create(clock)
                .jobDefinitionId("abc-123")
                .jobDefinitionType("event-processor-execution-v1")
                .startTime(clock.nowUTC().plusSeconds(60))
                .nextTime(clock.nowUTC().plusSeconds(30))
                .schedule(IntervalJobSchedule.builder()
                        .interval(1)
                        .unit(TimeUnit.SECONDS)
                        .build())
                .build());

        // First try is empty because the next time of the trigger is in the future
        assertThat(dbJobTriggerService.nextRunnableTrigger()).isEmpty();

        clock.plus(20, TimeUnit.SECONDS);

        // Now the next time is in the past and we must get a locked trigger
        assertNextTrigger(dbJobTriggerService.nextRunnableTrigger(), trigger2);

        // The second call should return the second trigger
        assertNextTrigger(dbJobTriggerService.nextRunnableTrigger(), trigger1);

        clock.plus(20, TimeUnit.SECONDS);

        // The third call should return the third trigger
        assertNextTrigger(dbJobTriggerService.nextRunnableTrigger(), trigger3);

        // No trigger left because trigger4's startTime is still in the future
        assertThat(dbJobTriggerService.nextRunnableTrigger()).isEmpty();

        clock.plus(20, TimeUnit.SECONDS);

        // The fourth call should return the third trigger
        assertNextTrigger(dbJobTriggerService.nextRunnableTrigger(), trigger4);

        // No trigger left
        assertThat(dbJobTriggerService.nextRunnableTrigger()).isEmpty();
    }