From 6661f49d5f2231ac702b38ac94685a371f0a54d8 Mon Sep 17 00:00:00 2001
From: Navin Viswanath <navinviswanath@Navins-MacBook-Air.local>
Date: Sat, 27 Sep 2025 12:53:56 -0700
Subject: [PATCH 1/2] Makes error message for fractional values for MemorySize
 explicit

Currently for a fractional value for memory size, like 1.5g, the exception message is generated from failing
to parse the string ".5g". Improving the exception handling for this case.
---
 .../main/java/org/apache/flink/configuration/MemorySize.java  | 4 ++++
 .../java/org/apache/flink/configuration/MemorySizeTest.java   | 4 ++++
 2 files changed, 8 insertions(+)

diff --git a/flink-core-api/src/main/java/org/apache/flink/configuration/MemorySize.java b/flink-core-api/src/main/java/org/apache/flink/configuration/MemorySize.java
index d860c6e30fa43..e170c23a4a86a 100644
--- a/flink-core-api/src/main/java/org/apache/flink/configuration/MemorySize.java
+++ b/flink-core-api/src/main/java/org/apache/flink/configuration/MemorySize.java
@@ -287,6 +287,10 @@ public static long parseBytes(String text) throws IllegalArgumentException {
             throw new NumberFormatException("text does not start with a number");
         }
 
+        if(unit.startsWith(".")) {
+            throw new IllegalArgumentException("Memory size must be an integer value. Fractional values are not supported. Found: " + text);
+        }
+
         final long value;
         try {
             value = Long.parseLong(number); // this throws a NumberFormatException on overflow
diff --git a/flink-core-api/src/test/java/org/apache/flink/configuration/MemorySizeTest.java b/flink-core-api/src/test/java/org/apache/flink/configuration/MemorySizeTest.java
index a619f9f00470d..2d2c58e254034 100644
--- a/flink-core-api/src/test/java/org/apache/flink/configuration/MemorySizeTest.java
+++ b/flink-core-api/src/test/java/org/apache/flink/configuration/MemorySizeTest.java
@@ -179,6 +179,10 @@ void testParseInvalid() {
         // negative number
         assertThatThrownBy(() -> MemorySize.parseBytes("-100 bytes"))
                 .isInstanceOf(IllegalArgumentException.class);
+
+        // fractional number
+        assertThatThrownBy(() -> MemorySize.parseBytes("1.5g"))
+                .isInstanceOf(IllegalArgumentException.class);
     }
 
     @Test

From f5d85818abce1ad36e1e01a7f196e7022ad428f2 Mon Sep 17 00:00:00 2001
From: Navin Viswanath <navinviswanath@mac.lan>
Date: Mon, 29 Sep 2025 15:50:30 -0700
Subject: [PATCH 2/2] Use regex

---
 .../flink/configuration/MemorySize.java       | 27 +++++++++++++++++--
 1 file changed, 25 insertions(+), 2 deletions(-)

diff --git a/flink-core-api/src/main/java/org/apache/flink/configuration/MemorySize.java b/flink-core-api/src/main/java/org/apache/flink/configuration/MemorySize.java
index e170c23a4a86a..2784291f508c5 100644
--- a/flink-core-api/src/main/java/org/apache/flink/configuration/MemorySize.java
+++ b/flink-core-api/src/main/java/org/apache/flink/configuration/MemorySize.java
@@ -27,6 +27,8 @@
 import java.util.Locale;
 import java.util.Objects;
 import java.util.Optional;
+import java.util.regex.Matcher;
+import java.util.regex.Pattern;
 import java.util.stream.IntStream;
 
 import static org.apache.flink.configuration.MemorySize.MemoryUnit.BYTES;
@@ -287,8 +289,14 @@ public static long parseBytes(String text) throws IllegalArgumentException {
             throw new NumberFormatException("text does not start with a number");
         }
 
-        if(unit.startsWith(".")) {
-            throw new IllegalArgumentException("Memory size must be an integer value. Fractional values are not supported. Found: " + text);
+        String unitsRegex = MemoryUnit.getRegularExpressionForAllUnits();
+        Pattern regexPattern = Pattern.compile("^(\\d+)\\s*" + unitsRegex + "?$");
+        Matcher matcher = regexPattern.matcher(trimmed.toLowerCase(Locale.US));
+
+        if (!matcher.matches()) {
+            throw new IllegalArgumentException(
+                    "Memory size must be an integer value optionally followed by a unit. Found: "
+                            + text);
         }
 
         final long value;
@@ -392,6 +400,21 @@ public static String getAllUnits() {
                     TERA_BYTES.getUnits());
         }
 
+        public static String getRegularExpressionForAllUnits() {
+            String delimiter = "|";
+            String units =
+                    String.join(
+                            delimiter,
+                            new String[] {
+                                String.join(delimiter, BYTES.getUnits()),
+                                String.join(delimiter, KILO_BYTES.getUnits()),
+                                String.join(delimiter, MEGA_BYTES.getUnits()),
+                                String.join(delimiter, GIGA_BYTES.getUnits()),
+                                String.join(delimiter, TERA_BYTES.getUnits()),
+                            });
+            return "(" + units + ")";
+        }
+
         public static boolean hasUnit(String text) {
             Objects.requireNonNull(text, "text cannot be null");