@Test public void testJobBatchSizeUpdate() { 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 startDelay = Duration.ofMillis(10); Duration interBatchDelay = Duration.ofMillis(5); AtomicReference<BatchJobExecutorService.BatchJob<Integer>> job = new AtomicReference<>(); job.set(batchExecutor.startJob((batchSize) -> { if (batchSize == 42) { job.get().setBatchSizes(10); } else if (batchSize == 10) { return true; } return false; }, 42, startDelay, interBatchDelay)); verify(scheduledExecutor, times(1)).schedule(isA(Runnable.class), eq(startDelay.toNanos()), eq(TimeUnit.NANOSECONDS)); runnableToInvoke.get().run(); verify(scheduledExecutor, times(1)).schedule(isA(Runnable.class), eq(interBatchDelay.toNanos()), eq(TimeUnit.NANOSECONDS)); runnableToInvoke.get().run(); verify(scheduledExecutor, times(2)).schedule(isA(Runnable.class), anyLong(), any()); }
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);  }