@Test
public void testUpdate() throws Exception {

	final Time timeout = Time.seconds(10L);
	// ========= setup TaskManager =================================================================================
	JobID jobID = new JobID();
	ResourceID tmRID = ResourceID.generate();
	// ========= setup JobManager ==================================================================================
	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));
	// ========= setup QueryServices ================================================================================
	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));
	// ========= start MetricFetcher testing =======================================================================
	MetricFetcher fetcher = new MetricFetcher(
		retriever,
		queryServiceRetriever,
		Executors.directExecutor(),
		timeout);
	// verify that update fetches metrics and updates the store
	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"));
	}
}