You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@flink.apache.org by tw...@apache.org on 2023/08/07 14:40:34 UTC
[flink] branch master updated: [FLINK-32689][table] Improve validation of local time zone
This is an automated email from the ASF dual-hosted git repository.
twalthr 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 8c9b1767e08 [FLINK-32689][table] Improve validation of local time zone
8c9b1767e08 is described below
commit 8c9b1767e08e3fdb3b60a0cc7f09a92647181d64
Author: Timo Walther <tw...@apache.org>
AuthorDate: Wed Jul 26 18:27:01 2023 +0200
[FLINK-32689][table] Improve validation of local time zone
---
.../file/table/FileSystemTableFactory.java | 41 ++++++++++-----
.../org/apache/flink/table/api/TableConfig.java | 37 ++++++-------
.../table/api/internal/TableConfigValidation.java | 55 +++++++++++++++++++
.../apache/flink/table/api/TableConfigTest.java | 61 ++++++++++------------
.../table/planner/utils/TableConfigUtils.java | 26 +++------
.../file/table/FileSystemTableFactoryTest.java | 4 +-
6 files changed, 135 insertions(+), 89 deletions(-)
diff --git a/flink-connectors/flink-connector-files/src/main/java/org/apache/flink/connector/file/table/FileSystemTableFactory.java b/flink-connectors/flink-connector-files/src/main/java/org/apache/flink/connector/file/table/FileSystemTableFactory.java
index 2dd052480ed..3f0f415b863 100644
--- a/flink-connectors/flink-connector-files/src/main/java/org/apache/flink/connector/file/table/FileSystemTableFactory.java
+++ b/flink-connectors/flink-connector-files/src/main/java/org/apache/flink/connector/file/table/FileSystemTableFactory.java
@@ -23,6 +23,7 @@ import org.apache.flink.configuration.ConfigOption;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.connector.file.table.factories.BulkReaderFormatFactory;
import org.apache.flink.connector.file.table.factories.BulkWriterFormatFactory;
+import org.apache.flink.table.api.TableConfig;
import org.apache.flink.table.api.ValidationException;
import org.apache.flink.table.connector.format.DecodingFormat;
import org.apache.flink.table.connector.format.EncodingFormat;
@@ -38,6 +39,7 @@ import org.apache.flink.table.factories.FactoryUtil;
import org.apache.flink.table.factories.SerializationFormatFactory;
import org.apache.flink.table.factories.TableFactory;
+import java.time.ZoneId;
import java.util.HashSet;
import java.util.LinkedList;
import java.util.List;
@@ -46,8 +48,6 @@ import java.util.Set;
import java.util.stream.Collectors;
import java.util.stream.Stream;
-import static java.time.ZoneId.SHORT_IDS;
-
/**
* File system {@link TableFactory}.
*
@@ -155,18 +155,9 @@ public class FileSystemTableFactory implements DynamicTableSourceFactory, Dynami
helper.validateExcept(helper.getOptions().get(FactoryUtil.FORMAT) + ".");
// validate time zone of watermark
- String watermarkTimeZone =
+ validateTimeZone(
helper.getOptions()
- .get(FileSystemConnectorOptions.SINK_PARTITION_COMMIT_WATERMARK_TIME_ZONE);
- if (watermarkTimeZone.startsWith("UTC+")
- || watermarkTimeZone.startsWith("UTC-")
- || SHORT_IDS.containsKey(watermarkTimeZone)) {
- throw new ValidationException(
- String.format(
- "The supported watermark time zone is either a full name such as 'America/Los_Angeles',"
- + " or a custom time zone id such as 'GMT-08:00', but configured time zone is '%s'.",
- watermarkTimeZone));
- }
+ .get(FileSystemConnectorOptions.SINK_PARTITION_COMMIT_WATERMARK_TIME_ZONE));
}
private <I, F extends DecodingFormatFactory<I>> DecodingFormat<I> discoverDecodingFormat(
@@ -220,4 +211,28 @@ public class FileSystemTableFactory implements DynamicTableSourceFactory, Dynami
return !matchingFactories.isEmpty();
}
+
+ /** Similar logic as for {@link TableConfig}. */
+ private void validateTimeZone(String zone) {
+ boolean isValid;
+ try {
+ // We enforce a zone string that is compatible with both java.util.TimeZone and
+ // java.time.ZoneId to avoid bugs.
+ // In general, advertising either TZDB ID, GMT+xx:xx, or UTC is the best we can do.
+ isValid = java.util.TimeZone.getTimeZone(zone).toZoneId().equals(ZoneId.of(zone));
+ } catch (Exception e) {
+ isValid = false;
+ }
+
+ if (!isValid) {
+ throw new ValidationException(
+ String.format(
+ "Invalid time zone for '%s'. The value should be a Time Zone Database (TZDB) ID "
+ + "such as 'America/Los_Angeles' to include daylight saving time. Fixed "
+ + "offsets are supported using 'GMT-03:00' or 'GMT+03:00'. Or use 'UTC' "
+ + "without time zone and daylight saving time.",
+ FileSystemConnectorOptions.SINK_PARTITION_COMMIT_WATERMARK_TIME_ZONE
+ .key()));
+ }
+ }
}
diff --git a/flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/api/TableConfig.java b/flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/api/TableConfig.java
index f9600eb5e31..3f6808daa62 100644
--- a/flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/api/TableConfig.java
+++ b/flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/api/TableConfig.java
@@ -35,11 +35,12 @@ import org.apache.flink.util.Preconditions;
import java.time.Duration;
import java.time.ZoneId;
+import java.time.ZoneOffset;
import java.util.HashMap;
import java.util.Map;
import java.util.Optional;
-import static java.time.ZoneId.SHORT_IDS;
+import static org.apache.flink.table.api.internal.TableConfigValidation.validateTimeZone;
/**
* Configuration for the current {@link TableEnvironment} session to adjust Table & SQL API
@@ -242,11 +243,12 @@ public final class TableConfig implements WritableConfig, ReadableConfig {
* @see org.apache.flink.table.types.logical.LocalZonedTimestampType
*/
public ZoneId getLocalTimeZone() {
- String zone = configuration.getString(TableConfigOptions.LOCAL_TIME_ZONE);
+ final String zone = configuration.getString(TableConfigOptions.LOCAL_TIME_ZONE);
+ if (TableConfigOptions.LOCAL_TIME_ZONE.defaultValue().equals(zone)) {
+ return ZoneId.systemDefault();
+ }
validateTimeZone(zone);
- return TableConfigOptions.LOCAL_TIME_ZONE.defaultValue().equals(zone)
- ? ZoneId.systemDefault()
- : ZoneId.of(zone);
+ return ZoneId.of(zone);
}
/**
@@ -296,22 +298,17 @@ public final class TableConfig implements WritableConfig, ReadableConfig {
* @see org.apache.flink.table.types.logical.LocalZonedTimestampType
*/
public void setLocalTimeZone(ZoneId zoneId) {
- validateTimeZone(zoneId.toString());
- configuration.setString(TableConfigOptions.LOCAL_TIME_ZONE, zoneId.toString());
- }
-
- /** Validates user configured time zone. */
- private void validateTimeZone(String zone) {
- final String zoneId = zone.toUpperCase();
- if (zoneId.startsWith("UTC+")
- || zoneId.startsWith("UTC-")
- || SHORT_IDS.containsKey(zoneId)) {
- throw new IllegalArgumentException(
- String.format(
- "The supported Zone ID is either a full name such as 'America/Los_Angeles',"
- + " or a custom timezone id such as 'GMT-08:00', but configured Zone ID is '%s'.",
- zone));
+ final String zone;
+ if (zoneId instanceof ZoneOffset) {
+ // Give ZoneOffset a timezone for backwards compatibility reasons.
+ // In general, advertising either TZDB ID, GMT+xx:xx, or UTC is the best we can do.
+ zone = ZoneId.ofOffset("GMT", (ZoneOffset) zoneId).toString();
+ } else {
+ zone = zoneId.toString();
}
+ validateTimeZone(zone);
+
+ configuration.setString(TableConfigOptions.LOCAL_TIME_ZONE, zone);
}
/** Returns the current configuration of Planner for Table API and SQL queries. */
diff --git a/flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/api/internal/TableConfigValidation.java b/flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/api/internal/TableConfigValidation.java
new file mode 100644
index 00000000000..b6d1eae5f02
--- /dev/null
+++ b/flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/api/internal/TableConfigValidation.java
@@ -0,0 +1,55 @@
+/*
+ * 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.table.api.internal;
+
+import org.apache.flink.annotation.Internal;
+import org.apache.flink.table.api.TableConfig;
+import org.apache.flink.table.api.ValidationException;
+
+import java.time.ZoneId;
+
+/** Utilities around validation to keep {@link TableConfig} clean. */
+@Internal
+public class TableConfigValidation {
+
+ /** Validates user configured time zone. */
+ public static void validateTimeZone(String zone) {
+ boolean isValid;
+ try {
+ // We enforce a zone string that is compatible with both java.util.TimeZone and
+ // java.time.ZoneId to avoid bugs.
+ // In general, advertising either TZDB ID, GMT+xx:xx, or UTC is the best we can do.
+ isValid = java.util.TimeZone.getTimeZone(zone).toZoneId().equals(ZoneId.of(zone));
+ } catch (Exception e) {
+ isValid = false;
+ }
+
+ if (!isValid) {
+ throw new ValidationException(
+ "Invalid time zone. The value should be a Time Zone Database (TZDB) ID "
+ + "such as 'America/Los_Angeles' to include daylight saving time. Fixed "
+ + "offsets are supported using 'GMT-03:00' or 'GMT+03:00'. Or use 'UTC' "
+ + "without time zone and daylight saving time.");
+ }
+ }
+
+ private TableConfigValidation() {
+ // No instantiation
+ }
+}
diff --git a/flink-table/flink-table-api-java/src/test/java/org/apache/flink/table/api/TableConfigTest.java b/flink-table/flink-table-api-java/src/test/java/org/apache/flink/table/api/TableConfigTest.java
index 6b221711afb..81b786d2066 100644
--- a/flink-table/flink-table-api-java/src/test/java/org/apache/flink/table/api/TableConfigTest.java
+++ b/flink-table/flink-table-api-java/src/test/java/org/apache/flink/table/api/TableConfigTest.java
@@ -18,7 +18,6 @@
package org.apache.flink.table.api;
-import org.apache.flink.configuration.Configuration;
import org.apache.flink.table.api.config.TableConfigOptions;
import org.junit.jupiter.api.Test;
@@ -35,12 +34,10 @@ public class TableConfigTest {
private static final TableConfig CONFIG_BY_METHOD = TableConfig.getDefault();
private static final TableConfig CONFIG_BY_CONFIGURATION = TableConfig.getDefault();
- private static final Configuration configuration = new Configuration();
@Test
void testSetAndGetSqlDialect() {
- configuration.setString("table.sql-dialect", "HIVE");
- CONFIG_BY_CONFIGURATION.addConfiguration(configuration);
+ CONFIG_BY_CONFIGURATION.set("table.sql-dialect", "HIVE");
CONFIG_BY_METHOD.setSqlDialect(SqlDialect.HIVE);
assertThat(CONFIG_BY_METHOD.getSqlDialect()).isEqualTo(SqlDialect.HIVE);
@@ -49,8 +46,7 @@ public class TableConfigTest {
@Test
void testSetAndGetMaxGeneratedCodeLength() {
- configuration.setString("table.generated-code.max-length", "5000");
- CONFIG_BY_CONFIGURATION.addConfiguration(configuration);
+ CONFIG_BY_CONFIGURATION.set("table.generated-code.max-length", "5000");
CONFIG_BY_METHOD.setMaxGeneratedCodeLength(5000);
assertThat(CONFIG_BY_METHOD.getMaxGeneratedCodeLength()).isEqualTo(Integer.valueOf(5000));
@@ -60,29 +56,28 @@ public class TableConfigTest {
@Test
void testSetAndGetLocalTimeZone() {
- configuration.setString("table.local-time-zone", "Asia/Shanghai");
- CONFIG_BY_CONFIGURATION.addConfiguration(configuration);
+ CONFIG_BY_CONFIGURATION.set("table.local-time-zone", "Asia/Shanghai");
CONFIG_BY_METHOD.setLocalTimeZone(ZoneId.of("Asia/Shanghai"));
-
assertThat(CONFIG_BY_METHOD.getLocalTimeZone()).isEqualTo(ZoneId.of("Asia/Shanghai"));
assertThat(CONFIG_BY_CONFIGURATION.getLocalTimeZone())
.isEqualTo(ZoneId.of("Asia/Shanghai"));
- configuration.setString("table.local-time-zone", "GMT-08:00");
- CONFIG_BY_CONFIGURATION.addConfiguration(configuration);
+ CONFIG_BY_CONFIGURATION.set("table.local-time-zone", "GMT-08:00");
CONFIG_BY_METHOD.setLocalTimeZone(ZoneId.of("GMT-08:00"));
-
assertThat(CONFIG_BY_METHOD.getLocalTimeZone()).isEqualTo(ZoneId.of("GMT-08:00"));
assertThat(CONFIG_BY_CONFIGURATION.getLocalTimeZone()).isEqualTo(ZoneId.of("GMT-08:00"));
+
+ CONFIG_BY_CONFIGURATION.set("table.local-time-zone", "UTC");
+ CONFIG_BY_METHOD.setLocalTimeZone(ZoneId.of("UTC"));
+ assertThat(CONFIG_BY_METHOD.getLocalTimeZone()).isEqualTo(ZoneId.of("UTC"));
+ assertThat(CONFIG_BY_CONFIGURATION.getLocalTimeZone()).isEqualTo(ZoneId.of("UTC"));
}
@Test
public void testSetInvalidLocalTimeZone() {
assertThatThrownBy(() -> CONFIG_BY_METHOD.setLocalTimeZone(ZoneId.of("UTC-10:00")))
- .isInstanceOf(IllegalArgumentException.class)
- .hasMessage(
- "The supported Zone ID is either a full name such as 'America/Los_Angeles',"
- + " or a custom timezone id such as 'GMT-08:00', but configured Zone ID is 'UTC-10:00'.");
+ .isInstanceOf(ValidationException.class)
+ .hasMessageContaining("Invalid time zone.");
}
@Test
@@ -93,31 +88,32 @@ public class TableConfigTest {
}
@Test
- void testGetInvalidLocalTimeZone() {
- configuration.setString("table.local-time-zone", "UTC+8");
- CONFIG_BY_CONFIGURATION.addConfiguration(configuration);
+ void testGetInvalidLocalTimeZoneUTC() {
+ CONFIG_BY_CONFIGURATION.set("table.local-time-zone", "UTC+8");
+ assertThatThrownBy(CONFIG_BY_CONFIGURATION::getLocalTimeZone)
+ .isInstanceOf(ValidationException.class)
+ .hasMessageContaining("Invalid time zone.");
+ }
+
+ @Test
+ void testGetInvalidLocalTimeZoneUT() {
+ CONFIG_BY_CONFIGURATION.set("table.local-time-zone", "UT+8");
assertThatThrownBy(CONFIG_BY_CONFIGURATION::getLocalTimeZone)
- .isInstanceOf(IllegalArgumentException.class)
- .hasMessage(
- "The supported Zone ID is either a full name such as 'America/Los_Angeles',"
- + " or a custom timezone id such as 'GMT-08:00', but configured Zone ID is 'UTC+8'.");
+ .isInstanceOf(ValidationException.class)
+ .hasMessageContaining("Invalid time zone.");
}
@Test
void testGetInvalidAbbreviationLocalTimeZone() {
- configuration.setString("table.local-time-zone", "PST");
- CONFIG_BY_CONFIGURATION.addConfiguration(configuration);
+ CONFIG_BY_CONFIGURATION.set("table.local-time-zone", "PST");
assertThatThrownBy(CONFIG_BY_CONFIGURATION::getLocalTimeZone)
- .isInstanceOf(IllegalArgumentException.class)
- .hasMessage(
- "The supported Zone ID is either a full name such as 'America/Los_Angeles',"
- + " or a custom timezone id such as 'GMT-08:00', but configured Zone ID is 'PST'.");
+ .isInstanceOf(ValidationException.class)
+ .hasMessageContaining("Invalid time zone.");
}
@Test
void testSetAndGetIdleStateRetention() {
- configuration.setString("table.exec.state.ttl", "1 h");
- CONFIG_BY_CONFIGURATION.addConfiguration(configuration);
+ CONFIG_BY_CONFIGURATION.set("table.exec.state.ttl", "1 h");
CONFIG_BY_METHOD.setIdleStateRetention(Duration.ofHours(1));
assertThat(CONFIG_BY_METHOD.getIdleStateRetention()).isEqualTo(Duration.ofHours(1));
@@ -126,8 +122,7 @@ public class TableConfigTest {
@Test
void testDisplayMaxColumnLength() {
- configuration.setString("table.display.max-column-width", "100");
- CONFIG_BY_CONFIGURATION.addConfiguration(configuration);
+ CONFIG_BY_CONFIGURATION.set("table.display.max-column-width", "100");
CONFIG_BY_METHOD.set(TableConfigOptions.DISPLAY_MAX_COLUMN_WIDTH, 100);
assertThat(CONFIG_BY_METHOD.get(TableConfigOptions.DISPLAY_MAX_COLUMN_WIDTH))
diff --git a/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/utils/TableConfigUtils.java b/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/utils/TableConfigUtils.java
index c3526947438..534cfd7d6fa 100644
--- a/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/utils/TableConfigUtils.java
+++ b/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/utils/TableConfigUtils.java
@@ -30,9 +30,9 @@ import java.time.ZoneId;
import java.util.HashSet;
import java.util.Set;
-import static java.time.ZoneId.SHORT_IDS;
import static org.apache.flink.table.api.config.ExecutionConfigOptions.TABLE_EXEC_DISABLED_OPERATORS;
import static org.apache.flink.table.api.config.OptimizerConfigOptions.TABLE_OPTIMIZER_AGG_PHASE_STRATEGY;
+import static org.apache.flink.table.api.internal.TableConfigValidation.validateTimeZone;
/** Utility class for {@link TableConfig} related helper functions. */
public class TableConfigUtils {
@@ -101,11 +101,12 @@ public class TableConfigUtils {
* @see TableConfig#getLocalTimeZone()
*/
public static ZoneId getLocalTimeZone(ReadableConfig tableConfig) {
- String zone = tableConfig.get(TableConfigOptions.LOCAL_TIME_ZONE);
+ final String zone = tableConfig.get(TableConfigOptions.LOCAL_TIME_ZONE);
+ if (TableConfigOptions.LOCAL_TIME_ZONE.defaultValue().equals(zone)) {
+ return ZoneId.systemDefault();
+ }
validateTimeZone(zone);
- return TableConfigOptions.LOCAL_TIME_ZONE.defaultValue().equals(zone)
- ? ZoneId.systemDefault()
- : ZoneId.of(zone);
+ return ZoneId.of(zone);
}
/**
@@ -118,21 +119,6 @@ public class TableConfigUtils {
return tableConfig.get(ExecutionConfigOptions.IDLE_STATE_RETENTION).toMillis() * 3 / 2;
}
- /** Validates user configured time zone. */
- private static void validateTimeZone(String zone) {
- final String zoneId = zone.toUpperCase();
- if (zoneId.startsWith("UTC+")
- || zoneId.startsWith("UTC-")
- || SHORT_IDS.containsKey(zoneId)) {
- throw new IllegalArgumentException(
- String.format(
- "The supported Zone ID is either a full name such as "
- + "'America/Los_Angeles', or a custom timezone id such as "
- + "'GMT-08:00', but configured Zone ID is '%s'.",
- zone));
- }
- }
-
// Make sure that we cannot instantiate this class
private TableConfigUtils() {}
}
diff --git a/flink-table/flink-table-planner/src/test/java/org/apache/flink/connector/file/table/FileSystemTableFactoryTest.java b/flink-table/flink-table-planner/src/test/java/org/apache/flink/connector/file/table/FileSystemTableFactoryTest.java
index 9f87bbc5c8c..0ffb9b6bbff 100644
--- a/flink-table/flink-table-planner/src/test/java/org/apache/flink/connector/file/table/FileSystemTableFactoryTest.java
+++ b/flink-table/flink-table-planner/src/test/java/org/apache/flink/connector/file/table/FileSystemTableFactoryTest.java
@@ -132,9 +132,7 @@ public class FileSystemTableFactoryTest {
.satisfies(
anyCauseMatches(
ValidationException.class,
- "The supported watermark time zone is either a full name such "
- + "as 'America/Los_Angeles', or a custom time zone id such "
- + "as 'GMT-08:00', but configured time zone is 'UTC+8'."));
+ "Invalid time zone for 'sink.partition-commit.watermark-time-zone'."));
}
@Test