PeriodicApplicationMaintainer extends ApplicationMaintainer { @Override protected Set<ApplicationId> applicationsNeedingMaintenance() { if (deployer().bootstrapping()) return Set.of(); Map<ApplicationId, Instant> deploymentTimes = nodesNeedingMaintenance().stream() .map(node -> node.allocation().get().owner()) .distinct() .filter(this::canDeployNow) .collect(Collectors.toMap(Function.identity(), this::getLastDeployTime)); return deploymentTimes.entrySet().stream() .sorted(Map.Entry.comparingByValue()) .map(Map.Entry::getKey) .filter(id -> shouldMaintain(id)) .collect(Collectors.toCollection(LinkedHashSet::new)); } PeriodicApplicationMaintainer(Deployer deployer, Metric metric, NodeRepository nodeRepository,
                                  Duration interval, Duration minTimeBetweenRedeployments, FlagSource flagSource);   }
@Test(timeout = 60_000) public void queues_all_eligible_applications_for_deployment() throws Exception { fixture.activate(); clock.advance(Duration.ofMinutes(30).plus(Duration.ofSeconds(1))); fixture.setBootstrapping(false); fixture.deployer.lock().lockInterruptibly(); try { assertEquals(2, fixture.maintainer.applicationsNeedingMaintenance().size()); fixture.runApplicationMaintainer(false); assertEquals(2, fixture.maintainer.pendingDeployments()); clock.advance(Duration.ofMinutes(30).plus(Duration.ofSeconds(1))); fixture.runApplicationMaintainer(false); assertEquals(2, fixture.maintainer.pendingDeployments()); fixture.deployer.lock().unlock(); fixture.runApplicationMaintainer(); Instant deployTime = clock.instant(); assertEquals(deployTime, fixture.deployer.lastDeployTime(fixture.app1).get()); assertEquals(deployTime, fixture.deployer.lastDeployTime(fixture.app2).get()); clock.advance(Duration.ofMinutes(5)); assertEquals(0, fixture.maintainer.applicationsNeedingMaintenance().size()); } finally { if (fixture.deployer.lock().isHeldByCurrentThread()) { fixture.deployer.lock().unlock(); } } }