class PartitionWorker {

  CompletableFuture<Boolean> dispatchOverdue(String partition) {
    Stopwatch stopwatch = Stopwatch.createStarted();
    CompletableFuture<Boolean> ret = new CompletableFuture<>();
    scheduleManager
        .getOverdueJobs(partition)
        .whenCompleteAsync(
            (res, ex) -> {
              if (ex != null) {
                log.error("Error querying overdue jobs for partition {}", partition, ex);
                metrics.queryOverdueError(ex, partition);
                ret.complete(true);
              } else {
                metrics.queryOverdue(partition);
                CompletableFuture.allOf(
                        res.getSchedules()
                            .stream()
                            .map(
                                schedule ->
                                    scheduleManager
                                        .updateStatus(
                                            schedule,
                                            Schedule.JobStatus.SCHEDULED,
                                            Schedule.JobStatus.ACQUIRED)
                                        .thenCompose(this::dispatchToDestination))
                            .toArray(CompletableFuture[]::new))
                    .whenCompleteAsync(
                        (it, exception) -> {
                          // explicitly don't handle any exceptions here because:
                          // - acquire errors are harmless,
                          //   either CAS exception or will be picked up by the next iteration
                          // - dispatch errors are harmless,
                          //   - dispatch to SQS failures will be set back to ACQUIRED
                          //   - set back to ACQUIRED failures will be picked up by fix stuck jobs
                          ret.complete(res.isShouldImmediatelyQueryAgain());
                          long time = stopwatch.elapsed(TimeUnit.NANOSECONDS);
                          log.debug("Dispatched overdue jobs for partition {}", partition);
                          metrics.dispatchOverdue(time, partition);
                        },
                        executorService);
              }
            },
            executorService);
    return ret;
  }

}