class AggregatingSubtasksMetricsHandler extends AbstractAggregatingMetricsHandler {

	@Override
	protected Collection<? extends MetricStore.ComponentMetricStore> getStores(MetricStore store, Map<String, String> pathParameters, Map<String, String> queryParameters) {
		String jobID = pathParameters.get(PARAMETER_JOB_ID);
		String taskID = pathParameters.get(PARAMETER_VERTEX_ID);
		if (jobID == null) {
			return Collections.emptyList();
		}
		if (taskID == null) {
			return Collections.emptyList();
		}
		String subtasksList = queryParameters.get("subtasks");
		if (subtasksList == null || subtasksList.isEmpty()) {
			return store.getTaskMetricStore(jobID, taskID).getAllSubtaskMetricStores();
		} else {
			Iterable<Integer> subtasks = getIntegerRangeFromString(subtasksList);
			Collection<MetricStore.ComponentMetricStore> subtaskStores = new ArrayList<>();
			for (int subtask : subtasks) {
				subtaskStores.add(store.getSubtaskMetricStore(jobID, taskID, subtask));
			}
			return subtaskStores;
		}
	}

	public  AggregatingSubtasksMetricsHandler(Executor executor, MetricFetcher fetcher);

}