[FLINK-39130][metrics] Allow native types in MetricConfig

flink-metrics/flink-metrics-core/src/main/java/org/apache/flink/metrics/MetricConfig.java

@@ -22,7 +22,14 @@
 
 import java.util.Properties;
 
-/** A properties class with added utility method to extract primitives. */
+/**
+ * A properties class with added utility methods to extract primitives.
+ *
+ * <p>Values may be stored as strings via {@link #setProperty(String, String)} or as native Java
+ * types via {@link #put(Object, Object)} (e.g., when Flink's YAML configuration parser stores
+ * Integer, Long, or Boolean values directly). The getter methods handle both representations
+ * transparently.
+ */
 @Public
 public class MetricConfig extends Properties {
 
@@ -31,72 +38,108 @@ public String getString(String key, String defaultValue) {
     }
 
     /**
-     * Searches for the property with the specified key in this property list. If the key is not
-     * found in this property list, the default property list, and its defaults, recursively, are
-     * then checked. The method returns the default value argument if the property is not found.
+     * Returns the value associated with the given key as an {@code int}.
+     *
+     * <p>If the value is a {@link Number}, its {@code intValue()} is returned directly. Otherwise,
+     * the value's string representation is parsed via {@link Integer#parseInt(String)}.
      *
      * @param key the hashtable key.
      * @param defaultValue a default value.
-     * @return the value in this property list with the specified key value parsed as an int.
+     * @return the value in this property list with the specified key value as an int.
      */
     public int getInteger(String key, int defaultValue) {
-        String argument = getProperty(key, null);
-        return argument == null ? defaultValue : Integer.parseInt(argument);
+        final Object value = get(key);
+        if (value == null) {
+            return defaultValue;
+        }
+        if (value instanceof Number) {
+            return ((Number) value).intValue();
+        }
+        return Integer.parseInt(value.toString());
     }
 
     /**
-     * Searches for the property with the specified key in this property list. If the key is not
-     * found in this property list, the default property list, and its defaults, recursively, are
-     * then checked. The method returns the default value argument if the property is not found.
+     * Returns the value associated with the given key as a {@code long}.
+     *
+     * <p>If the value is a {@link Number}, its {@code longValue()} is returned directly. Otherwise,
+     * the value's string representation is parsed via {@link Long#parseLong(String)}.
      *
      * @param key the hashtable key.
      * @param defaultValue a default value.
-     * @return the value in this property list with the specified key value parsed as a long.
+     * @return the value in this property list with the specified key value as a long.
      */
     public long getLong(String key, long defaultValue) {
-        String argument = getProperty(key, null);
-        return argument == null ? defaultValue : Long.parseLong(argument);
+        final Object value = get(key);
+        if (value == null) {
+            return defaultValue;
+        }
+        if (value instanceof Number) {
+            return ((Number) value).longValue();
+        }
+        return Long.parseLong(value.toString());
     }
 
     /**
-     * Searches for the property with the specified key in this property list. If the key is not
-     * found in this property list, the default property list, and its defaults, recursively, are
-     * then checked. The method returns the default value argument if the property is not found.
+     * Returns the value associated with the given key as a {@code float}.
+     *
+     * <p>If the value is a {@link Number}, its {@code floatValue()} is returned directly.
+     * Otherwise, the value's string representation is parsed via {@link Float#parseFloat(String)}.
      *
      * @param key the hashtable key.
      * @param defaultValue a default value.
-     * @return the value in this property list with the specified key value parsed as a float.
+     * @return the value in this property list with the specified key value as a float.
      */
     public float getFloat(String key, float defaultValue) {
-        String argument = getProperty(key, null);
-        return argument == null ? defaultValue : Float.parseFloat(argument);
+        final Object value = get(key);
+        if (value == null) {
+            return defaultValue;
+        }
+        if (value instanceof Number) {
+            return ((Number) value).floatValue();
+        }
+        return Float.parseFloat(value.toString());
     }
 
     /**
-     * Searches for the property with the specified key in this property list. If the key is not
-     * found in this property list, the default property list, and its defaults, recursively, are
-     * then checked. The method returns the default value argument if the property is not found.
+     * Returns the value associated with the given key as a {@code double}.
+     *
+     * <p>If the value is a {@link Number}, its {@code doubleValue()} is returned directly.
+     * Otherwise, the value's string representation is parsed via {@link
+     * Double#parseDouble(String)}.
      *
      * @param key the hashtable key.
      * @param defaultValue a default value.
-     * @return the value in this property list with the specified key value parsed as a double.
+     * @return the value in this property list with the specified key value as a double.
      */
     public double getDouble(String key, double defaultValue) {
-        String argument = getProperty(key, null);
-        return argument == null ? defaultValue : Double.parseDouble(argument);
+        final Object value = get(key);
+        if (value == null) {
+            return defaultValue;
+        }
+        if (value instanceof Number) {
+            return ((Number) value).doubleValue();
+        }
+        return Double.parseDouble(value.toString());
     }
 
     /**
-     * Searches for the property with the specified key in this property list. If the key is not
-     * found in this property list, the default property list, and its defaults, recursively, are
-     * then checked. The method returns the default value argument if the property is not found.
+     * Returns the value associated with the given key as a {@code boolean}.
+     *
+     * <p>If the value is a {@link Boolean}, it is returned directly. Otherwise, the value's string
+     * representation is parsed via {@link Boolean#parseBoolean(String)}.
      *
      * @param key the hashtable key.
      * @param defaultValue a default value.
-     * @return the value in this property list with the specified key value parsed as a boolean.
+     * @return the value in this property list with the specified key value as a boolean.
      */
     public boolean getBoolean(String key, boolean defaultValue) {
-        String argument = getProperty(key, null);
-        return argument == null ? defaultValue : Boolean.parseBoolean(argument);
+        final Object value = get(key);
+        if (value == null) {
+            return defaultValue;
+        }
+        if (value instanceof Boolean) {
+            return (Boolean) value;
+        }
+        return Boolean.parseBoolean(value.toString());
     }
 }

