convert rowtime timezone

todd5167 · todd5167 · commit dfe3f6bb6371 · 2019-04-10T16:08:56.000+08:00
diff --git a/core/src/main/java/com/dtstack/flink/sql/table/SourceTableInfo.java b/core/src/main/java/com/dtstack/flink/sql/table/SourceTableInfo.java
@@ -21,9 +21,12 @@
 package com.dtstack.flink.sql.table;
 
 import org.apache.flink.calcite.shaded.com.google.common.base.Strings;
+import org.apache.flink.calcite.shaded.com.google.common.collect.Lists;
 import org.apache.flink.calcite.shaded.com.google.common.collect.Maps;
 
+import java.util.ArrayList;
 import java.util.Map;
+import java.util.TimeZone;
 
 /**
  * Reason:
@@ -36,6 +39,10 @@ public abstract class SourceTableInfo extends TableInfo {
 
     public static final String SOURCE_SUFFIX = "Source";
 
+    public static final String TIME_ZONE_KEY="timezone";
+
+    private String timeZone="Asia/Shanghai";
+
     private String eventTimeField;
 
     private Integer maxOutOrderness = 10;
@@ -101,4 +108,23 @@ public String getAdaptSelectSql(){
     public String getAdaptName(){
         return getName() + "_adapt";
     }
+
+    public String getTimeZone() {
+        return timeZone;
+    }
+
+    public void setTimeZone(String timeZone) {
+        if (timeZone==null){
+            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!");
+        }
+    }
 }
diff --git a/core/src/main/java/com/dtstack/flink/sql/watermarker/CustomerWaterMarkerForLong.java b/core/src/main/java/com/dtstack/flink/sql/watermarker/CustomerWaterMarkerForLong.java
@@ -16,17 +16,18 @@
  * limitations under the License.
  */
 
- 
+
 
 package com.dtstack.flink.sql.watermarker;
 
 import com.dtstack.flink.sql.util.MathUtil;
-import org.apache.flink.streaming.api.functions.timestamps.BoundedOutOfOrdernessTimestampExtractor;
 import org.apache.flink.streaming.api.windowing.time.Time;
 import org.apache.flink.types.Row;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
