class DeploymentMetricsMaintainer extends ControllerMaintainer {

    @Override
    protected boolean maintain() {
        AtomicInteger failures = new AtomicInteger(0);
        AtomicInteger attempts = new AtomicInteger(0);
        AtomicReference<Exception> lastException = new AtomicReference<>(null);

        // Run parallel stream inside a custom ForkJoinPool so that we can control the number of threads used
        ForkJoinPool pool = new ForkJoinPool(applicationsToUpdateInParallel);
        pool.submit(() ->
            applications.readable().parallelStream().forEach(application -> {
                for (Instance instance : application.instances().values())
                    for (Deployment deployment : instance.deployments().values()) {
                        attempts.incrementAndGet();
                        try {
                            if (deployment.version().getMajor() < 7) continue;
                            var collectedMetrics = controller().metrics().getDeploymentMetrics(instance.id(), deployment.zone());
                            var now = controller().clock().instant();
                            applications.lockApplicationIfPresent(application.id(), locked -> {
                                Deployment existingDeployment = locked.get().require(instance.name()).deployments().get(deployment.zone());
                                if (existingDeployment == null) return; // Deployment removed since we started collecting metrics
                                DeploymentMetrics newMetrics = existingDeployment.metrics()
                                                                                 .withQueriesPerSecond(collectedMetrics.queriesPerSecond())
                                                                                 .withWritesPerSecond(collectedMetrics.writesPerSecond())
                                                                                 .withDocumentCount(collectedMetrics.documentCount())
                                                                                 .withQueryLatencyMillis(collectedMetrics.queryLatencyMillis())
                                                                                 .withWriteLatencyMillis(collectedMetrics.writeLatencyMillis())
                                                                                 .at(now);
                                applications.store(locked.with(instance.name(),
                                                               lockedInstance -> lockedInstance.with(existingDeployment.zone(), newMetrics)
                                                                                               .recordActivityAt(now, existingDeployment.zone())));

                            });
                        } catch (Exception e) {
                            failures.incrementAndGet();
                            lastException.set(e);
                        }
                    }
            })
        );
        pool.shutdown();
        try {
            pool.awaitTermination(30, TimeUnit.MINUTES);
            if (lastException.get() != null) {
                log.log(Level.WARNING,
                        String.format("Failed to gather metrics for %d/%d applications. Retrying in %s. Last error: %s",
                                      failures.get(),
                                      attempts.get(),
                                      interval(),
                                      Exceptions.toMessageString(lastException.get())));
            }
        } catch (InterruptedException e) {
            throw new RuntimeException(e);
        }
        return lastException.get() == null;
    }

    public  DeploymentMetricsMaintainer(Controller controller, Duration duration);

}