TaskThrottler implements EventSubscriber { @Subscribe public void taskChangedState(final TaskStateChange stateChange) { if (stateChange.getNewState() == THROTTLED) { long readyAtMs = Tasks.getLatestEvent(stateChange.getTask()).getTimestamp() + rescheduleCalculator.getFlappingPenaltyMs(stateChange.getTask()); long delayMs = Math.max(0, readyAtMs - clock.nowMillis()); throttleStats.accumulate(delayMs); executor.execute( () -> storage.write(storeProvider -> stateManager.changeState( storeProvider, stateChange.getTaskId(), Optional.of(THROTTLED), PENDING, Optional.absent())), Amount.of(delayMs, Time.MILLISECONDS)); } } @Inject  TaskThrottler(
      RescheduleCalculator rescheduleCalculator,
      Clock clock,
      @AsyncExecutor DelayExecutor executor,
      Storage storage,
      StateManager stateManager); @Subscribe void taskChangedState(final TaskStateChange stateChange);  }
@Test public void testIgnoresNonThrottledTasks() { control.replay(); throttler.taskChangedState(TaskStateChange.transition(makeTask("a", PENDING), INIT)); throttler.taskChangedState(TaskStateChange.transition(makeTask("a", RUNNING), PENDING)); }
@Test public void testThrottledTask() { IScheduledTask task = makeTask("a", THROTTLED); long penaltyMs = 100; expect(rescheduleCalculator.getFlappingPenaltyMs(task)).andReturn(penaltyMs); Capture<Runnable> stateChangeCapture = expectThrottled(penaltyMs); expectMovedToPending(task); control.replay(); throttler.taskChangedState(TaskStateChange.transition(task, INIT)); stateChangeCapture.getValue().run(); }
@Test public void testThrottledTaskReady() { IScheduledTask task = makeTask("a", THROTTLED); long penaltyMs = 100; expect(rescheduleCalculator.getFlappingPenaltyMs(task)).andReturn(penaltyMs); Capture<Runnable> stateChangeCapture = expectThrottled(0); expectMovedToPending(task); control.replay(); clock.advance(Amount.of(1L, Time.HOURS)); throttler.taskChangedState(TaskStateChange.transition(task, INIT)); stateChangeCapture.getValue().run(); }