@Override
    public void close() {
        super.close();

        parent.removeTaskMetricGroup(executionId);
    }
@Test
    void testTaskMetricGroupCleanup() throws Exception {
        CountingMetricRegistry registry = new CountingMetricRegistry(new Configuration());
        TaskManagerMetricGroup taskManagerMetricGroup =
                TaskManagerMetricGroup.createTaskManagerMetricGroup(
                        registry, "localhost", new ResourceID("0"));

        int initialMetricsCount = registry.getNumberRegisteredMetrics();
        TaskMetricGroup taskMetricGroup =
                taskManagerMetricGroup
                        .addJob(new JobID(), "job")
                        .addTask(createExecutionAttemptId(), "task");

        // the io metric should have registered predefined metrics
        assertThat(registry.getNumberRegisteredMetrics()).isGreaterThan(initialMetricsCount);

        taskMetricGroup.close();

        // now all registered metrics should have been unregistered
        assertThat(registry.getNumberRegisteredMetrics()).isEqualTo(initialMetricsCount);

        registry.closeAsync().get();
    }