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; }    }
@Test public void testDispatchJob() throws Exception { DyneinJobSpec jobSpec = getTestJobSpec(validToken, "test3"); Schedule schedule = getSchedule(jobSpec, false); when(asyncClient.add(schedule.getJobSpec(), "test3")) .thenReturn(CompletableFuture.completedFuture(null)); CompletableFuture<Void> ret = worker.dispatchToDestination(schedule); ret.get(1000, TimeUnit.MILLISECONDS); verify(asyncClient, times(1)).add(schedule.getJobSpec(), "test3"); verify(scheduleManager, times(1)).deleteDispatchedJob(eq(schedule)); verifyNoMoreInteractions(asyncClient, scheduleManager); }
@Test public void testDispatch_EnqueueFail() throws Exception { DyneinJobSpec jobSpec = getTestJobSpec(validToken, "test4"); Schedule schedule = getSchedule(jobSpec, false); Exception addException = new Exception(); CompletableFuture<Void> error = new CompletableFuture<>(); error.completeExceptionally(addException); when(asyncClient.add(schedule.getJobSpec(), "test4")).thenReturn(error); CompletableFuture<Void> ret = worker.dispatchToDestination(schedule); try { ret.get(1000, TimeUnit.MILLISECONDS); } catch (ExecutionException e) { assertSame(e.getCause(), addException); } verify(asyncClient, times(1)).add(schedule.getJobSpec(), "test4"); verify(scheduleManager, times(1)) .updateStatus(schedule, Schedule.JobStatus.ACQUIRED, Schedule.JobStatus.SCHEDULED); verifyNoMoreInteractions(scheduleManager, asyncClient); }
@Test public void testDispatch_DeleteFromTableFail() throws Exception { DyneinJobSpec jobSpec = getTestJobSpec(validToken, "test6"); Schedule schedule = getSchedule(jobSpec, false); CompletableFuture<Void> response = new CompletableFuture<>(); Exception exception = new Exception(); response.completeExceptionally(exception); when(asyncClient.add(schedule.getJobSpec(), "test6")) .thenReturn(CompletableFuture.completedFuture(null)); when(scheduleManager.deleteDispatchedJob(schedule)).thenReturn(response); CompletableFuture<Void> ret = worker.dispatchToDestination(schedule); ret.get(1000, TimeUnit.MILLISECONDS); verify(asyncClient, times(1)).add(schedule.getJobSpec(), "test6"); verify(scheduleManager, times(1)).deleteDispatchedJob(schedule); verifyNoMoreInteractions(scheduleManager, asyncClient); }