[FLINK-22525][table-api] Fix gmt format in Flink from GMT-8:00 to GMT-08:00

Author: leonardBang

docs/content.zh/docs/connectors/table/filesystem.md

@@ -224,7 +224,7 @@ file sink 支持文件合并，以允许应用程序可以使用较小的检查
             该参数只有在参数 `sink.partition-commit.trigger` 被设置为 'partition-time' 时才生效。
             如果该参数设置的不正确，比如在 TIMESTAMP_LTZ 列上定义了 source rowtime, 但没有设置该参数，则用户可能在若干个小时后才看到分区的提交。
             该参数的默认值是 'UTC', 代表水印是定义在 TIMESTAMP 列上或没有定义水印。 如果水印定义在 TIMESTAMP_LTZ 列上，则水印的时区是会话的时区。
-            该参数的可选值要么是完整的时区名比如 'America/Los_Angeles'，要么是自定义的时区 id 比如 'GMT-8:00'.</td>
+            该参数的可选值要么是完整的时区名比如 'America/Los_Angeles'，要么是自定义的时区 id 比如 'GMT-08:00'.</td>
     </tr>        
   </tbody>
 </table>

docs/content/docs/connectors/table/filesystem.md

@@ -229,7 +229,7 @@ To define when to commit a partition, providing partition commit trigger:
         <td><h5>sink.partition-commit.watermark-time-zone</h5></td>
         <td style="word-wrap: break-word;">UTC</td>
         <td>String</td>
-        <td>The time zone to parse the long watermark value to TIMESTAMP value, the parsed watermark timestamp is used to compare with partition time to decide the partition should commit or not. This option is only take effect when `sink.partition-commit.trigger` is set to 'partition-time'. If this option is not configured correctly, e.g. source rowtime is defined on TIMESTAMP_LTZ column, but this config is not configured, then users may see the partition committed after a few hours. The default value is 'UTC', which means the watermark is defined on TIMESTAMP column or not defined. If the watermark is defined on TIMESTAMP_LTZ column, the time zone of watermark is the session time zone. The option value is either a full name such as 'America/Los_Angeles', or a custom timezone id such as 'GMT-8:00'.</td>
+        <td>The time zone to parse the long watermark value to TIMESTAMP value, the parsed watermark timestamp is used to compare with partition time to decide the partition should commit or not. This option is only take effect when `sink.partition-commit.trigger` is set to 'partition-time'. If this option is not configured correctly, e.g. source rowtime is defined on TIMESTAMP_LTZ column, but this config is not configured, then users may see the partition committed after a few hours. The default value is 'UTC', which means the watermark is defined on TIMESTAMP column or not defined. If the watermark is defined on TIMESTAMP_LTZ column, the time zone of watermark is the session time zone. The option value is either a full name such as 'America/Los_Angeles', or a custom timezone id such as 'GMT-08:00'.</td>
     </tr>    
   </tbody>
 </table>

docs/data/sql_functions.yml

@@ -471,7 +471,7 @@ temporal:
     table: timestampDiff(TIMEPOINTUNIT, TIMEPOINT1, TIMEPOINT2)
     description: 'Returns the (signed) number of timepointunit between timepoint1 and timepoint2. The unit for the interval is given by the first argument, which should be one of the following values: SECOND, MINUTE, HOUR, DAY, MONTH, or YEAR.' 
   - sql: CONVERT_TZ(string1, string2, string3)
-    description: Converts a datetime string1 (with default ISO timestamp format 'yyyy-MM-dd HH:mm:ss') from time zone string2 to time zone string3. The format of time zone should be either an abbreviation such as "PST", a full name such as "America/Los_Angeles", or a custom ID such as "GMT-8:00". E.g., CONVERT_TZ('1970-01-01 00:00:00', 'UTC', 'America/Los_Angeles') returns '1969-12-31 16:00:00'.
+    description: Converts a datetime string1 (with default ISO timestamp format 'yyyy-MM-dd HH:mm:ss') from time zone string2 to time zone string3. The format of time zone should be either an abbreviation such as "PST", a full name such as "America/Los_Angeles", or a custom ID such as "GMT-08:00". E.g., CONVERT_TZ('1970-01-01 00:00:00', 'UTC', 'America/Los_Angeles') returns '1969-12-31 16:00:00'.
   - sql: FROM_UNIXTIME(numeric[, string])
     description: Returns a representation of the numeric argument as a value in string format (default is 'yyyy-MM-dd HH:mm:ss'). numeric is an internal timestamp value representing seconds since '1970-01-01 00:00:00' UTC, such as produced by the UNIX_TIMESTAMP() function. The return value is expressed in the session time zone (specified in TableConfig). E.g., FROM_UNIXTIME(44) returns '1970-01-01 00:00:44' if in UTC time zone, but returns '1970-01-01 09:00:44' if in 'Asia/Tokyo' time zone.
   - sql: UNIX_TIMESTAMP()

