### Question:
TaskMetricGroup extends ComponentMetricGroup<TaskManagerJobMetricGroup> { @Override public void close() { super.close(); parent.removeTaskMetricGroup(executionId); } TaskMetricGroup(
			MetricRegistry registry,
			TaskManagerJobMetricGroup parent,
			@Nullable JobVertexID vertexId,
			AbstractID executionId,
			@Nullable String taskName,
			int subtaskIndex,
			int attemptNumber); final TaskManagerJobMetricGroup parent(); AbstractID executionId(); @Nullable AbstractID vertexId(); @Nullable String taskName(); int subtaskIndex(); int attemptNumber(); TaskIOMetricGroup getIOMetricGroup(); OperatorMetricGroup addOperator(String name); OperatorMetricGroup addOperator(OperatorID operatorID, String name); @Override void close();  }### Answer:
@Test public void testTaskMetricGroupCleanup() throws Exception { CountingMetricRegistry registry = new CountingMetricRegistry(new Configuration()); TaskManagerMetricGroup taskManagerMetricGroup = new TaskManagerMetricGroup(registry, "localhost", "0"); TaskManagerJobMetricGroup taskManagerJobMetricGroup = new TaskManagerJobMetricGroup(registry, taskManagerMetricGroup, new JobID(), "job"); TaskMetricGroup taskMetricGroup = new TaskMetricGroup(registry, taskManagerJobMetricGroup, new JobVertexID(), new AbstractID(), "task", 0, 0); assertTrue(registry.getNumberRegisteredMetrics() > 0); taskMetricGroup.close(); assertEquals(0, registry.getNumberRegisteredMetrics()); registry.shutdown().get(); }