public CompletableFuture<Void> processAvailableJobs() {
    return Flux.from(dynamoDbAsyncClient.queryPaginator(QueryRequest.builder()
            .tableName(tableName)
            .keyConditionExpression("#schedulerName = :schedulerName AND #runAt <= :maxRunAt")
            .expressionAttributeNames(Map.of(
                "#schedulerName", KEY_SCHEDULER_NAME,
                "#runAt", ATTR_RUN_AT))
            .expressionAttributeValues(Map.of(
                ":schedulerName", AttributeValue.fromS(getSchedulerName()),
                ":maxRunAt", buildMaxRunAtAttribute(clock.instant())))
            .build())
        .items())
        .flatMap(item -> {
          final byte[] jobData = item.containsKey(ATTR_JOB_DATA)
              ? item.get(ATTR_JOB_DATA).b().asByteArray()
              : null;

          return Mono.fromFuture(processJob(jobData))
              .doOnNext(outcome -> Metrics.counter(PROCESS_JOB_COUNTER_NAME,
                      SCHEDULER_NAME_TAG, getSchedulerName(),
                      OUTCOME_TAG, outcome)
                  .increment())
              .then(Mono.fromFuture(() -> deleteJob(item.get(KEY_SCHEDULER_NAME), item.get(ATTR_RUN_AT))))
              .onErrorResume(throwable -> {
                logger.warn("Failed to process job", throwable);
                return Mono.empty();
              });
        }, MAX_CONCURRENCY)
        .then()
        .toFuture();
  }
@Test
  void processAvailableJobs() {
    final TestClock testClock = TestClock.pinned(CURRENT_TIME);

    final TestJobScheduler scheduler = new TestJobScheduler(DYNAMO_DB_EXTENSION.getDynamoDbAsyncClient(),
        DynamoDbExtensionSchema.Tables.SCHEDULED_JOBS.tableName(),
        testClock);

    scheduler.scheduleJob(scheduler.buildRunAtAttribute(CURRENT_TIME, 0L), CURRENT_TIME, null).join();

    // Clock time is before scheduled job time
    testClock.pin(CURRENT_TIME.minusMillis(1));

    scheduler.processAvailableJobs().join();
    assertEquals(0, scheduler.jobsProcessed.get());

    // Clock time is after scheduled job time
    testClock.pin(CURRENT_TIME.plusMillis(1));

    scheduler.processAvailableJobs().join();
    assertEquals(1, scheduler.jobsProcessed.get());

    scheduler.processAvailableJobs().join();
    assertEquals(1, scheduler.jobsProcessed.get(),
        "Jobs should be cleared after successful processing; job counter should not increment on second run");
  }