docs/layouts/shortcodes/generated/table_config_configuration.html

@@ -30,7 +30,7 @@
             <td><h5>table.local-time-zone</h5><br> <span class="label label-primary">Batch</span> <span class="label label-primary">Streaming</span></td>
             <td style="word-wrap: break-word;">"default"</td>
             <td>String</td>
-            <td>The local time zone defines current session time zone id. It is used when converting to/from &lt;code&gt;TIMESTAMP WITH LOCAL TIME ZONE&lt;/code&gt;. Internally, timestamps with local time zone are always represented in the UTC time zone. However, when converting to data types that don't include a time zone (e.g. TIMESTAMP, TIME, or simply STRING), the session time zone is used during conversion. The input of option is either a full name such as "America/Los_Angeles", or a custom timezone id such as "GMT-8:00".</td>
+            <td>The local time zone defines current session time zone id. It is used when converting to/from &lt;code&gt;TIMESTAMP WITH LOCAL TIME ZONE&lt;/code&gt;. Internally, timestamps with local time zone are always represented in the UTC time zone. However, when converting to data types that don't include a time zone (e.g. TIMESTAMP, TIME, or simply STRING), the session time zone is used during conversion. The input of option is either a full name such as "America/Los_Angeles", or a custom timezone id such as "GMT-08:00".</td>
         </tr>
         <tr>
             <td><h5>table.planner</h5><br> <span class="label label-primary">Batch</span> <span class="label label-primary">Streaming</span></td>

flink-python/pyflink/datastream/connectors.py

@@ -573,7 +573,7 @@ def date_time_bucket_assigner(format_str: str = "yyyy-MM-dd--HH", timezone_id: s
         :param format_str: The format string used to determine the bucket id.
         :param timezone_id: The timezone id, either an abbreviation such as "PST", a full name
                             such as "America/Los_Angeles", or a custom timezone_id such as
-                            "GMT-8:00". Th e default time zone will b used if it's None.
+                            "GMT-08:00". Th e default time zone will b used if it's None.
         """
         if timezone_id is not None and isinstance(timezone_id, str):
             j_timezone = get_gateway().jvm.java.time.ZoneId.of(timezone_id)

flink-python/pyflink/table/table_config.py

@@ -56,7 +56,7 @@ def get_local_timezone(self) -> str:
         """
         Returns the local timezone id for timestamp with local time zone, either an abbreviation
         such as "PST", a full name such as "America/Los_Angeles", or a custom timezone_id such
-        as "GMT-8:00".
+        as "GMT-08:00".
         """
         return self._j_table_config.getLocalTimeZone().getId()
 
@@ -66,7 +66,7 @@ def set_local_timezone(self, timezone_id: str):
 
         :param timezone_id: The timezone id, either an abbreviation such as "PST", a full name
                             such as "America/Los_Angeles", or a custom timezone_id such as
-                            "GMT-8:00".
+                            "GMT-08:00".
         """
         if timezone_id is not None and isinstance(timezone_id, str):
             j_timezone = get_gateway().jvm.java.time.ZoneId.of(timezone_id)

flink-streaming-java/src/test/java/org/apache/flink/streaming/runtime/operators/windowing/TimeWindowTest.java

@@ -53,7 +53,7 @@ public void testGetWindowStartWithOffset() {
         Assert.assertEquals(TimeWindow.getWindowStartWithOffset(7, offset, 7), 5);
         Assert.assertEquals(TimeWindow.getWindowStartWithOffset(12, offset, 7), 12);
 
-        // for GMT+8:00
+        // for GMT+08:00
         offset = -TimeUnit.HOURS.toMillis(8);
         long size = TimeUnit.DAYS.toMillis(1);
         Assert.assertEquals(

flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/api/TableConfig.java

@@ -182,7 +182,7 @@ private void validateTimeZone(String zone) {
             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-8:00', but configured Zone ID is '%s'.",
+                                    + " or a custom timezone id such as 'GMT-08:00', but configured Zone ID is '%s'.",
                             zone));
         }
     }

flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/api/config/TableConfigOptions.java

@@ -90,7 +90,7 @@ private TableConfigOptions() {}
                                     + "<code>TIMESTAMP WITH LOCAL TIME ZONE</code>. Internally, timestamps with local time zone are always represented in the UTC time zone. "
                                     + "However, when converting to data types that don't include a time zone (e.g. TIMESTAMP, TIME, or simply STRING), "
                                     + "the session time zone is used during conversion. The input of option is either a full name "
-                                    + "such as \"America/Los_Angeles\", or a custom timezone id such as \"GMT-8:00\".");
+                                    + "such as \"America/Los_Angeles\", or a custom timezone id such as \"GMT-08:00\".");
 
     @Documentation.TableOption(execMode = Documentation.ExecMode.BATCH_STREAMING)
     public static final ConfigOption<Integer> MAX_LENGTH_GENERATED_CODE =

flink-table/flink-table-api-java/src/test/java/org/apache/flink/table/api/TableConfigTest.java

@@ -66,23 +66,36 @@ public void testSetAndGetLocalTimeZone() {
 
         assertEquals(ZoneId.of("Asia/Shanghai"), configByMethod.getLocalTimeZone());
         assertEquals(ZoneId.of("Asia/Shanghai"), configByConfiguration.getLocalTimeZone());
+
+        configuration.setString("table.local-time-zone", "GMT-08:00");
+        configByConfiguration.addConfiguration(configuration);
+        configByMethod.setLocalTimeZone(ZoneId.of("GMT-08:00"));
+
+        assertEquals(ZoneId.of("GMT-08:00"), configByMethod.getLocalTimeZone());
+        assertEquals(ZoneId.of("GMT-08:00"), configByConfiguration.getLocalTimeZone());
     }
 
     @Test
     public void testSetInvalidLocalTimeZone() {
         expectedException.expectMessage(
                 "The supported Zone ID is either a full name such as 'America/Los_Angeles',"
-                        + " or a custom timezone id such as 'GMT-8:00', but configured Zone ID is 'UTC-10:00'.");
+                        + " or a custom timezone id such as 'GMT-08:00', but configured Zone ID is 'UTC-10:00'.");
         configByMethod.setLocalTimeZone(ZoneId.of("UTC-10:00"));
     }
 
+    @Test
+    public void testInvalidGmtLocalTimeZone() {
+        expectedException.expectMessage("Invalid ID for offset-based ZoneId: GMT-8:00");
+        configByMethod.setLocalTimeZone(ZoneId.of("GMT-8:00"));
+    }
+
     @Test
     public void testGetInvalidLocalTimeZone() {
         configuration.setString("table.local-time-zone", "UTC+8");
         configByConfiguration.addConfiguration(configuration);
         expectedException.expectMessage(
                 "The supported Zone ID is either a full name such as 'America/Los_Angeles',"
-                        + " or a custom timezone id such as 'GMT-8:00', but configured Zone ID is 'UTC+8'.");
+                        + " or a custom timezone id such as 'GMT-08:00', but configured Zone ID is 'UTC+8'.");
         configByConfiguration.getLocalTimeZone();
     }
 
@@ -92,7 +105,7 @@ public void testGetInvalidAbbreviationLocalTimeZone() {
         configByConfiguration.addConfiguration(configuration);
         expectedException.expectMessage(
                 "The supported Zone ID is either a full name such as 'America/Los_Angeles',"
-                        + " or a custom timezone id such as 'GMT-8:00', but configured Zone ID is 'PST'.");
+                        + " or a custom timezone id such as 'GMT-08:00', but configured Zone ID is 'PST'.");
         configByConfiguration.getLocalTimeZone();
     }