@Test
public void testDispatchOverdue_dispatchPartialFailure() throws Exception {

  String partition = "failure3";
  DyneinJobSpec jobSpec_1 = getTestJobSpec(validToken, "test10");
  Schedule schedule_1 = getSchedule(jobSpec_1, false);
  DyneinJobSpec jobSpec_2 = getTestJobSpec(validToken, "test10.1");
  Schedule schedule_2 = getSchedule(jobSpec_2, true);
  CompletableFuture<Void> dispatchResponse = new CompletableFuture<>();
  dispatchResponse.completeExceptionally(new Exception());
  when(scheduleManager.getOverdueJobs(partition))
      .thenReturn(
          CompletableFuture.completedFuture(
              SchedulesQueryResponse.of(asList(schedule_1, schedule_2), false)));
  when(asyncClient.add(schedule_1.getJobSpec(), "test10"))
      .thenReturn(CompletableFuture.completedFuture(null));
  when(asyncClient.add(schedule_2.getJobSpec(), "test10.1")).thenReturn(dispatchResponse);
  when(scheduleManager.updateStatus(
          schedule_2.withStatus(JobStatus.ACQUIRED),
          Schedule.JobStatus.ACQUIRED,
          Schedule.JobStatus.SCHEDULED))
      .thenReturn(CompletableFuture.completedFuture(schedule_2));
  CompletableFuture<Boolean> ret = worker.dispatchOverdue(partition);
  ret.get(1000, TimeUnit.MILLISECONDS);
  verify(scheduleManager, times(1)).getOverdueJobs(partition);
  verify(scheduleManager, times(1))
      .updateStatus(schedule_1, JobStatus.SCHEDULED, JobStatus.ACQUIRED);
  verify(scheduleManager, times(1))
      .updateStatus(schedule_2, JobStatus.SCHEDULED, JobStatus.ACQUIRED);
  verify(asyncClient, times(1)).add(schedule_1.getJobSpec(), "test10");
  verify(asyncClient, times(1)).add(schedule_2.getJobSpec(), "test10.1");
  verify(scheduleManager, times(1))
      .deleteDispatchedJob(schedule_1.withStatus(JobStatus.ACQUIRED));
  verify(scheduleManager, times(1))
      .updateStatus(
          schedule_2.withStatus(JobStatus.ACQUIRED), JobStatus.ACQUIRED, JobStatus.SCHEDULED);
  verifyNoMoreInteractions(scheduleManager, asyncClient);
}