[FLINK-30409] Return a new metric group when creating closed metric g…

…roup

This closes apache#21539

flink-runtime/src/main/java/org/apache/flink/runtime/metrics/groups/AbstractMetricGroup.java

@@ -461,8 +461,8 @@ private AbstractMetricGroup<?> addGroup(String name, ChildType childType) {
 
  AbstractMetricGroup newGroup = createChildGroup(name, childType);
  AbstractMetricGroup prior = groups.put(name, newGroup);
- if (prior == null) {
- // no prior group with that name
+ if (prior == null || prior.isClosed()) {
+ // no prior group or closed group with that name
  return newGroup;
  } else {
  // had a prior group with that name, add the prior group back

flink-runtime/src/test/java/org/apache/flink/runtime/metrics/groups/MetricGroupTest.java

@@ -310,6 +310,29 @@ public void closedGroupCreatesClosedGroups() {
  assertTrue(subgroup.isClosed());
  }
 
+ @Test
+ public void addClosedGroupReturnsNewGroupInstance() {
+ GenericMetricGroup mainGroup =
+ new GenericMetricGroup(
+ exceptionOnRegister,
+ new DummyAbstractMetricGroup(exceptionOnRegister),
+ "mainGroup");
+
+ AbstractMetricGroup<?> subGroup = (AbstractMetricGroup<?>) mainGroup.addGroup("subGroup");
+
+ assertFalse(subGroup.isClosed());
+
+ subGroup.close();
+ assertTrue(subGroup.isClosed());
+
+ AbstractMetricGroup<?> newSubGroupWithSameNameAsClosedGroup =
+ (AbstractMetricGroup<?>) mainGroup.addGroup("subGroup");
+ assertFalse(
+ "The new subgroup should not be closed",
+ newSubGroupWithSameNameAsClosedGroup.isClosed());
+ assertTrue("The old sub group is not modified", subGroup.isClosed());
+ }
+
  @Test
  public void tolerateMetricNameCollisions() {
  final String name = "abctestname";