@Test public void testExceptionsDontPreventFutureBatches() { ScheduledFuture<?> future = Mockito.mock(ScheduledFuture.class); AtomicReference<Runnable> runnableToInvoke = new AtomicReference<>(); when(scheduledExecutor.schedule(isA(Runnable.class), anyLong(), any())).then((invoc) -> { runnableToInvoke.set(((Runnable) invoc.getArgument(0))); return future; }); Duration delay = Duration.ofMillis(10); AtomicLong counter = new AtomicLong(); batchExecutor.startJob((batchSize) -> { switch ((int) counter.incrementAndGet()) { case 3: return true; default: throw new Error(); } }, 42, delay, delay); verify(scheduledExecutor, times(1)).schedule(isA(Runnable.class), eq(delay.toNanos()), eq(TimeUnit.NANOSECONDS)); runnableToInvoke.get().run(); verify(scheduledExecutor, times(2)).schedule(isA(Runnable.class), eq(delay.toNanos()), eq(TimeUnit.NANOSECONDS)); runnableToInvoke.get().run(); verify(scheduledExecutor, times(3)).schedule(isA(Runnable.class), eq(delay.toNanos()), eq(TimeUnit.NANOSECONDS)); runnableToInvoke.get().run(); verify(scheduledExecutor, times(3)).schedule(isA(Runnable.class), eq(delay.toNanos()), eq(TimeUnit.NANOSECONDS)); }
public <T> BatchJob<T> startJob(BatchProcessor<T> task, T batchSizes, Duration startDelay, Duration interBatchDelay) { BatchJob<T> job = new BatchJob<>(task, batchSizes, interBatchDelay); job.scheduleNextBatch(startDelay); return job; }
BatchJobExecutorService { public <T> BatchJob<T> startJob(BatchProcessor<T> task, T batchSizes, Duration startDelay, Duration interBatchDelay) { BatchJob<T> job = new BatchJob<>(task, batchSizes, interBatchDelay); job.scheduleNextBatch(startDelay); return job; } }
BatchJobExecutorService { public <T> BatchJob<T> startJob(BatchProcessor<T> task, T batchSizes, Duration startDelay, Duration interBatchDelay) { BatchJob<T> job = new BatchJob<>(task, batchSizes, interBatchDelay); job.scheduleNextBatch(startDelay); return job; }  }
BatchJobExecutorService { public <T> BatchJob<T> startJob(BatchProcessor<T> task, T batchSizes, Duration startDelay, Duration interBatchDelay) { BatchJob<T> job = new BatchJob<>(task, batchSizes, interBatchDelay); job.scheduleNextBatch(startDelay); return job; }  BatchJob<T> startJob(BatchProcessor<T> task, T batchSizes, Duration startDelay, Duration interBatchDelay); BatchJob<T> startJob(Consumer<T> task, T batchSizes, Duration startDelay, Duration interBatchDelay); }
BatchJobExecutorService { public <T> BatchJob<T> startJob(BatchProcessor<T> task, T batchSizes, Duration startDelay, Duration interBatchDelay) { BatchJob<T> job = new BatchJob<>(task, batchSizes, interBatchDelay); job.scheduleNextBatch(startDelay); return job; }  BatchJob<T> startJob(BatchProcessor<T> task, T batchSizes, Duration startDelay, Duration interBatchDelay); BatchJob<T> startJob(Consumer<T> task, T batchSizes, Duration startDelay, Duration interBatchDelay);  }