class PartitionWorker {

  CompletableFuture<Void> dispatchToDestination(Schedule schedule) {
    Stopwatch stopwatch = Stopwatch.createStarted();
    CompletableFuture<Void> ret = new CompletableFuture<>();
    DyneinJobSpec jobSpec = jobSpecTransformer.deserializeJobSpec(schedule.getJobSpec());

    asyncSqsClient
        .add(schedule.getJobSpec(), jobSpec.getQueueName())
        .whenCompleteAsync(
            (it, ex) -> {
              if (ex != null) {
                log.error(
                    "Error dispatching job {} to destination queue {}",
                    jobSpec.getJobToken(),
                    jobSpec.getQueueName(),
                    ex);
                metrics.dispatchJobError(ex, jobSpec.getQueueName());
                scheduleManager
                    .updateStatus(
                        schedule, Schedule.JobStatus.ACQUIRED, Schedule.JobStatus.SCHEDULED)
                    .whenComplete((response, exception) -> ret.completeExceptionally(ex));
              } else {
                log.info(
                    "Dispatched job {} to destination queue {}",
                    jobSpec.getJobToken(),
                    jobSpec.getQueueName());
                long time = stopwatch.elapsed(TimeUnit.NANOSECONDS);
                metrics.dispatchScheduledJob(time, jobSpec, schedule.getShardId());
                scheduleManager
                    .deleteDispatchedJob(schedule)
                    .whenComplete((response, exception) -> ret.complete(null));
              }
            },
            executorService);

    return ret;
  }

}