MetricFetcher { public void update() { synchronized (this) { long currentTime = System.currentTimeMillis(); if (currentTime - lastUpdateTime > 10000) { lastUpdateTime = currentTime; fetchMetrics(); } } } MetricFetcher(
			GatewayRetriever<T> retriever,
			MetricQueryServiceRetriever queryServiceRetriever,
			Executor executor,
			Time timeout); MetricStore getMetricStore(); void update();  }
@Test public void testUpdate() throws Exception { final Time timeout = Time.seconds(10L); JobID jobID = new JobID(); ResourceID tmRID = ResourceID.generate(); JobDetails details = mock(JobDetails.class); when(details.getJobId()).thenReturn(jobID); final String jmMetricQueryServicePath = "/jm/" + MetricQueryService.METRIC_QUERY_SERVICE_NAME; final String tmMetricQueryServicePath = "/tm/" + MetricQueryService.METRIC_QUERY_SERVICE_NAME + "_" + tmRID.getResourceIdString(); JobManagerGateway jobManagerGateway = mock(JobManagerGateway.class); when(jobManagerGateway.requestJobDetails(anyBoolean(), anyBoolean(), any(Time.class))) .thenReturn(CompletableFuture.completedFuture(new MultipleJobsDetails(Collections.emptyList(), Collections.emptyList()))); when(jobManagerGateway.requestMetricQueryServicePaths(any(Time.class))).thenReturn( CompletableFuture.completedFuture(Collections.singleton(jmMetricQueryServicePath))); when(jobManagerGateway.requestTaskManagerMetricQueryServicePaths(any(Time.class))).thenReturn( CompletableFuture.completedFuture(Collections.singleton(Tuple2.of(tmRID, tmMetricQueryServicePath)))); GatewayRetriever<JobManagerGateway> retriever = mock(AkkaJobManagerRetriever.class); when(retriever.getNow()) .thenReturn(Optional.of(jobManagerGateway)); MetricQueryServiceGateway jmQueryService = mock(MetricQueryServiceGateway.class); MetricQueryServiceGateway tmQueryService = mock(MetricQueryServiceGateway.class); MetricDumpSerialization.MetricSerializationResult requestMetricsAnswer = createRequestDumpAnswer(tmRID, jobID); when(jmQueryService.queryMetrics(any(Time.class))) .thenReturn(CompletableFuture.completedFuture(new MetricDumpSerialization.MetricSerializationResult(new byte[0], 0, 0, 0, 0))); when(tmQueryService.queryMetrics(any(Time.class))) .thenReturn(CompletableFuture.completedFuture(requestMetricsAnswer)); MetricQueryServiceRetriever queryServiceRetriever = mock(MetricQueryServiceRetriever.class); when(queryServiceRetriever.retrieveService(eq(jmMetricQueryServicePath))).thenReturn(CompletableFuture.completedFuture(jmQueryService)); when(queryServiceRetriever.retrieveService(eq(tmMetricQueryServicePath))).thenReturn(CompletableFuture.completedFuture(tmQueryService)); MetricFetcher fetcher = new MetricFetcher( retriever, queryServiceRetriever, Executors.directExecutor(), timeout); fetcher.update(); MetricStore store = fetcher.getMetricStore(); synchronized (store) { assertEquals("7", store.getJobManagerMetricStore().getMetric("abc.hist_min")); assertEquals("6", store.getJobManagerMetricStore().getMetric("abc.hist_max")); assertEquals("4.0", store.getJobManagerMetricStore().getMetric("abc.hist_mean")); assertEquals("0.5", store.getJobManagerMetricStore().getMetric("abc.hist_median")); assertEquals("5.0", store.getJobManagerMetricStore().getMetric("abc.hist_stddev")); assertEquals("0.75", store.getJobManagerMetricStore().getMetric("abc.hist_p75")); assertEquals("0.9", store.getJobManagerMetricStore().getMetric("abc.hist_p90")); assertEquals("0.95", store.getJobManagerMetricStore().getMetric("abc.hist_p95")); assertEquals("0.98", store.getJobManagerMetricStore().getMetric("abc.hist_p98")); assertEquals("0.99", store.getJobManagerMetricStore().getMetric("abc.hist_p99")); assertEquals("0.999", store.getJobManagerMetricStore().getMetric("abc.hist_p999")); assertEquals("x", store.getTaskManagerMetricStore(tmRID.toString()).metrics.get("abc.gauge")); assertEquals("5.0", store.getJobMetricStore(jobID.toString()).metrics.get("abc.jc")); assertEquals("2", store.getTaskMetricStore(jobID.toString(), "taskid").metrics.get("2.abc.tc")); assertEquals("1", store.getTaskMetricStore(jobID.toString(), "taskid").metrics.get("2.opname.abc.oc")); } }