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