public boolean releaseTrigger(JobTriggerDto trigger, JobTriggerUpdate triggerUpdate) {
        requireNonNull(trigger, "trigger cannot be null");
        requireNonNull(triggerUpdate, "triggerUpdate cannot be null");

        final var filter = and(
                // Make sure that the owner still owns the trigger
                eq(FIELD_LOCK_OWNER, nodeId),
                idEq(getId(trigger)),
                // Only release running triggers. The trigger might have been paused while the trigger was running
                // so we don't want to set it to RUNNABLE again.
                // TODO: This is an issue. If a user set it to PAUSED, we will not unlock it. Figure something out.
                //       Maybe a manual trigger pause will set "nextStatus" if the trigger is currently running?
                //       That next status would need to be set on release.
                eq(FIELD_STATUS, JobTriggerStatus.RUNNING)
        );

        final List<Bson> updates = new ArrayList<>();
        updates.add(unset(FIELD_LOCK_OWNER));

        if (triggerUpdate.concurrencyReschedule()) {
            updates.add(inc(FIELD_CONCURRENCY_RESCHEDULE_COUNT, 1));
        } else {
            updates.add(set(FIELD_CONCURRENCY_RESCHEDULE_COUNT, 0));
        }

        // An empty next time indicates that this trigger should not be fired anymore. (e.g. for "once" schedules)
        if (triggerUpdate.nextTime().isPresent()) {
            if (triggerUpdate.status().isPresent()) {
                updates.add(set(FIELD_STATUS, triggerUpdate.status().get()));
            } else {
                updates.add(set(FIELD_STATUS, JobTriggerStatus.RUNNABLE));
            }
            updates.add(set(FIELD_NEXT_TIME, triggerUpdate.nextTime().get()));
        } else {
            updates.add(set(FIELD_STATUS, triggerUpdate.status().orElse(JobTriggerStatus.COMPLETE)));
        }

        if (triggerUpdate.data().isPresent()) {
            updates.add(set(FIELD_DATA, triggerUpdate.data()));
        }
        trigger.triggeredAt().ifPresent(triggeredAt -> {
            var duration = new org.joda.time.Duration(triggeredAt, clock.nowUTC());
            updates.add(set(FIELD_LAST_EXECUTION_DURATION, Optional.of(duration.getMillis())));
        });

        return collection.updateOne(filter, combine(updates)).getModifiedCount() == 1;
    }
@Test
    public void releaseTrigger() {
        final JobTriggerDto trigger1 = dbJobTriggerService.create(JobTriggerDto.Builder.create(clock)
                .jobDefinitionId("abc-123")
                .jobDefinitionType("event-processor-execution-v1")
                .concurrencyRescheduleCount(42)
                .schedule(IntervalJobSchedule.builder()
                        .interval(1)
                        .unit(TimeUnit.SECONDS)
                        .build())
                .build());
        final JobTriggerData newData = TestJobTriggerData.create(Collections.singletonMap("hello", "world"));
        final JobTriggerUpdate update = JobTriggerUpdate.withNextTimeAndData(clock.nowUTC().plusSeconds(20), newData);

        // Releasing the trigger should not do anything because the trigger has not been locked yet
        assertThat(dbJobTriggerService.releaseTrigger(trigger1, update)).isFalse();

        // Lock the trigger
        final Optional<JobTriggerDto> runnableTrigger = dbJobTriggerService.nextRunnableTrigger();
        assertThat(runnableTrigger).isNotEmpty();

        clock.plus(15, TimeUnit.SECONDS);

        // Releasing the trigger should work now
        assertThat(dbJobTriggerService.releaseTrigger(runnableTrigger.get(), update)).isTrue();

        assertThat(dbJobTriggerService.get(trigger1.id()))
                .isPresent()
                .get()
                .satisfies(trigger -> {
                    // Make sure the lock is gone
                    assertThat(trigger.lock().owner()).isNull();
                    assertThat(trigger.status()).isEqualTo(JobTriggerStatus.RUNNABLE);
                    assertThat(trigger.nextTime()).isEqualTo(update.nextTime().orElse(null));
                    assertThat(trigger.executionDurationMs()).isPresent().get().isEqualTo(15_000L);
                    assertThat(trigger.concurrencyRescheduleCount()).isEqualTo(0);
                    assertThat(trigger.data()).isPresent().get().satisfies(data -> {
                        assertThat(data).isInstanceOf(TestJobTriggerData.class);
                        assertThat(data).isEqualTo(TestJobTriggerData.create(Collections.singletonMap("hello", "world")));
                    });
                });

        // Releasing it again doesn't do anything
        assertThat(dbJobTriggerService.releaseTrigger(trigger1, update)).isFalse();
    }