Merge branch 'feat_1.10_dealTimeZone' into 'v1.10.0_dev'

Feat 1.10 deal time zone

See merge request dt-insight-engine/flinkStreamSQL!70

Author: zoudaokoulife

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

@@ -70,11 +70,8 @@
 import java.net.URL;
 import java.net.URLClassLoader;
 import java.net.URLDecoder;
-import java.util.Arrays;
-import java.util.List;
-import java.util.Map;
-import java.util.Properties;
-import java.util.Set;
+import java.time.ZoneId;
+import java.util.*;
 
 /**
  *  任务执行时的流程方法
@@ -88,6 +85,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 {
@@ -358,9 +357,21 @@ public static StreamTableEnvironment getStreamTableEnv(StreamExecutionEnvironmen
                 .inStreamingMode()
                 .build();
 
-        StreamTableEnvironment tableEnv = StreamTableEnvironmentImpl.create(env, settings, new TableConfig());
+        TableConfig tableConfig = new TableConfig();
+
+        timeZoneCheck(confProperties.getProperty(TIME_ZONE));
+
+        tableConfig.setLocalTimeZone(ZoneId.of(confProperties.getProperty(TIME_ZONE)));
+
+        StreamTableEnvironment tableEnv = StreamTableEnvironmentImpl.create(env, settings, tableConfig);
         StreamEnvConfigManager.streamTableEnvironmentStateTTLConfig(tableEnv, confProperties);
         return tableEnv;
     }
 
+    private static 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/table/AbstractSourceTableInfo.java

@@ -21,13 +21,9 @@
 package com.dtstack.flink.sql.table;
 
 import com.google.common.base.Strings;
-import com.google.common.collect.Lists;
 import com.google.common.collect.Maps;
-import org.apache.flink.util.StringUtils;
 
-import java.util.ArrayList;
 import java.util.Map;
-import java.util.TimeZone;
 
 /**
  * Reason:
@@ -40,10 +36,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 +101,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

@@ -29,8 +29,6 @@
 import org.apache.flink.streaming.api.functions.timestamps.BoundedOutOfOrdernessTimestampExtractor;
 import org.apache.flink.streaming.api.windowing.time.Time;
 
-import java.util.TimeZone;
-
 /**
  * Reason:
  * Date: 2018/10/18
@@ -53,8 +51,6 @@ public abstract class AbstractCustomerWaterMarker<T> extends BoundedOutOfOrderne
 
     protected long lastTime = 0;
 
-    protected TimeZone timezone;
-
     public AbstractCustomerWaterMarker(Time maxOutOfOrderness) {
         super(maxOutOfOrderness);
     }
@@ -102,8 +98,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

@@ -26,8 +26,6 @@
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
-import java.util.TimeZone;
-
 /**
  * Custom watermark --- for eventtime
  * Date: 2017/12/28
@@ -39,10 +37,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(), "|")));