TaskGroups implements EventSubscriber { @Subscribe public synchronized void taskChangedState(TaskStateChange stateChange) { if (stateChange.getNewState() == PENDING) { IScheduledTask task = stateChange.getTask(); TaskGroupKey key = TaskGroupKey.from(task.getAssignedTask().getTask()); TaskGroup newGroup = new TaskGroup(key, Tasks.id(task)); TaskGroup existing = groups.putIfAbsent(key, newGroup); if (existing == null) { long penaltyMs; if (stateChange.isTransition()) { penaltyMs = firstScheduleDelay; } else { penaltyMs = rescheduleCalculator.getStartupScheduleDelayMs(task); } newGroup.setPenaltyMs(penaltyMs); startGroup(newGroup); } else { existing.offer(Tasks.id(task)); } } } @Inject  TaskGroups(
      @AsyncExecutor DelayExecutor executor,
      TaskGroupsSettings settings,
      TaskScheduler taskScheduler,
      RescheduleCalculator rescheduleCalculator); @Subscribe synchronized void taskChangedState(TaskStateChange stateChange); @Subscribe synchronized void tasksDeleted(TasksDeleted deleted); Iterable<TaskGroup> getGroups();  }
@Test public void testEvaluatedAfterFirstSchedulePenalty() { expect(rateLimiter.acquire()).andReturn(0D); expect(taskScheduler.schedule(TASK_A_ID)).andReturn(true); control.replay(); taskGroups.taskChangedState(TaskStateChange.transition(makeTask(TASK_A_ID), INIT)); clock.advance(FIRST_SCHEDULE_DELAY); }
@Test public void testEvaluatedOnStartup() { expect(rateLimiter.acquire()).andReturn(0D); expect(rescheduleCalculator.getStartupScheduleDelayMs(makeTask(TASK_A_ID))).andReturn(1L); expect(taskScheduler.schedule(TASK_A_ID)).andReturn(true); control.replay(); taskGroups.taskChangedState(TaskStateChange.initialized(makeTask(TASK_A_ID))); clock.advance(FIRST_SCHEDULE_DELAY); clock.advance(RESCHEDULE_DELAY); }
@Test public void testResistStarvation() { expect(rateLimiter.acquire()).andReturn(0D).times(2); expect(taskScheduler.schedule("a0")).andReturn(true); expect(taskScheduler.schedule("b0")).andReturn(true); control.replay(); taskGroups.taskChangedState(TaskStateChange.transition(makeTask(JOB_A, "a0", 0), INIT)); taskGroups.taskChangedState(TaskStateChange.transition(makeTask(JOB_A, "a1", 1), INIT)); taskGroups.taskChangedState(TaskStateChange.transition(makeTask(JOB_A, "a2", 2), INIT)); taskGroups.taskChangedState(TaskStateChange.transition( makeTask(IJobKey.build(JOB_A.newBuilder().setName("jobB")), "b0", 0), INIT)); clock.advance(FIRST_SCHEDULE_DELAY); }
@Test public void testNonPendingIgnored() { control.replay(); IScheduledTask task = IScheduledTask.build(makeTask(TASK_A_ID).newBuilder().setStatus(ASSIGNED)); taskGroups.taskChangedState(TaskStateChange.initialized(task)); }