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