【opt】opt timezone

Author: FlechazoW

core/src/main/java/com/dtstack/flink/sql/exec/ExecuteProcessHelper.java

@@ -25,19 +25,15 @@
 import com.dtstack.flink.sql.parser.SqlParser;
 import com.dtstack.flink.sql.parser.SqlTree;
 import org.apache.flink.api.common.typeinfo.TypeInformation;
-import org.apache.flink.api.java.tuple.Tuple2;
 import org.apache.flink.api.java.typeutils.RowTypeInfo;
 import org.apache.flink.streaming.api.datastream.DataStream;
 import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
 import org.apache.flink.table.api.*;
-import org.apache.flink.table.api.internal.TableEnvironmentImpl;
 import org.apache.flink.table.api.java.StreamTableEnvironment;
 import org.apache.flink.table.api.java.internal.StreamTableEnvironmentImpl;
 import org.apache.flink.table.sinks.TableSink;
-import org.apache.flink.types.Row;
 
 import com.dtstack.flink.sql.classloader.ClassLoaderManager;
-import com.dtstack.flink.sql.constrant.ConfigConstrant;
 import com.dtstack.flink.sql.enums.ClusterMode;
 import com.dtstack.flink.sql.enums.ECacheType;
 import com.dtstack.flink.sql.enums.EPluginLoadMode;
@@ -74,6 +70,7 @@
 import java.net.URL;
 import java.net.URLClassLoader;
 import java.net.URLDecoder;
+import java.time.ZoneId;
 import java.util.Arrays;
 import java.util.List;
 import java.util.Map;
