[FLINK-37668][core] TimeUtils supports to parse ISO-8601 duration format

Author: masteryhx

flink-core/src/main/java/org/apache/flink/util/TimeUtils.java

@@ -20,6 +20,7 @@
 
 import java.math.BigInteger;
 import java.time.Duration;
+import java.time.format.DateTimeParseException;
 import java.time.temporal.ChronoUnit;
 import java.util.Arrays;
 import java.util.Collections;
@@ -43,7 +44,8 @@ public class TimeUtils {
     /**
      * Parse the given string to a java {@link Duration}. The string is in format "{length
      * value}{time unit label}", e.g. "123ms", "321 s". If no time unit label is specified, it will
-     * be considered as milliseconds.
+     * be considered as milliseconds. If above rules are not matched, it will fall back to parse
+     * ISO-8601 duration format.
      *
      * <p>Supported time unit labels are:
      *
@@ -77,7 +79,19 @@ public static Duration parseDuration(String text) {
         final String unitLabel = trimmed.substring(pos).trim().toLowerCase(Locale.US);
 
         if (number.isEmpty()) {
-            throw new NumberFormatException("text does not start with a number");
+            try {
+                // Fall back to parse ISO-8601 duration format
+                Duration parsedDuration = Duration.parse(trimmed);
+                if (parsedDuration.isNegative()) {
+                    // Don't support negative duration which is consistent with before format
+                    throw new NumberFormatException("negative duration is not supported");
+                }
+                return parsedDuration;
+            } catch (DateTimeParseException e) {
+                throw new NumberFormatException(
+                        "text does not start with a number, and is not a valid ISO-8601 duration format: "
+                                + trimmed);
+            }
         }
 
         final BigInteger value;

flink-core/src/test/java/org/apache/flink/util/TimeUtilsTest.java

@@ -116,6 +116,14 @@ void testParseDurationTrim() {
         assertThat(TimeUtils.parseDuration("      155      ms   ").toMillis()).isEqualTo(155L);
     }
 
+    @Test
+    public void testParseDurationISO8601() {
+        assertThat(TimeUtils.parseDuration("PT20.345S").toMillis()).isEqualTo(20345);
+        assertThat(TimeUtils.parseDuration("PT15M").toMinutes()).isEqualTo(15);
+        assertThat(TimeUtils.parseDuration("PT10H").toHours()).isEqualTo(10);
+        assertThat(TimeUtils.parseDuration("P2DT3H4M").toMinutes()).isEqualTo(3064);
+    }
+
     @Test
     void testParseDurationInvalid() {
         // null
@@ -145,6 +153,10 @@ void testParseDurationInvalid() {
         // negative number
         assertThatThrownBy(() -> TimeUtils.parseDuration("-100 ms"))
                 .isInstanceOf(IllegalArgumentException.class);
+
+        // negative ISO-8601
+        assertThatThrownBy(() -> TimeUtils.parseDuration("-PT6H3M"))
+                .isInstanceOf(NumberFormatException.class);
     }
 
     @Test