class MetricStore {

	@VisibleForTesting
	public void add(MetricDump metric) {
		try {
			QueryScopeInfo info = metric.scopeInfo;
			TaskManagerMetricStore tm;
			JobMetricStore job;
			TaskMetricStore task;
			ComponentMetricStore subtask;

			String name = info.scope.isEmpty()
				? metric.name
				: info.scope + "." + metric.name;

			if (name.isEmpty()) { // malformed transmission
				return;
			}

			switch (info.getCategory()) {
				case INFO_CATEGORY_JM:
					addMetric(jobManager.metrics, name, metric);
					break;
				case INFO_CATEGORY_TM:
					String tmID = ((QueryScopeInfo.TaskManagerQueryScopeInfo) info).taskManagerID;
					tm = taskManagers.computeIfAbsent(tmID, k -> new TaskManagerMetricStore());
					if (name.contains("GarbageCollector")) {
						String gcName = name.substring("Status.JVM.GarbageCollector.".length(), name.lastIndexOf('.'));
						tm.addGarbageCollectorName(gcName);
					}
					addMetric(tm.metrics, name, metric);
					break;
				case INFO_CATEGORY_JOB:
					QueryScopeInfo.JobQueryScopeInfo jobInfo = (QueryScopeInfo.JobQueryScopeInfo) info;
					job = jobs.computeIfAbsent(jobInfo.jobID, k -> new JobMetricStore());
					addMetric(job.metrics, name, metric);
					break;
				case INFO_CATEGORY_TASK:
					QueryScopeInfo.TaskQueryScopeInfo taskInfo = (QueryScopeInfo.TaskQueryScopeInfo) info;
					job = jobs.computeIfAbsent(taskInfo.jobID, k -> new JobMetricStore());
					task = job.tasks.computeIfAbsent(taskInfo.vertexID, k -> new TaskMetricStore());
					subtask = task.subtasks.computeIfAbsent(taskInfo.subtaskIndex, k -> new ComponentMetricStore());
					/**
					 * The duplication is intended. Metrics scoped by subtask are useful for several job/task handlers,
					 * while the WebInterface task metric queries currently do not account for subtasks, so we don't
					 * divide by subtask and instead use the concatenation of subtask index and metric name as the name
					 * for those.
					 */
					addMetric(subtask.metrics, name, metric);
					addMetric(task.metrics, taskInfo.subtaskIndex + "." + name, metric);
					break;
				case INFO_CATEGORY_OPERATOR:
					QueryScopeInfo.OperatorQueryScopeInfo operatorInfo = (QueryScopeInfo.OperatorQueryScopeInfo) info;
					job = jobs.computeIfAbsent(operatorInfo.jobID, k -> new JobMetricStore());
					task = job.tasks.computeIfAbsent(operatorInfo.vertexID, k -> new TaskMetricStore());
					subtask = task.subtasks.computeIfAbsent(operatorInfo.subtaskIndex, k -> new ComponentMetricStore());
					/**
					 * As the WebInterface does not account for operators (because it can't) we don't
					 * divide by operator and instead use the concatenation of subtask index, operator name and metric name
					 * as the name.
					 */
					addMetric(subtask.metrics, operatorInfo.operatorName + "." + name, metric);
					addMetric(task.metrics, operatorInfo.subtaskIndex + "." + operatorInfo.operatorName + "." + name, metric);
					break;
				default:
					LOG.debug("Invalid metric dump category: " + info.getCategory());
			}
		} catch (Exception e) {
			LOG.debug("Malformed metric dump.", e);
		}
	}

}