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

            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 SubtaskMetricStore());

                    // The attempt is the representative one if the current execution attempt
                    // number for the subtask is not present in the currentExecutionAttempts,
                    // which means there should be only one execution
                    isRepresentativeAttempt =
                            isRepresentativeAttempt(
                                    taskInfo.jobID,
                                    taskInfo.vertexID,
                                    taskInfo.subtaskIndex,
                                    taskInfo.attemptNumber);
                    attempt =
                            subtask.attempts.computeIfAbsent(
                                    taskInfo.attemptNumber, k -> new ComponentMetricStore());
                    addMetric(attempt.metrics, name, metric);
                    // If the attempt is representative one, its metrics can be updated to the
                    // subtask and task metric store.
                    if (isRepresentativeAttempt) {
                        /**
                         * 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 SubtaskMetricStore());

                    isRepresentativeAttempt =
                            isRepresentativeAttempt(
                                    operatorInfo.jobID,
                                    operatorInfo.vertexID,
                                    operatorInfo.subtaskIndex,
                                    operatorInfo.attemptNumber);

                    attempt =
                            subtask.attempts.computeIfAbsent(
                                    operatorInfo.attemptNumber, k -> new ComponentMetricStore());
                    addMetric(attempt.metrics, operatorInfo.operatorName + "." + name, metric);

                    // If the attempt is representative one, its metrics can be updated to the
                    // subtask and task metric store.
                    if (isRepresentativeAttempt) {
                        /**
                         * 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;
                case INFO_CATEGORY_JM_OPERATOR:
                    QueryScopeInfo.JobManagerOperatorQueryScopeInfo jmOperatorInfo =
                            (QueryScopeInfo.JobManagerOperatorQueryScopeInfo) info;
                    job = jobs.computeIfAbsent(jmOperatorInfo.jobID, k -> new JobMetricStore());
                    task =
                            job.tasks.computeIfAbsent(
                                    jmOperatorInfo.vertexID, k -> new TaskMetricStore());
                    jmOperator =
                            task.jmOperator == null ? new ComponentMetricStore() : task.jmOperator;
                    addMetric(
                            jmOperator.metrics,
                            // to distinguish between operators
                            jmOperatorInfo.operatorName + "." + name,
                            metric);
                    break;
                default:
                    LOG.debug("Invalid metric dump category: " + info.getCategory());
            }
        } catch (Exception e) {
            LOG.debug("Malformed metric dump.", e);
        }
    }
@Test
    void testAdd() {
        MetricStore store = setupStore(new MetricStore());

        assertThat(store.getJobManagerMetricStore().getMetric("abc.metric1", "-1")).isEqualTo("0");
        assertThat(store.getTaskManagerMetricStore("tmid").getMetric("abc.metric2", "-1"))
                .isEqualTo("1");
        assertThat(store.getJobMetricStore(JOB_ID.toString()).getMetric("abc.metric3", "-1"))
                .isEqualTo("2");
        assertThat(store.getJobMetricStore(JOB_ID.toString()).getMetric("abc.metric4", "-1"))
                .isEqualTo("3");

        assertThat(
                        store.getTaskMetricStore(JOB_ID.toString(), "taskid")
                                .getMetric("8.abc.metric5", "-1"))
                .isEqualTo("14");
        assertThat(
                        store.getSubtaskMetricStore(JOB_ID.toString(), "taskid", 8)
                                .getMetric("abc.metric5", "-1"))
                .isEqualTo("14");
        assertThat(
                        store.getSubtaskAttemptMetricStore(JOB_ID.toString(), "taskid", 8, 1)
                                .getMetric("abc.metric5", "-1"))
                .isEqualTo("4");
        assertThat(
                        store.getSubtaskAttemptMetricStore(JOB_ID.toString(), "taskid", 8, 2)
                                .getMetric("abc.metric5", "-1"))
                .isEqualTo("14");

        assertThat(
                        store.getTaskMetricStore(JOB_ID.toString(), "taskid")
                                .getMetric("8.opname.abc.metric6", "-1"))
                .isEqualTo("5");
        assertThat(
                        store.getTaskMetricStore(JOB_ID.toString(), "taskid")
                                .getMetric("8.opname.abc.metric7", "-1"))
                .isEqualTo("6");
        assertThat(
                        store.getTaskMetricStore(JOB_ID.toString(), "taskid")
                                .getMetric("1.opname.abc.metric7", "-1"))
                .isEqualTo("6");
        assertThat(
                        store.getSubtaskMetricStore(JOB_ID.toString(), "taskid", 1)
                                .getMetric("opname.abc.metric7", "-1"))
                .isEqualTo("6");
        assertThat(store.getSubtaskAttemptMetricStore(JOB_ID.toString(), "taskid", 1, 2)).isNull();
        assertThat(
                        store.getSubtaskAttemptMetricStore(JOB_ID.toString(), "taskid", 1, 3)
                                .getMetric("opname.abc.metric7", "-1"))
                .isEqualTo("6");
        assertThat(
                        store.getSubtaskAttemptMetricStore(JOB_ID.toString(), "taskid", 8, 2)
                                .getMetric("opname.abc.metric7", "-1"))
                .isEqualTo("6");
        assertThat(
                        store.getSubtaskAttemptMetricStore(JOB_ID.toString(), "taskid", 8, 4)
                                .getMetric("opname.abc.metric7", "-1"))
                .isEqualTo("16");

        assertThat(
                        store.getTaskMetricStore(JOB_ID.toString(), "taskid")
                                .getJobManagerOperatorMetricStore()
                                .getMetric("opname.abc.metric8", "-1"))
                .isEqualTo("19");

        assertThat(
                        store.getTaskMetricStore(JOB_ID.toString(), "taskid")
                                .getJobManagerOperatorMetricStore()
                                .getMetric("opname.abc.metric9", "-1"))
                .isEqualTo("20");
    }