+import java.util.TimeZone;
+
 /**
  * Custom watermark --- for eventtime
  * Date: 2017/12/28
@@ -44,23 +45,34 @@ public class CustomerWaterMarkerForLong extends AbsCustomerWaterMarker<Row> {
 
     private long lastTime = 0;
 
-    public CustomerWaterMarkerForLong(Time maxOutOfOrderness, int pos) {
+    private TimeZone timezone;
+
+    public CustomerWaterMarkerForLong(Time maxOutOfOrderness, int pos,String timezone) {
         super(maxOutOfOrderness);
         this.pos = pos;
+        this.timezone= TimeZone.getTimeZone(timezone);
     }
 
     @Override
     public long extractTimestamp(Row row) {
 
         try{
             Long eveTime = MathUtil.getLongVal(row.getField(pos));
-            lastTime = eveTime;
-            eventDelayGauge.setDelayTime(MathUtil.getIntegerVal((System.currentTimeMillis() - eveTime)/1000));
-            return eveTime;
+            Long extractTime=eveTime;
+
+            lastTime = extractTime + timezone.getOffset(extractTime);
+
+            eventDelayGauge.setDelayTime(MathUtil.getIntegerVal((System.currentTimeMillis() - convertTimeZone(extractTime))/1000));
+
+            return lastTime;
         }catch (Exception e){
             logger.error("", e);
         }
-
         return lastTime;
     }
+
+    public long convertTimeZone(long evenTime){
+        long res = evenTime - timezone.getOffset(evenTime) + TimeZone.getDefault().getOffset(evenTime);
+        return res;
+    }
 }
diff --git a/core/src/main/java/com/dtstack/flink/sql/watermarker/CustomerWaterMarkerForTimeStamp.java b/core/src/main/java/com/dtstack/flink/sql/watermarker/CustomerWaterMarkerForTimeStamp.java
@@ -27,6 +27,7 @@
 import org.slf4j.LoggerFactory;
 
 import java.sql.Timestamp;
+import java.util.TimeZone;
 
 /**
  * Custom watermark --- for eventtime
@@ -45,25 +46,34 @@ public class CustomerWaterMarkerForTimeStamp extends AbsCustomerWaterMarker<Row>
 
     private long lastTime = 0;
 
+    private TimeZone timezone;
 
-    public CustomerWaterMarkerForTimeStamp(Time maxOutOfOrderness, int pos) {
+    public CustomerWaterMarkerForTimeStamp(Time maxOutOfOrderness, int pos,String timezone) {
         super(maxOutOfOrderness);
         this.pos = pos;
+        this.timezone= TimeZone.getTimeZone(timezone);
     }
 
     @Override
     public long extractTimestamp(Row row) {
         try {
             Timestamp time = (Timestamp) row.getField(pos);
-            lastTime = time.getTime();
 
-            eventDelayGauge.setDelayTime(MathUtil.getIntegerVal((System.currentTimeMillis() - time.getTime())/1000));
-            return time.getTime();
+            long extractTime=time.getTime();
+
+            lastTime = extractTime + timezone.getOffset(extractTime);
+
+            eventDelayGauge.setDelayTime(MathUtil.getIntegerVal((System.currentTimeMillis() - convertTimeZone(extractTime))/1000));
+
+            return lastTime;
         } catch (RuntimeException e) {
             logger.error("", e);
         }
         return lastTime;
     }
 
-
+    public long convertTimeZone(long evenTime){
+        long res = evenTime - timezone.getOffset(evenTime) + TimeZone.getDefault().getOffset(evenTime);
+        return res;
+    }
 }
diff --git a/core/src/main/java/com/dtstack/flink/sql/watermarker/WaterMarkerAssigner.java b/core/src/main/java/com/dtstack/flink/sql/watermarker/WaterMarkerAssigner.java
@@ -54,6 +54,8 @@ public DataStream assignWaterMarker(DataStream<Row> dataStream, RowTypeInfo type
 
         int maxOutOrderness = sourceTableInfo.getMaxOutOrderness();
 
+        String timeZone=sourceTableInfo.getTimeZone();
+
         String[] fieldNames = typeInfo.getFieldNames();
         TypeInformation<?>[] fieldTypes = typeInfo.getFieldTypes();
 
@@ -75,9 +77,9 @@ public DataStream assignWaterMarker(DataStream<Row> dataStream, RowTypeInfo type
 
         AbsCustomerWaterMarker waterMarker = null;
         if(fieldType.getTypeClass().getTypeName().equalsIgnoreCase("java.sql.Timestamp")){
-            waterMarker = new CustomerWaterMarkerForTimeStamp(Time.milliseconds(maxOutOrderness), pos);
+            waterMarker = new CustomerWaterMarkerForTimeStamp(Time.milliseconds(maxOutOrderness), pos,timeZone);
         }else if(fieldType.getTypeClass().getTypeName().equalsIgnoreCase("java.lang.Long")){
-            waterMarker = new CustomerWaterMarkerForLong(Time.milliseconds(maxOutOrderness), pos);
+            waterMarker = new CustomerWaterMarkerForLong(Time.milliseconds(maxOutOrderness), pos,timeZone);
         }else{
             throw new IllegalArgumentException("not support type of " + fieldType + ", current only support(timestamp, long).");
         }
diff --git a/kafka09/kafka09-source/src/main/java/com/dtstack/flink/sql/source/kafka/table/KafkaSourceParser.java b/kafka09/kafka09-source/src/main/java/com/dtstack/flink/sql/source/kafka/table/KafkaSourceParser.java
@@ -48,6 +48,7 @@ public TableInfo getTableInfo(String tableName, String fieldsInfo, Map<String, O
         kafka09SourceTableInfo.setTopic(MathUtil.getString(props.get(KafkaSourceTableInfo.TOPIC_KEY.toLowerCase())));
         kafka09SourceTableInfo.setOffsetReset(MathUtil.getString(props.get(KafkaSourceTableInfo.OFFSETRESET_KEY.toLowerCase())));
         kafka09SourceTableInfo.setTopicIsPattern(MathUtil.getBoolean(props.get(KafkaSourceTableInfo.TOPICISPATTERN_KEY.toLowerCase())));
+        kafka09SourceTableInfo.setTimeZone(MathUtil.getString(props.get(KafkaSourceTableInfo.TIME_ZONE_KEY.toLowerCase())));
         kafka09SourceTableInfo.check();
         return kafka09SourceTableInfo;
     }
diff --git a/kafka10/kafka10-source/src/main/java/com/dtstack/flink/sql/source/kafka/table/KafkaSourceParser.java b/kafka10/kafka10-source/src/main/java/com/dtstack/flink/sql/source/kafka/table/KafkaSourceParser.java
@@ -48,6 +48,7 @@ public TableInfo getTableInfo(String tableName, String fieldsInfo, Map<String, O
         kafka10SourceTableInfo.setTopic(MathUtil.getString(props.get(KafkaSourceTableInfo.TOPIC_KEY.toLowerCase())));
         kafka10SourceTableInfo.setOffsetReset(MathUtil.getString(props.get(KafkaSourceTableInfo.OFFSETRESET_KEY.toLowerCase())));
         kafka10SourceTableInfo.setTopicIsPattern(MathUtil.getBoolean(props.get(KafkaSourceTableInfo.TOPICISPATTERN_KEY.toLowerCase())));
+        kafka10SourceTableInfo.setTimeZone(MathUtil.getString(props.get(KafkaSourceTableInfo.TIME_ZONE_KEY.toLowerCase())));
         kafka10SourceTableInfo.check();
         return kafka10SourceTableInfo;
     }
diff --git a/kafka11/kafka11-source/src/main/java/com/dtstack/flink/sql/source/kafka/table/KafkaSourceParser.java b/kafka11/kafka11-source/src/main/java/com/dtstack/flink/sql/source/kafka/table/KafkaSourceParser.java
@@ -48,6 +48,7 @@ public TableInfo getTableInfo(String tableName, String fieldsInfo, Map<String, O
         kafka11SourceTableInfo.setTopic(MathUtil.getString(props.get(KafkaSourceTableInfo.TOPIC_KEY.toLowerCase())));
         kafka11SourceTableInfo.setOffsetReset(MathUtil.getString(props.get(KafkaSourceTableInfo.OFFSETRESET_KEY.toLowerCase())));
         kafka11SourceTableInfo.setTopicIsPattern(MathUtil.getBoolean(props.get(KafkaSourceTableInfo.TOPICISPATTERN_KEY.toLowerCase())));
+        kafka11SourceTableInfo.setTimeZone(MathUtil.getString(props.get(KafkaSourceTableInfo.TIME_ZONE_KEY.toLowerCase())));
         kafka11SourceTableInfo.check();
         return kafka11SourceTableInfo;
     }

Original file line number	Diff line number	Diff line change
`@@ -48,6 +48,7 @@ public TableInfo getTableInfo(String tableName, String fieldsInfo, Map<String, O`
`48`	`48`	`kafka09SourceTableInfo.setTopic(MathUtil.getString(props.get(KafkaSourceTableInfo.TOPIC_KEY.toLowerCase())));`
`49`	`49`	`kafka09SourceTableInfo.setOffsetReset(MathUtil.getString(props.get(KafkaSourceTableInfo.OFFSETRESET_KEY.toLowerCase())));`
`50`	`50`	`kafka09SourceTableInfo.setTopicIsPattern(MathUtil.getBoolean(props.get(KafkaSourceTableInfo.TOPICISPATTERN_KEY.toLowerCase())));`
	`51`	`+ kafka09SourceTableInfo.setTimeZone(MathUtil.getString(props.get(KafkaSourceTableInfo.TIME_ZONE_KEY.toLowerCase())));`
`51`	`52`	`kafka09SourceTableInfo.check();`
`52`	`53`	`return kafka09SourceTableInfo;`
`53`	`54`	`}`