[flink] 02/02: [FLINK-13198][table-api] Add a utility to validate and parse duration in DescriptorProperties

[ja...@apache.org]

This is an automated email from the ASF dual-hosted git repository.

jark pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/flink.git

commit 1426b218678224344844e1f1bdd7f1b1c66e0a32
Author: TsReaper <ts...@gmail.com>
AuthorDate: Thu Jul 11 18:57:35 2019 +0800

    [FLINK-13198][table-api] Add a utility to validate and parse duration in DescriptorProperties
---
 .../table/descriptors/DescriptorProperties.java    | 65 ++++++++++++++++++++++
 1 file changed, 65 insertions(+)

diff --git a/flink-table/flink-table-common/src/main/java/org/apache/flink/table/descriptors/DescriptorProperties.java b/flink-table/flink-table-common/src/main/java/org/apache/flink/table/descriptors/DescriptorProperties.java
index 446fb73..3606504 100644
--- a/flink-table/flink-table-common/src/main/java/org/apache/flink/table/descriptors/DescriptorProperties.java
+++ b/flink-table/flink-table-common/src/main/java/org/apache/flink/table/descriptors/DescriptorProperties.java
@@ -29,8 +29,10 @@ import org.apache.flink.table.utils.EncodingUtils;
 import org.apache.flink.table.utils.TypeStringUtils;
 import org.apache.flink.util.InstantiationUtil;
 import org.apache.flink.util.Preconditions;
+import org.apache.flink.util.TimeUtils;
 
 import java.math.BigDecimal;
+import java.time.Duration;
 import java.util.ArrayList;
 import java.util.Arrays;
 import java.util.Collections;
@@ -548,6 +550,26 @@ public class DescriptorProperties {
 	}
 
 	/**
+	 * Returns a Java {@link Duration} under the given key if it exists.
+	 */
+	public Optional<Duration> getOptionalDuration(String key) {
+		return optionalGet(key).map((value) -> {
+			try {
+				return TimeUtils.parseDuration(value);
+			} catch (Exception e) {
+				throw new ValidationException("Invalid duration value for key '" + key + "'.", e);
+			}
+		});
+	}
+
+	/**
+	 * Returns a java {@link Duration} under the given existing key.
+	 */
+	public Duration getDuration(String key) {
+		return getOptionalDuration(key).orElseThrow(exceptionSupplier(key));
+	}
+
+	/**
 	 * Returns the property keys of fixed indexed properties.
 	 *
 	 * <p>For example:
@@ -1043,6 +1065,49 @@ public class DescriptorProperties {
 	}
 
 	/**
+	 * Validates a Java {@link Duration}.
+	 *
+	 * <p>The precision defines the allowed minimum unit in milliseconds (e.g. 1000 would only allow seconds).
+	 */
+	public void validateDuration(String key, boolean isOptional, int precision) {
+		validateDuration(key, isOptional, precision, 0L, Long.MAX_VALUE);
+	}
+
+	/**
+	 * Validates a Java {@link Duration}. The boundaries are inclusive and in milliseconds.
+	 *
+	 * <p>The precision defines the allowed minimum unit in milliseconds (e.g. 1000 would only allow seconds).
+	 */
+	public void validateDuration(String key, boolean isOptional, int precision, long min) {
+		validateDuration(key, isOptional, precision, min, Long.MAX_VALUE);
+	}
+
+	/**
+	 * Validates a Java {@link Duration}. The boundaries are inclusive and in milliseconds.
+	 *
+	 * <p>The precision defines the allowed minimum unit in milliseconds (e.g. 1000 would only allow seconds).
+	 */
+	public void validateDuration(String key, boolean isOptional, int precision, long min, long max) {
+		Preconditions.checkArgument(precision > 0);
+
+		validateComparable(
+			key,
+			isOptional,
+			min,
+			max,
+			"time interval (in milliseconds)",
+			(value) -> {
+				final long ms = TimeUtils.parseDuration(value).toMillis();
+				if (ms % precision != 0) {
+					throw new ValidationException(
+						"Duration for key '" + key + "' must be a multiple of " + precision + " milliseconds but was: " + value);
+				}
+				return ms;
+			}
+		);
+	}
+
+	/**
 	 * Validates an enum property with a set of validation logic for each enum value.
 	 */
 	public void validateEnum(String key, boolean isOptional, Map<String, Consumer<String>> enumValidation) {