[FLINK-36572][pipeline-connector][starrocks] Fix the issue that the local time zone is wrongly set

This closes  apache#3733.

Author: herunkang2018

flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-starrocks/src/main/java/org/apache/flink/cdc/connectors/starrocks/sink/StarRocksDataSink.java

@@ -17,6 +17,7 @@
 
 package org.apache.flink.cdc.connectors.starrocks.sink;
 
+import org.apache.flink.cdc.common.annotation.VisibleForTesting;
 import org.apache.flink.cdc.common.event.Event;
 import org.apache.flink.cdc.common.sink.DataSink;
 import org.apache.flink.cdc.common.sink.EventSinkProvider;
@@ -78,4 +79,9 @@ public MetadataApplier getMetadataApplier() {
                         sinkOptions.getPassword());
         return new StarRocksMetadataApplier(catalog, tableCreateConfig, schemaChangeConfig);
     }
+
+    @VisibleForTesting
+    public ZoneId getZoneId() {
+        return zoneId;
+    }
 }

flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-starrocks/src/main/java/org/apache/flink/cdc/connectors/starrocks/sink/StarRocksDataSinkFactory.java

@@ -51,7 +51,7 @@ public DataSink createDataSink(Context context) {
                 TableCreateConfig.from(context.getFactoryConfiguration());
         SchemaChangeConfig schemaChangeConfig =
                 SchemaChangeConfig.from(context.getFactoryConfiguration());
-        String zoneStr = context.getFactoryConfiguration().get(PIPELINE_LOCAL_TIME_ZONE);
+        String zoneStr = context.getPipelineConfiguration().get(PIPELINE_LOCAL_TIME_ZONE);
         ZoneId zoneId =
                 PIPELINE_LOCAL_TIME_ZONE.defaultValue().equals(zoneStr)
                         ? ZoneId.systemDefault()

flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-starrocks/src/test/java/org/apache/flink/cdc/connectors/starrocks/sink/StarRocksDataSinkFactoryTest.java

@@ -30,6 +30,7 @@
 import org.assertj.core.api.Assertions;
 import org.junit.jupiter.api.Test;
 
+import java.time.ZoneId;
 import java.util.HashMap;
 import java.util.List;
 import java.util.Map;
@@ -150,4 +151,36 @@ void testPrefixRequireOption() {
                                 conf, conf, Thread.currentThread().getContextClassLoader()));
         Assertions.assertThat(dataSink).isInstanceOf(StarRocksDataSink.class);
     }
+
+    @Test
+    void testCreateDataSinkWithSpecificedTimeZone() {
+        DataSinkFactory sinkFactory =
+                FactoryDiscoveryUtils.getFactoryByIdentifier("starrocks", DataSinkFactory.class);
+        Assertions.assertThat(sinkFactory).isInstanceOf(StarRocksDataSinkFactory.class);
+
+        Configuration factoryConfiguration =
+                Configuration.fromMap(
+                        ImmutableMap.<String, String>builder()
+                                .put("jdbc-url", "jdbc:mysql://127.0.0.1:9030")
+                                .put("load-url", "127.0.0.1:8030")
+                                .put("username", "root")
+                                .put("password", "")
+                                .build());
+        Configuration pipelineConfiguration =
+                Configuration.fromMap(
+                        ImmutableMap.<String, String>builder()
+                                .put("local-time-zone", "America/Los_Angeles")
+                                .build());
+        DataSink dataSink =
+                sinkFactory.createDataSink(
+                        new FactoryHelper.DefaultContext(
+                                factoryConfiguration,
+                                pipelineConfiguration,
+                                Thread.currentThread().getContextClassLoader()));
+        Assertions.assertThat(dataSink).isInstanceOf(StarRocksDataSink.class);
+
+        ZoneId zoneId = ((StarRocksDataSink) dataSink).getZoneId();
+        ZoneId expectedZondId = ZoneId.of("America/Los_Angeles");
+        Assertions.assertThat(zoneId).isEqualTo(expectedZondId);
+    }
 }