ClusterDeploymentMetricsRetriever { public Map<ClusterInfo, DeploymentMetricsAggregator> requestMetricsGroupedByCluster(Collection<URI> hosts) { Map<ClusterInfo, DeploymentMetricsAggregator> clusterMetricsMap = new ConcurrentHashMap<>(); long startTime = System.currentTimeMillis(); Runnable retrieveMetricsJob = () -> hosts.parallelStream().forEach(host -> getHostMetrics(host, clusterMetricsMap) ); ForkJoinPool threadPool = new ForkJoinPool(10); threadPool.submit(retrieveMetricsJob); threadPool.shutdown(); try { threadPool.awaitTermination(1, TimeUnit.MINUTES); } catch (InterruptedException e) { throw new RuntimeException(e); } log.log(Level.FINE, () -> String.format("Metric retrieval for %d nodes took %d milliseconds", hosts.size(), System.currentTimeMillis() - startTime) ); return clusterMetricsMap; }  Map<ClusterInfo, DeploymentMetricsAggregator> requestMetricsGroupedByCluster(Collection<URI> hosts);  }
@Test public void testMetricAggregation() throws IOException { List<URI> hosts = Stream.of(1, 2, 3) .map(item -> URI.create("http: .collect(Collectors.toList()); stubFor(get(urlEqualTo("/1")) .willReturn(aResponse() .withStatus(200) .withBody(contentMetrics()))); stubFor(get(urlEqualTo("/2")) .willReturn(aResponse() .withStatus(200) .withBody(contentMetrics()))); stubFor(get(urlEqualTo("/3")) .willReturn(aResponse() .withStatus(200) .withBody(containerMetrics()))); ClusterInfo expectedContentCluster = new ClusterInfo("content_cluster_id", "content"); ClusterInfo expectedContainerCluster = new ClusterInfo("container_cluster_id", "container"); Map<ClusterInfo, DeploymentMetricsAggregator> aggregatorMap = new ClusterDeploymentMetricsRetriever().requestMetricsGroupedByCluster(hosts); compareAggregators( new DeploymentMetricsAggregator().addDocumentCount(6000.0), aggregatorMap.get(expectedContentCluster) ); compareAggregators( new DeploymentMetricsAggregator() .addContainerLatency(3000, 43) .addContainerLatency(2000, 0) .addQrLatency(3000, 43) .addFeedLatency(3000, 43), aggregatorMap.get(expectedContainerCluster) ); wireMock.stop(); }