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

docs/content.zh/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/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();
     }
 

flink-table/flink-table-runtime-blink/src/main/java/org/apache/flink/table/filesystem/FileSystemOptions.java

@@ -229,7 +229,7 @@ public class FileSystemOptions {
                                     + " 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 user configured time zone,"
                                     + " the the value should be the user configured local 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'.");
+                                    + " such as 'America/Los_Angeles', or a custom timezone id such as 'GMT-08:00'.");
 
     public static final ConfigOption<String> SINK_PARTITION_COMMIT_POLICY_KIND =
             key("sink.partition-commit.policy.kind")

flink-table/flink-table-runtime-blink/src/main/java/org/apache/flink/table/filesystem/FileSystemTableFactory.java

@@ -133,7 +133,7 @@ private void validate(FactoryUtil.TableFactoryHelper helper) {
             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-8:00', but configured time zone is '%s'.",
+                                    + " or a custom time zone id such as 'GMT-08:00', but configured time zone is '%s'.",
                             watermarkTimeZone));
         }
     }

flink-table/flink-table-runtime-blink/src/main/java/org/apache/flink/table/runtime/util/TimeWindowUtil.java

@@ -43,8 +43,8 @@ public class TimeWindowUtil {
     /**
      * Convert a epoch mills to timestamp mills which can describe a locate date time.
      *
-     * <p>For example: The timestamp string of epoch mills 5 in GMT+8:00 is 1970-01-01 08:00:05, the
-     * timestamp mills is 8 * 60 * 60 * 1000 + 5.
+     * <p>For example: The timestamp string of epoch mills 5 in GMT+08:00 is 1970-01-01 08:00:05,
+     * the timestamp mills is 8 * 60 * 60 * 1000 + 5.
      *
      * @param epochMills the epoch mills.
      * @param shiftTimeZone the timezone that the given timestamp mills has been shifted.

flink-table/flink-table-runtime-blink/src/test/java/org/apache/flink/table/filesystem/FileSystemTableFactoryTest.java

@@ -166,7 +166,7 @@ public void testUnsupportedWatermarkTimeZone() {
                             .contains(
                                     "The supported watermark time zone is either a full name such "
                                             + "as 'America/Los_Angeles', or a custom time zone id such "
-                                            + "as 'GMT-8:00', but configured time zone is 'UTC+8'."));
+                                            + "as 'GMT-08:00', but configured time zone is 'UTC+8'."));
             return;
         }
 

flink-table/flink-table-runtime-blink/src/test/java/org/apache/flink/table/runtime/operators/aggregate/window/SlicingWindowAggOperatorTest.java

@@ -692,8 +692,8 @@ public void testProcessingTimeTumblingWindows() throws Exception {
 
         final SliceAssigner assigner =
                 SliceAssigners.tumbling(-1, shiftTimeZone, Duration.ofHours(5));
-        // the assigned windows should like as following, e.g. the given timeZone is GMT+8:00:
-        //  local windows(timestamp in GMT+8:00)   <=>  epoch windows(timestamp in UTC)
+        // the assigned windows should like as following, e.g. the given timeZone is GMT+08:00:
+        //  local windows(timestamp in GMT+08:00)   <=>  epoch windows(timestamp in UTC)
         // [1970-01-01 00:00, 1970-01-01 05:00] <=> [1969-12-31 16:00, 1969-12-31 21:00]
         // [1970-01-01 05:00, 1970-01-01 10:00] <=> [1969-12-31 21:00, 1970-01-01 02:00]