@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); }
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) -> { 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; }
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) -> { 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; } }
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) -> { 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; }  }
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) -> { 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; }   }
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) -> { 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; }    }