@Test
public void testDispatchOverdue() throws Exception {

  DyneinJobSpec jobSpec = getTestJobSpec(validToken, "test7");
  Schedule schedule = getSchedule(jobSpec, false);
  Schedule scheduleAcquired = schedule.withStatus(JobStatus.ACQUIRED);
  List<Schedule> items = asList(schedule, schedule);
  String partition = "1";
  SchedulesQueryResponse response = SchedulesQueryResponse.of(items, false);
  when(scheduleManager.getOverdueJobs(partition))
      .thenReturn(CompletableFuture.completedFuture(response));
  when(asyncClient.add(schedule.getJobSpec(), "test7"))
      .thenReturn(CompletableFuture.completedFuture(null));
  when(scheduleManager.deleteDispatchedJob(scheduleAcquired))
      .thenReturn(CompletableFuture.completedFuture(null));
  CompletableFuture<Boolean> ret = worker.dispatchOverdue(partition);
  Assert.assertEquals(
      ret.get(1000, TimeUnit.MILLISECONDS), response.isShouldImmediatelyQueryAgain());
  verify(scheduleManager, times(1)).getOverdueJobs(partition);
  verify(scheduleManager, times(2))
      .updateStatus(schedule, Schedule.JobStatus.SCHEDULED, Schedule.JobStatus.ACQUIRED);
  verify(asyncClient, times(2)).add(schedule.getJobSpec(), "test7");
  verify(scheduleManager, times(2)).deleteDispatchedJob(scheduleAcquired);
  verifyNoMoreInteractions(scheduleManager, asyncClient);
}