protected CompletableFuture<Void> scheduleJob(final Instant runAt, @Nullable final byte[] jobData) {
    return Mono.fromFuture(() -> scheduleJob(buildRunAtAttribute(runAt), runAt.plus(jobExpiration), jobData))
        .retryWhen(Retry.backoff(8, Duration.ofSeconds(1)).maxBackoff(Duration.ofSeconds(4)))
        .toFuture()
        .thenRun(() -> Metrics.counter(SCHEDULE_JOB_COUNTER_NAME, SCHEDULER_NAME_TAG, getSchedulerName()).increment());
  }
@Test
  void scheduleJob() {
    final TestJobScheduler scheduler = new TestJobScheduler(DYNAMO_DB_EXTENSION.getDynamoDbAsyncClient(),
        DynamoDbExtensionSchema.Tables.SCHEDULED_JOBS.tableName(),
        Clock.fixed(CURRENT_TIME, ZoneId.systemDefault()));

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

    final CompletionException completionException = assertThrows(CompletionException.class, () ->
        scheduler.scheduleJob(scheduler.buildRunAtAttribute(CURRENT_TIME, 0L), CURRENT_TIME, null).join(),
        "Scheduling multiple jobs with identical sort keys should fail");

    assertInstanceOf(ConditionalCheckFailedException.class, completionException.getCause());
  }