TaskTimeout extends AbstractIdleService implements EventSubscriber { @Subscribe public void recordStateChange(TaskStateChange change) { if (isTransient(change.getNewState())) { executor.execute( new TimedOutTaskHandler(change.getTaskId(), change.getNewState()), timeout); } } @Inject  TaskTimeout(
      @AsyncExecutor DelayExecutor executor,
      Storage storage,
      StateManager stateManager,
      Amount<Long, Time> timeout,
      StatsProvider statsProvider); @Subscribe void recordStateChange(TaskStateChange change);  }
@Test public void testStorageStart() { expectTaskWatch(TIMEOUT); expectTaskWatch(TIMEOUT); expectTaskWatch(TIMEOUT); replayAndCreate(); clock.setNowMillis(TIMEOUT.as(Time.MILLISECONDS) * 2); for (IScheduledTask task : ImmutableList.of( makeTask("a", ASSIGNED, 0), makeTask("b", KILLING, TIMEOUT.as(Time.MILLISECONDS)), makeTask("c", PREEMPTING, clock.nowMillis() + TIMEOUT.as(Time.MILLISECONDS)))) { timeout.recordStateChange(TaskStateChange.initialized(task)); } changeState("a", ASSIGNED, RUNNING); changeState("b", KILLING, KILLED); changeState("c", PREEMPTING, FINISHED); }