ScalingSweepTaskScheduler implements Closeable { void start(int initialThreads) { for (int i = 0; i < initialThreads; i++) { increaseNumberOfTasks(INITIAL_DELAY); } } ScalingSweepTaskScheduler(
            ScheduledExecutorService executorService,
            Clock clock,
            SweepDelay delay,
            Duration coolDown,
            Callable<SweepIterationResult> singleIteration,
            BooleanSupplier scalingEnabled); static ScalingSweepTaskScheduler createStarted(
            SweepDelay delay,
            int initialThreads,
            Callable<SweepIterationResult> task,
            BooleanSupplier scalingEnabled); @Override void close();  }
@Test public void withExpectedEntriesKeepReschedulingAfterDelay() throws Exception { when(sweepIteration.call()).thenReturn(SUCCESS_MEDIUM); scheduler.start(2); runSweepIterations(5); tickClock(); runSweepIterations(5); verify(sweepIteration, times(10 * 2)).call(); }
@Test public void whenScalingDisabledUseInitialPause() throws Exception { when(sweepIteration.call()).thenReturn(SUCCESS_MEDIUM); scheduler.start(2); runSweepIterations(5); schedulerEnabled.set(false); runSweepIterations(5, INITIAL_PAUSE); verify(sweepIteration, times(20)).call(); }
@Test public void noTasksAreSpawnedBeforeCoolDownPasses() throws Exception { when(sweepIteration.call()).thenReturn(SUCCESS_LARGE); scheduler.start(1); runSweepIterations(10); verify(sweepIteration, times(10)).call(); }
@Test public void withManyEntriesSpawnOneTaskEachTimeCoolDownPasses() throws Exception { when(sweepIteration.call()).thenReturn(SUCCESS_LARGE); scheduler.start(1); tickClock(); runSweepIterations(1 + 4); tickClockForHalfCoolDown(); runSweepIterations(5); tickClockForHalfCoolDown(); runSweepIterations(1 + 4); verify(sweepIteration, times(1 + 10 * 2 + 4 * 3)).call(); }
@Test public void doNotExceedMaximumParallelism() throws Exception { when(sweepIteration.call()).thenReturn(SUCCESS_LARGE); scheduler.start(ScalingSweepTaskScheduler.MAX_PARALLELISM); tickClock(); runSweepIterations(5); tickClock(); runSweepIterations(5); verify(sweepIteration, times(10 * MAX_PARALLELISM)).call(); }
@Test public void tasksNotReducedBeforeCoolDownPasses() throws Exception { when(sweepIteration.call()).thenReturn(SUCCESS_SMALL); scheduler.start(10); runSweepIterations(10); verify(sweepIteration, times(10 * 10)).call(); }
@Test public void withFewEntriesReduceByOneTaskEachTimeCoolDownPasses() throws Exception { when(sweepIteration.call()).thenReturn(SUCCESS_SMALL); scheduler.start(10); tickClock(); runSweepIterations(1 + 4); tickClockForHalfCoolDown(); runSweepIterations(5); tickClockForHalfCoolDown(); runSweepIterations(1 + 4); verify(sweepIteration, times(10 + 10 * 9 + 4 * 8)).call(); }
@Test public void doNotReduceToZeroTasks() throws Exception { when(sweepIteration.call()).thenReturn(SUCCESS_SMALL); scheduler.start(1); tickClock(); runSweepIterations(5); tickClock(); runSweepIterations(5); verify(sweepIteration, times(10)).call(); }
@Test public void increaseCanPreventReduction() throws Exception { when(sweepIteration.call()).thenReturn( SUCCESS_LARGE, SUCCESS_LARGE, SUCCESS_SMALL); scheduler.start(2); tickClock(); runSweepIterations(1 + 4); tickClockForHalfCoolDown(); runSweepIterations(5); tickClockForHalfCoolDown(); runSweepIterations(1 + 4); verify(sweepIteration, times(1 * 2 + 10 * 3 + 4 * 2)).call(); }
@Test public void attemptedIncreaseCanPreventReduction() throws Exception { when(sweepIteration.call()).thenReturn( SUCCESS_LARGE, SUCCESS_LARGE, SUCCESS_SMALL); scheduler.start(2); tickClockForHalfCoolDown(); runSweepIterations(5); tickClockForHalfCoolDown(); runSweepIterations(5); tickClockForHalfCoolDown(); runSweepIterations(1 + 4); verify(sweepIteration, times(11 * 2 + 4 * 1)).call(); }
@Test public void reductionCanPreventIncrease() throws Exception { when(sweepIteration.call()).thenReturn( SUCCESS_SMALL, SUCCESS_SMALL, SUCCESS_LARGE); scheduler.start(2); tickClock(); runSweepIterations(1 + 4); tickClockForHalfCoolDown(); runSweepIterations(5); tickClockForHalfCoolDown(); runSweepIterations(1 + 4); verify(sweepIteration, times(1 * 2 + 10 * 1 + 4 * 2)).call(); }
@Test public void attemptedReductionDoesNotPreventIncrease() throws Exception { when(sweepIteration.call()).thenReturn( SUCCESS_SMALL, SUCCESS_SMALL, SUCCESS_LARGE); scheduler.start(2); tickClockForHalfCoolDown(); runSweepIterations(5); tickClockForHalfCoolDown(); runSweepIterations(1 + 4); tickClockForHalfCoolDown(); runSweepIterations(5); verify(sweepIteration, times(6 * 2 + 9 * 3)).call(); }
@Test public void whenUnableToAcquireShardReduceTasksIgnoringCoolDown() throws Exception { when(sweepIteration.call()).thenReturn( SUCCESS_SMALL, SUCCESS_MEDIUM, SUCCESS_LARGE, SweepIterationResults.unableToAcquireShard()); scheduler.start(3 + 7); runSweepIterations(1 + 1); verify(sweepIteration, times(10 + 3)).call(); }
@Test public void whenUnableToAcquireShardOnLastTaskRescheduleAfterMaxPause() throws Exception { when(sweepIteration.call()).thenReturn(SweepIterationResults.unableToAcquireShard()); schedulerWithDelay.start(10); runSweepIterations(1 + 9, SweepDelay.DEFAULT_MAX_PAUSE_MILLIS); verify(sweepIteration, times(10 + 9 * 1)).call(); }
@Test public void whenInsufficientConsistencyRescheduleAfterBackoff() throws Exception { when(sweepIteration.call()).thenReturn( SweepIterationResults.insufficientConsistency(), SweepIterationResults.insufficientConsistency(), SUCCESS_MEDIUM); schedulerWithDelay.start(2); runSweepIterations(2, SweepDelay.BACKOFF); runSweepIterations(5, DELAY); verify(sweepIteration, times(7 * 2)).call(); }
@Test public void whenDisabledReduceTasksIgnoringCoolDown() throws Exception { when(sweepIteration.call()).thenReturn( SUCCESS_SMALL, SUCCESS_SMALL, SUCCESS_SMALL, SweepIterationResults.disabled()); scheduler.start(3 + 7); runSweepIterations(1 + 1); verify(sweepIteration, times(10 + 3)).call(); }
@Test public void whenDisabledOnLastTaskRescheduleAfterBackoff() throws Exception { when(sweepIteration.call()).thenReturn(SweepIterationResults.disabled()); schedulerWithDelay.start(10); runSweepIterations(1 + 9, SweepDelay.BACKOFF); verify(sweepIteration, times(10 + 9 * 1)).call(); }
@Test public void whenOtherErrorRescheduleAfterMaxPause() throws Exception { when(sweepIteration.call()).thenReturn( SweepIterationResults.otherError(), SweepIterationResults.otherError(), SUCCESS_MEDIUM); schedulerWithDelay.start(2); runSweepIterations(2, SweepDelay.DEFAULT_MAX_PAUSE_MILLIS); runSweepIterations(5, DELAY); verify(sweepIteration, times(7 * 2)).call(); }
@Test public void whenVeryFewEntriesIncreasePause() throws Exception { SweepDelay sweepDelay = new SweepDelay(100L); ScalingSweepTaskScheduler schedulerWithRealDelay = createScheduler(sweepDelay); when(sweepIteration.call()).thenReturn(SUCCESS_TINY); schedulerWithRealDelay.start(3); runSweepIterations(10, 100L); verify(sweepIteration, atMost(3 * 10 - 1)).call(); assertThat(sweepDelay.getNextPause(SUCCESS_MEDIUM)).isGreaterThan(100L); }
@Test public void whenVeryManyEntriesDecreasePause() throws Exception { SweepDelay sweepDelay = new SweepDelay(100L); ScalingSweepTaskScheduler schedulerWithRealDelay = createScheduler(sweepDelay); when(sweepIteration.call()).thenReturn(SUCCESS_HUGE); schedulerWithRealDelay.start(3); runSweepIterations(10, 100L); verify(sweepIteration, atLeast(3 * 10 + 1)).call(); assertThat(sweepDelay.getNextPause(SUCCESS_MEDIUM)).isLessThan(100L); }
@Test public void exceptionalIterationsDoNotAffectPause() throws Exception { SweepDelay sweepDelay = new SweepDelay(100L); ScalingSweepTaskScheduler schedulerWithRealDelay = createScheduler(sweepDelay); when(sweepIteration.call()).thenReturn( SweepIterationResults.otherError(), SweepIterationResults.unableToAcquireShard(), SweepIterationResults.otherError(), SweepIterationResults.insufficientConsistency()); schedulerWithRealDelay.start(4); runSweepIterations(1); verify(sweepIteration, times(4)).call(); assertThat(sweepDelay.getNextPause(SUCCESS_MEDIUM)).isEqualTo(100L); }