JobState { public void forEachTimedOutEntry( final long upperBound, final BiFunction<Long, JobRecord, Boolean> callback) { deadlinesColumnFamily.whileTrue( (compositeKey, zbNil) -> { final long deadline = compositeKey.getFirst().getValue(); final boolean isDue = deadline < upperBound; if (isDue) { final long jobKey = compositeKey.getSecond().getValue(); return visitJob(jobKey, callback, () -> deadlinesColumnFamily.delete(compositeKey)); } return false; }); } JobState(
      final ZeebeDb<ZbColumnFamilies> zeebeDb, final DbContext dbContext, final int partitionId); void create(final long key, final JobRecord record); void activate(final long key, final JobRecord record); void timeout(final long key, final JobRecord record); void complete(final long key, final JobRecord record); void cancel(final long key, final JobRecord record); void disable(final long key, final JobRecord record); void throwError(final long key, final JobRecord updatedValue); void delete(final long key, final JobRecord record); void fail(final long key, final JobRecord updatedValue); void resolve(final long key, final JobRecord updatedValue); void forEachTimedOutEntry(
      final long upperBound, final BiFunction<Long, JobRecord, Boolean> callback); boolean exists(final long jobKey); State getState(final long key); boolean isInState(final long key, final State state); void forEachActivatableJobs(
      final DirectBuffer type, final BiFunction<Long, JobRecord, Boolean> callback); JobRecord updateJobRetries(final long jobKey, final int retries); JobRecord getJob(final long key); void setJobsAvailableCallback(final Consumer<String> onJobsAvailableCallback);  }
@Test public void shouldOnlyIterateOverTimedoutWhileTrue() { createAndActivateJobRecord(1, newJobRecord().setDeadline(1L)); createAndActivateJobRecord(2, newJobRecord().setDeadline(256L)); createAndActivateJobRecord(3, newJobRecord().setDeadline(512L)); createAndActivateJobRecord(4, newJobRecord().setDeadline(65536L)); createAndActivateJobRecord(5, newJobRecord().setDeadline(4294967296L)); final List<Long> timedOutKeys = new ArrayList<>(); final long since = 65536L; jobState.forEachTimedOutEntry( since, (k, e) -> { timedOutKeys.add(k); return k.longValue() < 3; }); assertThat(timedOutKeys).hasSize(3); assertThat(timedOutKeys).containsExactly(1L, 2L, 3L); }