@@ -92,6 +89,8 @@ public class ExecuteProcessHelper {
     private static final Logger LOG = LoggerFactory.getLogger(ExecuteProcessHelper.class);
     private static final ObjectMapper OBJECT_MAPPER = new ObjectMapper();
 
+    private static final String TIME_ZONE = "timezone";
+
     public static FlinkPlanner flinkPlanner = new FlinkPlanner();
 
     public static ParamsInfo parseParams(String[] args) throws Exception {
@@ -108,7 +107,6 @@ public static ParamsInfo parseParams(String[] args) throws Exception {
         String remoteSqlPluginPath = options.getRemoteSqlPluginPath();
         String pluginLoadMode = options.getPluginLoadMode();
         String deployMode = options.getMode();
-        String logLevel = options.getLogLevel();
 
         Preconditions.checkArgument(checkRemoteSqlPluginPath(remoteSqlPluginPath, deployMode, pluginLoadMode),
                 "Non-local mode or shipfile deployment mode, remoteSqlPluginPath is required");
@@ -358,7 +356,11 @@ private static StreamTableEnvironment getStreamTableEnv(StreamExecutionEnvironme
                 .inStreamingMode()
                 .build();
 
-        StreamTableEnvironment tableEnv = StreamTableEnvironmentImpl.create(env, settings, new TableConfig());
+        TableConfig tableConfig = new TableConfig();
+
+        tableConfig.setLocalTimeZone(ZoneId.of(confProperties.getProperty(TIME_ZONE)));
+
+        StreamTableEnvironment tableEnv = StreamTableEnvironmentImpl.create(env, settings, tableConfig);
         StreamEnvConfigManager.streamTableEnvironmentStateTTLConfig(tableEnv, confProperties);
         return tableEnv;
     }

core/src/main/java/com/dtstack/flink/sql/table/AbstractSourceTableInfo.java

@@ -40,10 +40,6 @@ public abstract class AbstractSourceTableInfo extends AbstractTableInfo {
 
     public static final String SOURCE_SUFFIX = "Source";
 
-    public static final String TIME_ZONE_KEY="timezone";
-
-    private String timeZone=TimeZone.getDefault().getID();
-
     private String eventTimeField;
 
     private Integer maxOutOrderness = 10;
@@ -109,23 +105,4 @@ public String getAdaptSelectSql(){
     public String getAdaptName(){
         return getName() + "_adapt";
     }
-
-    public String getTimeZone() {
-        return timeZone;
-    }
-
-    public void setTimeZone(String timeZone) {
-        if (StringUtils.isNullOrWhitespaceOnly(timeZone)){
-            return;
-        }
-        timeZoneCheck(timeZone);
-        this.timeZone = timeZone;
-    }
-
-    private void timeZoneCheck(String timeZone) {
-        ArrayList<String> zones = Lists.newArrayList(TimeZone.getAvailableIDs());
-        if (!zones.contains(timeZone)){
-            throw  new IllegalArgumentException(" timezone is Incorrect!");
-        }
-    }
 }

core/src/main/java/com/dtstack/flink/sql/watermarker/AbstractCustomerWaterMarker.java

@@ -53,8 +53,6 @@ public abstract class AbstractCustomerWaterMarker<T> extends BoundedOutOfOrderne
 
     protected long lastTime = 0;
 
-    protected TimeZone timezone;
-
     public AbstractCustomerWaterMarker(Time maxOutOfOrderness) {
         super(maxOutOfOrderness);
     }
@@ -102,8 +100,7 @@ public void setFromSourceTag(String fromSourceTag) {
 
     protected long getExtractTimestamp(Long extractTime){
 
-        lastTime = extractTime + timezone.getOffset(extractTime);
-
+        lastTime = extractTime;
         eventDelayGauge.setDelayTime(MathUtil.getIntegerVal((System.currentTimeMillis() - extractTime)/1000));
 
         return lastTime;

core/src/main/java/com/dtstack/flink/sql/watermarker/CustomerWaterMarkerForLong.java

@@ -39,10 +39,9 @@ public class CustomerWaterMarkerForLong extends AbstractCustomerWaterMarker<Row>
 
     private static final Logger logger = LoggerFactory.getLogger(CustomerWaterMarkerForLong.class);
 
-    public CustomerWaterMarkerForLong(Time maxOutOfOrderness, int pos,String timezone) {
+    public CustomerWaterMarkerForLong(Time maxOutOfOrderness, int pos) {
         super(maxOutOfOrderness);
         this.pos = pos;
-        this.timezone= TimeZone.getTimeZone(timezone);
     }
 
     @Override

core/src/main/java/com/dtstack/flink/sql/watermarker/CustomerWaterMarkerForTimeStamp.java

@@ -39,10 +39,9 @@ public class CustomerWaterMarkerForTimeStamp extends AbstractCustomerWaterMarker
 
     private static final Logger logger = LoggerFactory.getLogger(CustomerWaterMarkerForTimeStamp.class);
 
-    public CustomerWaterMarkerForTimeStamp(Time maxOutOfOrderness, int pos,String timezone) {
+    public CustomerWaterMarkerForTimeStamp(Time maxOutOfOrderness, int pos) {
         super(maxOutOfOrderness);
         this.pos = pos;
-        this.timezone= TimeZone.getTimeZone(timezone);
     }
 
     @Override

core/src/main/java/com/dtstack/flink/sql/watermarker/WaterMarkerAssigner.java

@@ -48,8 +48,6 @@ public DataStream assignWaterMarker(DataStream<Row> dataStream, RowTypeInfo type
 
         int maxOutOrderness = sourceTableInfo.getMaxOutOrderness();
 
-        String timeZone=sourceTableInfo.getTimeZone();
-
         String[] fieldNames = typeInfo.getFieldNames();
         TypeInformation<?>[] fieldTypes = typeInfo.getFieldTypes();
 
@@ -71,9 +69,9 @@ public DataStream assignWaterMarker(DataStream<Row> dataStream, RowTypeInfo type
 
         AbstractCustomerWaterMarker waterMarker = null;
         if(fieldType.getTypeClass().isAssignableFrom(Timestamp.class)){
-            waterMarker = new CustomerWaterMarkerForTimeStamp(Time.milliseconds(maxOutOrderness), pos,timeZone);
+            waterMarker = new CustomerWaterMarkerForTimeStamp(Time.milliseconds(maxOutOrderness), pos);
         }else if(fieldType.getTypeClass().isAssignableFrom(Long.class)){
-            waterMarker = new CustomerWaterMarkerForLong(Time.milliseconds(maxOutOrderness), pos,timeZone);
+            waterMarker = new CustomerWaterMarkerForLong(Time.milliseconds(maxOutOrderness), pos);
         }else{
             throw new IllegalArgumentException("not support type of " + fieldType + ", current only support(timestamp, long).");
         }

kafka-base/kafka-base-source/src/main/java/com/dtstack/flink/sql/source/kafka/table/KafkaSourceParser.java

@@ -50,7 +50,6 @@ public AbstractTableInfo getTableInfo(String tableName, String fieldsInfo, Map<S
         kafkaSourceTableInfo.setTopic(MathUtil.getString(props.get(KafkaSourceTableInfo.TOPIC_KEY.toLowerCase())));
         kafkaSourceTableInfo.setTopicIsPattern(MathUtil.getBoolean(props.get(KafkaSourceTableInfo.TOPICISPATTERN_KEY.toLowerCase()), false));
         kafkaSourceTableInfo.setOffsetReset(MathUtil.getString(props.getOrDefault(KafkaSourceTableInfo.OFFSETRESET_KEY.toLowerCase(), EKafkaOffset.LATEST.name().toLowerCase())));
-        kafkaSourceTableInfo.setTimeZone(MathUtil.getString(props.get(KafkaSourceTableInfo.TIME_ZONE_KEY.toLowerCase())));
 
         kafkaSourceTableInfo.setSchemaString(MathUtil.getString(props.get(KafkaSourceTableInfo.SCHEMA_STRING_KEY.toLowerCase())));
         kafkaSourceTableInfo.setFieldDelimiter(MathUtil.getString(props.getOrDefault(KafkaSourceTableInfo.CSV_FIELD_DELIMITER_KEY.toLowerCase(), "|")));