This is an automated email from the ASF dual-hosted git repository. pnowojski pushed a commit to branch release-2.3 in repository https://gitbox.apache.org/repos/asf/flink.git
commit 8f6c6547b196cd1934aa1bcba9201ea70cf3d51c Author: Efrat Levitan <[email protected]> AuthorDate: Thu Sep 3 19:26:10 2026 +0300 [FLINK-40549][metrics] Remove metricGroup parent reference upon close() Today there's no API to deregister child metric groups added via MetricGroup#addGroup. This is not a problem as long as child metric groups are permanent through parent lifecycle. For split level metric group this is a problem because they might not live as long as the source (aka splitFinished). As a result we are left with a closed child metric group reference on the parent, preventing the object GC. --- .../metrics/groups/AbstractMetricGroup.java | 11 +++++ .../runtime/metrics/groups/MetricGroupTest.java | 49 ++++++++++++++++++++++ 2 files changed, 60 insertions(+) diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/metrics/groups/AbstractMetricGroup.java b/flink-runtime/src/main/java/org/apache/flink/runtime/metrics/groups/AbstractMetricGroup.java index 398f681c505..ec5509707f9 100644 --- a/flink-runtime/src/main/java/org/apache/flink/runtime/metrics/groups/AbstractMetricGroup.java +++ b/flink-runtime/src/main/java/org/apache/flink/runtime/metrics/groups/AbstractMetricGroup.java @@ -337,6 +337,17 @@ public abstract class AbstractMetricGroup<A extends AbstractMetricGroup<?>> impl metrics.clear(); } } + if (parent != null) { + parent.removeChildGroup(this); + } + } + + void removeChildGroup(AbstractMetricGroup<?> childGroup) { + synchronized (this) { + if (!closed) { + groups.values().remove(childGroup); + } + } } public final boolean isClosed() { diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/metrics/groups/MetricGroupTest.java b/flink-runtime/src/test/java/org/apache/flink/runtime/metrics/groups/MetricGroupTest.java index 28f9c95440d..b67706d01b6 100644 --- a/flink-runtime/src/test/java/org/apache/flink/runtime/metrics/groups/MetricGroupTest.java +++ b/flink-runtime/src/test/java/org/apache/flink/runtime/metrics/groups/MetricGroupTest.java @@ -374,6 +374,55 @@ public class MetricGroupTest { .isTrue(); } + /** + * Verifies that a group is no longer referenced by its parent once it is closed. Otherwise + * groups with unique names that are closed before their parent, e.g. per-split groups of a + * source, would leak until the parent is closed. + */ + @Test + void closedGroupIsRemovedFromParent() { + GenericMetricGroup parent = + new GenericMetricGroup( + exceptionOnRegister, + new DummyAbstractMetricGroup(exceptionOnRegister), + "parent") { + @Override + protected GenericMetricGroup createChildGroup( + String name, ChildType childType) { + return new CloseCountingMetricGroup(exceptionOnRegister, this, name); + } + }; + CloseCountingMetricGroup closedBeforeParent = + (CloseCountingMetricGroup) parent.addGroup("closedBeforeParent"); + CloseCountingMetricGroup closedWithParent = + (CloseCountingMetricGroup) parent.addGroup("closedWithParent"); + + closedBeforeParent.close(); + assertThat(closedBeforeParent.closeCalls).isOne(); + + // the parent closes every group it still references + parent.close(); + assertThat(closedWithParent.closeCalls).isOne(); + assertThat(closedBeforeParent.closeCalls) + .withFailMessage("The parent must not reference a group that was closed before") + .isOne(); + } + + private static class CloseCountingMetricGroup extends GenericMetricGroup { + int closeCalls = 0; + + CloseCountingMetricGroup( + MetricRegistry registry, AbstractMetricGroup<?> parent, String name) { + super(registry, parent, name); + } + + @Override + public void close() { + closeCalls++; + super.close(); + } + } + @Test void tolerateMetricNameCollisions() { final String name = "abctestname";