flink-metrics/flink-metrics-core/src/test/java/org/apache/flink/metrics/MetricConfigTest.java

@@ -0,0 +1,114 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.metrics;
+
+import org.junit.jupiter.params.ParameterizedTest;
+import org.junit.jupiter.params.provider.Arguments;
+import org.junit.jupiter.params.provider.MethodSource;
+
+import java.util.stream.Stream;
+
+import static org.assertj.core.api.Assertions.assertThat;
+
+/** Tests for {@link MetricConfig}. */
+class MetricConfigTest {
+
+    @ParameterizedTest
+    @MethodSource("fromStringCases")
+    void testGetFromString(
+            final String storedValue, final TypedGetter getter, final Object expected) {
+        final MetricConfig config = new MetricConfig();
+        config.setProperty("key", storedValue);
+        assertThat(getter.get(config, "key")).isEqualTo(expected);
+    }
+
+    @ParameterizedTest
+    @MethodSource("nativeTypeCases")
+    void testGetFromNativeType(
+            final Object storedValue, final TypedGetter getter, final Object expected) {
+        final MetricConfig config = new MetricConfig();
+        config.put("key", storedValue);
+        assertThat(getter.get(config, "key")).isEqualTo(expected);
+    }
+
+    @ParameterizedTest
+    @MethodSource("crossTypeCases")
+    void testGetCrossType(
+            final Object storedValue, final TypedGetter getter, final Object expected) {
+        final MetricConfig config = new MetricConfig();
+        config.put("key", storedValue);
+        assertThat(getter.get(config, "key")).isEqualTo(expected);
+    }
+
+    @ParameterizedTest
+    @MethodSource("defaultValueCases")
+    void testGetDefault(final TypedGetter getter, final Object expected) {
+        final MetricConfig config = new MetricConfig();
+        assertThat(getter.get(config, "missing")).isEqualTo(expected);
+    }
+
+    @FunctionalInterface
+    private interface TypedGetter {
+        Object get(MetricConfig config, String key);
+    }
+
+    private static Stream<Arguments> fromStringCases() {
+        return Stream.of(
+                Arguments.of("42", (TypedGetter) (c, k) -> c.getInteger(k, 0), 42),
+                Arguments.of(
+                        "123456789012345",
+                        (TypedGetter) (c, k) -> c.getLong(k, 0L),
+                        123456789012345L),
+                Arguments.of("3.14", (TypedGetter) (c, k) -> c.getFloat(k, 0.0f), 3.14f),
+                Arguments.of(
+                        "2.718281828", (TypedGetter) (c, k) -> c.getDouble(k, 0.0), 2.718281828),
+                Arguments.of("true", (TypedGetter) (c, k) -> c.getBoolean(k, false), true));
+    }
+
+    private static Stream<Arguments> nativeTypeCases() {
+        return Stream.of(
+                Arguments.of(42, (TypedGetter) (c, k) -> c.getInteger(k, 0), 42),
+                Arguments.of(
+                        123456789012345L,
+                        (TypedGetter) (c, k) -> c.getLong(k, 0L),
+                        123456789012345L),
+                Arguments.of(3.14f, (TypedGetter) (c, k) -> c.getFloat(k, 0.0f), 3.14f),
+                Arguments.of(2.718281828, (TypedGetter) (c, k) -> c.getDouble(k, 0.0), 2.718281828),
+                Arguments.of(true, (TypedGetter) (c, k) -> c.getBoolean(k, false), true));
+    }
+
+    private static Stream<Arguments> crossTypeCases() {
+        return Stream.of(
+                Arguments.of(42, (TypedGetter) (c, k) -> c.getLong(k, 0L), 42L),
+                Arguments.of(100L, (TypedGetter) (c, k) -> c.getInteger(k, 0), 100),
+                Arguments.of(42, (TypedGetter) (c, k) -> c.getDouble(k, 0.0), 42.0),
+                Arguments.of(42, (TypedGetter) (c, k) -> c.getFloat(k, 0.0f), 42.0f));
+    }
+
+    private static Stream<Arguments> defaultValueCases() {
+        return Stream.of(
+                Arguments.of((TypedGetter) (c, k) -> c.getInteger(k, 99), 99),
+                Arguments.of((TypedGetter) (c, k) -> c.getLong(k, 99L), 99L),
+                Arguments.of((TypedGetter) (c, k) -> c.getFloat(k, 1.5f), 1.5f),
+                Arguments.of((TypedGetter) (c, k) -> c.getDouble(k, 1.5), 1.5),
+                Arguments.of((TypedGetter) (c, k) -> c.getBoolean(k, true), true),
+                Arguments.of((TypedGetter) (c, k) -> c.getBoolean(k, false), false),
+                Arguments.of((TypedGetter) (c, k) -> c.getString(k, "default"), "default"));
+    }
+}