This is an automated email from the ASF dual-hosted git repository.
HuangZhenQiu pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/flink.git
The following commit(s) were added to refs/heads/master by this push:
new 990bc73dd25 [FLINK-39956][metrics] Add heap memory pool metrics
(Used/Max/CollectionUsed/CollectionMax) (#28582)
990bc73dd25 is described below
commit 990bc73dd250f1a88edc104b16fb55e36b431896
Author: Yiming Jin <[email protected]>
AuthorDate: Wed Jul 1 09:53:50 2026 -0700
[FLINK-39956][metrics] Add heap memory pool metrics
(Used/Max/CollectionUsed/CollectionMax) (#28582)
Expose per-pool heap memory metrics via instantiateMemoryPoolMetrics,
grouped
under a new "Pool" metric group, to surface finer-grained GC and memory pool
information for heap pools reported by MemoryPoolMXBean.
Co-authored-by: Yiming Jin <[email protected]>
---
.../flink/runtime/metrics/util/MetricUtils.java | 28 +++++++++++++++++++
.../runtime/metrics/util/MetricUtilsTest.java | 31 ++++++++++++++++++++++
2 files changed, 59 insertions(+)
diff --git
a/flink-runtime/src/main/java/org/apache/flink/runtime/metrics/util/MetricUtils.java
b/flink-runtime/src/main/java/org/apache/flink/runtime/metrics/util/MetricUtils.java
index e208c465400..162738ae40c 100644
---
a/flink-runtime/src/main/java/org/apache/flink/runtime/metrics/util/MetricUtils.java
+++
b/flink-runtime/src/main/java/org/apache/flink/runtime/metrics/util/MetricUtils.java
@@ -57,6 +57,7 @@ import java.lang.management.ClassLoadingMXBean;
import java.lang.management.GarbageCollectorMXBean;
import java.lang.management.ManagementFactory;
import java.lang.management.MemoryPoolMXBean;
+import java.lang.management.MemoryType;
import java.lang.management.MemoryUsage;
import java.lang.management.ThreadMXBean;
import java.time.Duration;
@@ -85,6 +86,8 @@ public class MetricUtils {
@VisibleForTesting static final String METRIC_GROUP_MEMORY = "Memory";
+ @VisibleForTesting static final String METRIC_GROUP_POOL_NAME = "Pool";
+
@VisibleForTesting static final String METRIC_GROUP_MANAGED_MEMORY =
"Managed";
private static final String WRITER_SUFFIX = ": " +
ConfigConstants.WRITER_NAME;
private static final String COMMITTER_SUFFIX = ": " +
ConfigConstants.COMMITTER_NAME;
@@ -263,6 +266,7 @@ public class MetricUtils {
instantiateHeapMemoryMetrics(metrics.addGroup(METRIC_GROUP_HEAP_NAME));
instantiateNonHeapMemoryMetrics(metrics.addGroup(METRIC_GROUP_NONHEAP_NAME));
instantiateMetaspaceMemoryMetrics(metrics);
+ instantiateMemoryPoolMetrics(metrics.addGroup(METRIC_GROUP_POOL_NAME));
final MBeanServer con = ManagementFactory.getPlatformMBeanServer();
@@ -342,6 +346,30 @@ public class MetricUtils {
}
}
+ @VisibleForTesting
+ static void instantiateMemoryPoolMetrics(MetricGroup metrics) {
+ for (final MemoryPoolMXBean pool :
ManagementFactory.getMemoryPoolMXBeans()) {
+ if (pool.getType() != MemoryType.HEAP) {
+ continue;
+ }
+ final MetricGroup poolGroup = metrics.addGroup(pool.getName());
+ poolGroup.<Long, Gauge<Long>>gauge("Used", () ->
usageUsed(pool.getUsage()));
+ poolGroup.<Long, Gauge<Long>>gauge("Max", () ->
usageMax(pool.getUsage()));
+ poolGroup.<Long, Gauge<Long>>gauge(
+ "CollectionUsed", () ->
usageUsed(pool.getCollectionUsage()));
+ poolGroup.<Long, Gauge<Long>>gauge(
+ "CollectionMax", () ->
usageMax(pool.getCollectionUsage()));
+ }
+ }
+
+ private static long usageUsed(MemoryUsage usage) {
+ return usage == null ? -1L : usage.getUsed();
+ }
+
+ private static long usageMax(MemoryUsage usage) {
+ return usage == null ? -1L : usage.getMax();
+ }
+
static void instantiateFileDescriptorMetrics(MetricGroup metrics) {
if (OperatingSystem.isWindows()) {
if (!fileDescriptorWarningShown) {
diff --git
a/flink-runtime/src/test/java/org/apache/flink/runtime/metrics/util/MetricUtilsTest.java
b/flink-runtime/src/test/java/org/apache/flink/runtime/metrics/util/MetricUtilsTest.java
index 8628d86fa68..b71df9a22a4 100644
---
a/flink-runtime/src/test/java/org/apache/flink/runtime/metrics/util/MetricUtilsTest.java
+++
b/flink-runtime/src/test/java/org/apache/flink/runtime/metrics/util/MetricUtilsTest.java
@@ -168,6 +168,37 @@ class MetricUtilsTest {
assertThat(perSecond.getRate()).isGreaterThan(0.);
}
+ @Test
+ void testMemoryPoolMetricsCompleteness() {
+ final Map<String, InterceptingOperatorMetricGroup> addedGroups = new
HashMap<>();
+ final InterceptingOperatorMetricGroup poolMetrics =
+ new InterceptingOperatorMetricGroup() {
+ @Override
+ public MetricGroup addGroup(String name) {
+ return addedGroups.computeIfAbsent(
+ name, k -> new
InterceptingOperatorMetricGroup());
+ }
+ };
+
+ MetricUtils.instantiateMemoryPoolMetrics(poolMetrics);
+
+ assertThat(addedGroups)
+ .withFailMessage("Expected at least one heap memory pool group
to be registered")
+ .isNotEmpty();
+ assertThat(addedGroups).doesNotContainKey("Metaspace");
+
+ for (Map.Entry<String, InterceptingOperatorMetricGroup> e :
addedGroups.entrySet()) {
+ final InterceptingOperatorMetricGroup g = e.getValue();
+ assertThat(g.get("Used")).isNotNull();
+ assertThat(g.get("Max")).isNotNull();
+ assertThat(g.get("CollectionUsed")).isNotNull();
+ assertThat(g.get("CollectionMax")).isNotNull();
+ @SuppressWarnings("unchecked")
+ final Gauge<Long> used = (Gauge<Long>) g.get("Used");
+ assertThat(used.getValue()).isGreaterThanOrEqualTo(0L);
+ }
+ }
+
@Test
void testHeapMetricsCompleteness() {
final InterceptingOperatorMetricGroup heapMetrics = new
InterceptingOperatorMetricGroup();
