TaskHistoryPruner implements EventSubscriber { @Subscribe public void recordStateChange(TaskStateChange change) { if (Tasks.isTerminated(change.getNewState())) { long timeoutBasis = change.isTransition() ? clock.nowMillis() : Iterables.getLast(change.getTask().getTaskEvents()).getTimestamp(); registerInactiveTask( Tasks.getJob(change.getTask()), change.getTaskId(), calculateTimeout(timeoutBasis)); } } @Inject  TaskHistoryPruner(
      @AsyncExecutor DelayExecutor executor,
      StateManager stateManager,
      Clock clock,
      HistoryPrunnerSettings settings,
      Storage storage,
      Lifecycle lifecycle); @Subscribe void recordStateChange(TaskStateChange change);  }
@Test public void testNoPruning() { long taskATimestamp = clock.nowMillis(); IScheduledTask a = makeTask("a", FINISHED); clock.advance(ONE_MS); long taskBTimestamp = clock.nowMillis(); IScheduledTask b = makeTask("b", LOST); expectNoImmediatePrune(ImmutableSet.of(a)); expectOneDelayedPrune(taskATimestamp); expectNoImmediatePrune(ImmutableSet.of(a, b)); expectOneDelayedPrune(taskBTimestamp); control.replay(); pruner.recordStateChange(TaskStateChange.initialized(a)); pruner.recordStateChange(TaskStateChange.initialized(b)); }
@Test public void testStorageStartedWithPruning() { long taskATimestamp = clock.nowMillis(); IScheduledTask a = makeTask("a", FINISHED); clock.advance(ONE_MINUTE); long taskBTimestamp = clock.nowMillis(); IScheduledTask b = makeTask("b", LOST); clock.advance(ONE_MINUTE); long taskCTimestamp = clock.nowMillis(); IScheduledTask c = makeTask("c", FINISHED); clock.advance(ONE_MINUTE); IScheduledTask d = makeTask("d", FINISHED); IScheduledTask e = makeTask(JobKeys.from("role", "env", "job-x"), "e", FINISHED); expectNoImmediatePrune(ImmutableSet.of(a)); expectOneDelayedPrune(taskATimestamp); expectNoImmediatePrune(ImmutableSet.of(a, b)); expectOneDelayedPrune(taskBTimestamp); expectImmediatePrune(ImmutableSet.of(a, b, c), a); expectOneDelayedPrune(taskCTimestamp); expectImmediatePrune(ImmutableSet.of(b, c, d), b); expectDefaultDelayedPrune(); expectNoImmediatePrune(ImmutableSet.of(e)); expectDefaultDelayedPrune(); control.replay(); for (IScheduledTask task : ImmutableList.of(a, b, c, d, e)) { pruner.recordStateChange(TaskStateChange.initialized(task)); } }