class AbstractAggregatingMetricsHandler extends AbstractJsonRequestHandler {

	@Override
	public CompletableFuture<String> handleJsonRequest(Map<String, String> pathParams, Map<String, String> queryParams, JobManagerGateway jobManagerGateway) {
		return CompletableFuture.supplyAsync(
			() -> {
				try {
					fetcher.update();
					String requestedMetricsList = queryParams.get(AbstractMetricsHandler.PARAMETER_METRICS);
					String aggTypeList = queryParams.get(PARAMETER_AGGREGATION);
					MetricStore store = fetcher.getMetricStore();

					Collection<? extends MetricStore.ComponentMetricStore> stores = getStores(store, pathParams, queryParams);
					if (stores == null){
						return "[]";
					}

					if (requestedMetricsList == null) {
						Collection<String> list = getAvailableMetrics(stores);
						return mapMetricListToJson(list);
					}

					if (requestedMetricsList.isEmpty()) {
						/*
						 * The WebInterface doesn't check whether the list of available metrics was empty. This can lead to a
						 * request for which the "get" parameter is an empty string.
						 */
						return "[]";
					}

					String[] requestedMetrics = requestedMetricsList.split(",");

					List<DoubleAccumulator.DoubleAccumulatorFactory<?>> requestedAggregationsFactories = new ArrayList<>();
					// by default we return all aggregations
					if (aggTypeList == null || aggTypeList.isEmpty()) {
						requestedAggregationsFactories.add(DoubleAccumulator.DoubleMinimumFactory.get());
						requestedAggregationsFactories.add(DoubleAccumulator.DoubleMaximumFactory.get());
						requestedAggregationsFactories.add(DoubleAccumulator.DoubleSumFactory.get());
						requestedAggregationsFactories.add(DoubleAccumulator.DoubleAverageFactory.get());
					} else {
						for (String aggregation : aggTypeList.split(",")) {
							switch (aggregation.toLowerCase()) {
								case DoubleAccumulator.DoubleMinimum.NAME:
									requestedAggregationsFactories.add(DoubleAccumulator.DoubleMinimumFactory.get());
									break;
								case DoubleAccumulator.DoubleMaximum.NAME:
									requestedAggregationsFactories.add(DoubleAccumulator.DoubleMaximumFactory.get());
									break;
								case DoubleAccumulator.DoubleSum.NAME:
									requestedAggregationsFactories.add(DoubleAccumulator.DoubleSumFactory.get());
									break;
								case DoubleAccumulator.DoubleAverage.NAME:
									requestedAggregationsFactories.add(DoubleAccumulator.DoubleAverageFactory.get());
									break;
								default:
									log.warn("Invalid aggregation specified: {}", aggregation.toLowerCase());
							}
						}
					}

					return getAggregatedMetricValues(stores, requestedMetrics, requestedAggregationsFactories);
				} catch (Exception e) {
					throw new CompletionException(new FlinkException("Could not retrieve metrics.", e));
				}
			},
			executor);
	}

	  AbstractAggregatingMetricsHandler(Executor executor, MetricFetcher fetcher);

}