Merge branch '1.8_release_3.10.x' into 1.8_release_4.0.x

gituser · gituser · commit 2b21311becb5 · 2020-07-22T09:56:30.000+08:00
diff --git a/cassandra/cassandra-side/cassandra-all-side/src/main/java/com/dtstack/flink/sql/side/cassandra/CassandraAllReqRow.java b/cassandra/cassandra-side/cassandra-all-side/src/main/java/com/dtstack/flink/sql/side/cassandra/CassandraAllReqRow.java
@@ -50,10 +50,7 @@
 import java.net.InetAddress;
 import java.sql.SQLException;
 import java.sql.Timestamp;
-import java.util.ArrayList;
-import java.util.Calendar;
-import java.util.List;
-import java.util.Map;
+import java.util.*;
 import java.util.concurrent.atomic.AtomicReference;
 
 /**
@@ -91,8 +88,10 @@ public Row fillData(Row input, Object sideInput) {
 
             //Type information for indicating event or processing time. However, it behaves like a regular SQL timestamp but is serialized as Long.
             if (obj instanceof Timestamp && isTimeIndicatorTypeInfo) {
-                obj = ((Timestamp) obj).getTime();
+                //去除上一层OutputRowtimeProcessFunction 调用时区导致的影响
+                obj = ((Timestamp) obj).getTime() + (long)LOCAL_TZ.getOffset(((Timestamp) obj).getTime());
             }
+
             row.setField(entry.getKey(), obj);
         }
 
diff --git a/cassandra/cassandra-side/cassandra-async-side/src/main/java/com/dtstack/flink/sql/side/cassandra/CassandraAsyncReqRow.java b/cassandra/cassandra-side/cassandra-async-side/src/main/java/com/dtstack/flink/sql/side/cassandra/CassandraAsyncReqRow.java
@@ -60,6 +60,7 @@
 import java.util.ArrayList;
 import java.util.List;
 import java.util.Map;
+import java.util.TimeZone;
 
 /**
  * Reason:
@@ -73,6 +74,8 @@ public class CassandraAsyncReqRow extends BaseAsyncReqRow {
 
     private static final Logger LOG = LoggerFactory.getLogger(CassandraAsyncReqRow.class);
 
+    private static final TimeZone LOCAL_TZ = TimeZone.getDefault();
+
     private final static int DEFAULT_VERTX_EVENT_LOOP_POOL_SIZE = 10;
 
     private final static int DEFAULT_VERTX_WORKER_POOL_SIZE = 20;
@@ -251,7 +254,8 @@ public Row fillData(Row input, Object line) {
             boolean isTimeIndicatorTypeInfo = TimeIndicatorTypeInfo.class.isAssignableFrom(sideInfo.getRowTypeInfo().getTypeAt(entry.getValue()).getClass());
 
             if (obj instanceof Timestamp && isTimeIndicatorTypeInfo) {
-                obj = ((Timestamp) obj).getTime();
+                //去除上一层OutputRowtimeProcessFunction 调用时区导致的影响
+                obj = ((Timestamp) obj).getTime() + (long)LOCAL_TZ.getOffset(((Timestamp) obj).getTime());
             }
 
             row.setField(entry.getKey(), obj);
diff --git a/core/src/main/java/com/dtstack/flink/sql/side/BaseAllReqRow.java b/core/src/main/java/com/dtstack/flink/sql/side/BaseAllReqRow.java
@@ -32,6 +32,7 @@
 import org.apache.calcite.sql.JoinType;
 
 import java.sql.SQLException;
+import java.util.TimeZone;
 import java.util.concurrent.ScheduledExecutorService;
 import java.util.concurrent.ScheduledThreadPoolExecutor;
 import java.util.concurrent.TimeUnit;
@@ -49,6 +50,8 @@ public abstract class BaseAllReqRow extends RichFlatMapFunction<CRow, CRow> impl
 
     public static final long LOAD_DATA_ERROR_SLEEP_TIME = 5_000L;
 
+    public static final TimeZone LOCAL_TZ = TimeZone.getDefault();
+
     protected BaseSideInfo sideInfo;
 
     private ScheduledExecutorService es;
diff --git a/elasticsearch6/elasticsearch6-side/elasticsearch6-all-side/src/main/java/com/dtstack/flink/sql/side/elasticsearch6/Elasticsearch6AllReqRow.java b/elasticsearch6/elasticsearch6-side/elasticsearch6-all-side/src/main/java/com/dtstack/flink/sql/side/elasticsearch6/Elasticsearch6AllReqRow.java
@@ -106,9 +106,11 @@ public Row fillData(Row input, Object sideInput) {
 
             //Type information for indicating event or processing time. However, it behaves like a regular SQL timestamp but is serialized as Long.
             if (obj instanceof Timestamp && isTimeIndicatorTypeInfo) {
-                obj = ((Timestamp) obj).getTime();
+                //去除上一层OutputRowtimeProcessFunction 调用时区导致的影响
+                obj = ((Timestamp) obj).getTime() + (long)LOCAL_TZ.getOffset(((Timestamp) obj).getTime());
             }
 
+
             row.setField(entry.getKey(), obj);
         }
 
diff --git a/elasticsearch6/elasticsearch6-side/elasticsearch6-async-side/src/main/java/com/dtstack/flink/sql/side/elasticsearch6/Elasticsearch6AsyncReqRow.java b/elasticsearch6/elasticsearch6-side/elasticsearch6-async-side/src/main/java/com/dtstack/flink/sql/side/elasticsearch6/Elasticsearch6AsyncReqRow.java
@@ -57,6 +57,7 @@
 import java.sql.Timestamp;
 import java.util.List;
 import java.util.Map;
+import java.util.TimeZone;
 
 /**
  * @author yinxi
@@ -65,6 +66,7 @@
 public class Elasticsearch6AsyncReqRow extends BaseAsyncReqRow implements Serializable {
 
     private static final Logger LOG = LoggerFactory.getLogger(Elasticsearch6AsyncReqRow.class);
+    private static final TimeZone LOCAL_TZ = TimeZone.getDefault();
     private transient RestHighLevelClient rhlClient;
     private SearchRequest searchRequest;
     private List<String> sqlJoinCompareOperate = Lists.newArrayList();
@@ -195,7 +197,8 @@ public Row fillData(Row input, Object line) {
             Object obj = input.getField(entry.getValue());
             boolean isTimeIndicatorTypeInfo = TimeIndicatorTypeInfo.class.isAssignableFrom(sideInfo.getRowTypeInfo().getTypeAt(entry.getValue()).getClass());
             if (obj instanceof Timestamp && isTimeIndicatorTypeInfo) {
-                obj = ((Timestamp) obj).getTime();
+                //去除上一层OutputRowtimeProcessFunction 调用时区导致的影响
+                obj = ((Timestamp) obj).getTime() + (long)LOCAL_TZ.getOffset(((Timestamp) obj).getTime());
             }
 
             row.setField(entry.getKey(), obj);
diff --git a/hbase/hbase-side/hbase-all-side/src/main/java/com/dtstack/flink/sql/side/hbase/HbaseAllReqRow.java b/hbase/hbase-side/hbase-all-side/src/main/java/com/dtstack/flink/sql/side/hbase/HbaseAllReqRow.java
@@ -98,8 +98,10 @@ public Row fillData(Row input, Object sideInput) {
 
             //Type information for indicating event or processing time. However, it behaves like a regular SQL timestamp but is serialized as Long.
             if(obj instanceof Timestamp && isTimeIndicatorTypeInfo){
-                obj = ((Timestamp)obj).getTime();
+                //去除上一层OutputRowtimeProcessFunction 调用时区导致的影响
+                obj = ((Timestamp) obj).getTime() + (long)LOCAL_TZ.getOffset(((Timestamp) obj).getTime());
             }
+
             row.setField(entry.getKey(), obj);
         }
 
diff --git a/hbase/hbase-side/hbase-async-side/src/main/java/com/dtstack/flink/sql/side/hbase/HbaseAsyncReqRow.java b/hbase/hbase-side/hbase-async-side/src/main/java/com/dtstack/flink/sql/side/hbase/HbaseAsyncReqRow.java
@@ -53,6 +53,7 @@
 import java.util.Collections;
 import java.util.List;
 import java.util.Map;
+import java.util.TimeZone;
 import java.util.concurrent.ExecutorService;
 import java.util.concurrent.LinkedBlockingQueue;
 import java.util.concurrent.ThreadPoolExecutor;
@@ -68,6 +69,8 @@ public class HbaseAsyncReqRow extends BaseAsyncReqRow {
 
     private static final long serialVersionUID = 2098635104857937717L;
 
+    private static final TimeZone LOCAL_TZ = TimeZone.getDefault();
+
     private static final Logger LOG = LoggerFactory.getLogger(HbaseAsyncReqRow.class);
 
     //match to the rule of netty3
@@ -164,7 +167,8 @@ public Row fillData(Row input, Object sideInput){
             boolean isTimeIndicatorTypeInfo = TimeIndicatorTypeInfo.class.isAssignableFrom(sideInfo.getRowTypeInfo().getTypeAt(entry.getValue()).getClass());
 
             if(obj instanceof Timestamp && isTimeIndicatorTypeInfo){
-                obj = ((Timestamp)obj).getTime();
+                //去除上一层OutputRowtimeProcessFunction 调用时区导致的影响
+                obj = ((Timestamp) obj).getTime() + (long)LOCAL_TZ.getOffset(((Timestamp) obj).getTime());
             }
 
             row.setField(entry.getKey(), obj);
diff --git a/kudu/kudu-side/kudu-all-side/src/main/java/com/dtstack/flink/sql/side/kudu/KuduAllReqRow.java b/kudu/kudu-side/kudu-all-side/src/main/java/com/dtstack/flink/sql/side/kudu/KuduAllReqRow.java
@@ -76,8 +76,10 @@ public Row fillData(Row input, Object sideInput) {
 
             //Type information for indicating event or processing time. However, it behaves like a regular SQL timestamp but is serialized as Long.
             if (obj instanceof Timestamp && isTimeIndicatorTypeInfo) {
-                obj = ((Timestamp) obj).getTime();
+                //去除上一层OutputRowtimeProcessFunction 调用时区导致的影响
+                obj = ((Timestamp) obj).getTime() + (long)LOCAL_TZ.getOffset(((Timestamp) obj).getTime());
             }
+
             row.setField(entry.getKey(), obj);
         }
 
diff --git a/kudu/kudu-side/kudu-async-side/src/main/java/com/dtstack/flink/sql/side/kudu/KuduAsyncReqRow.java b/kudu/kudu-side/kudu-async-side/src/main/java/com/dtstack/flink/sql/side/kudu/KuduAsyncReqRow.java
@@ -36,14 +36,14 @@
 import org.slf4j.LoggerFactory;
 
 import java.sql.Timestamp;
-import java.util.Arrays;
-import java.util.Collections;
-import java.util.List;
-import java.util.Map;
+import java.util.*;
 
 public class KuduAsyncReqRow extends BaseAsyncReqRow {
 
     private static final Logger LOG = LoggerFactory.getLogger(KuduAsyncReqRow.class);
+
+    private static final TimeZone LOCAL_TZ = TimeZone.getDefault();
+
     /**
      * 获取连接的尝试次数
      */
@@ -184,6 +184,8 @@ public Row fillData(Row input, Object sideInput) {
             //Type information for indicating event or processing time. However, it behaves like a regular SQL timestamp but is serialized as Long.
             if (obj instanceof Timestamp && isTimeIndicatorTypeInfo) {
                 obj = ((Timestamp) obj).getTime();
+                //去除上一层OutputRowtimeProcessFunction 调用时区导致的影响
+                obj = ((Timestamp) obj).getTime() + (long)LOCAL_TZ.getOffset(((Timestamp) obj).getTime());
             }
             row.setField(entry.getKey(), obj);
         }
diff --git a/mongo/mongo-side/mongo-all-side/src/main/java/com/dtstack/flink/sql/side/mongo/MongoAllReqRow.java b/mongo/mongo-side/mongo-all-side/src/main/java/com/dtstack/flink/sql/side/mongo/MongoAllReqRow.java
@@ -88,8 +88,10 @@ public Row fillData(Row input, Object sideInput) {
 
             //Type information for indicating event or processing time. However, it behaves like a regular SQL timestamp but is serialized as Long.
             if (obj instanceof Timestamp && isTimeIndicatorTypeInfo) {
-                obj = ((Timestamp) obj).getTime();
+                //去除上一层OutputRowtimeProcessFunction 调用时区导致的影响
+                obj = ((Timestamp) obj).getTime() + (long)LOCAL_TZ.getOffset(((Timestamp) obj).getTime());
             }
+
             row.setField(entry.getKey(), obj);
         }
 
diff --git a/mongo/mongo-side/mongo-async-side/src/main/java/com/dtstack/flink/sql/side/mongo/MongoAsyncReqRow.java b/mongo/mongo-side/mongo-async-side/src/main/java/com/dtstack/flink/sql/side/mongo/MongoAsyncReqRow.java
@@ -50,10 +50,7 @@
 import org.slf4j.LoggerFactory;
 
 import java.sql.Timestamp;
-import java.util.Collection;
-import java.util.Collections;
-import java.util.List;
-import java.util.Map;
+import java.util.*;
 import java.util.concurrent.atomic.AtomicInteger;
 
 /**
@@ -63,8 +60,11 @@
  * @author xuqianjin
  */
 public class MongoAsyncReqRow extends BaseAsyncReqRow {
+
     private static final long serialVersionUID = -1183158242862673706L;
 
+    private static final TimeZone LOCAL_TZ = TimeZone.getDefault();
+
     private static final Logger LOG = LoggerFactory.getLogger(MongoAsyncReqRow.class);
 
     private transient MongoClient mongoClient;
@@ -165,6 +165,8 @@ public Row fillData(Row input, Object line) {
 
             if (obj instanceof Timestamp && isTimeIndicatorTypeInfo) {
                 obj = ((Timestamp) obj).getTime();
+                //去除上一层OutputRowtimeProcessFunction 调用时区导致的影响
+                obj = ((Timestamp) obj).getTime() + (long)LOCAL_TZ.getOffset(((Timestamp) obj).getTime());
             }
 
             row.setField(entry.getKey(), obj);
diff --git a/rdb/rdb-side/src/main/java/com/dtstack/flink/sql/side/rdb/all/AbstractRdbAllReqRow.java b/rdb/rdb-side/src/main/java/com/dtstack/flink/sql/side/rdb/all/AbstractRdbAllReqRow.java
@@ -166,7 +166,8 @@ public Row fillData(Row input, Object sideInput) {
     protected Object dealTimeAttributeType(Class<? extends TypeInformation> entry, Object obj) {
         boolean isTimeIndicatorTypeInfo = TimeIndicatorTypeInfo.class.isAssignableFrom(entry);
         if (obj instanceof Timestamp && isTimeIndicatorTypeInfo) {
-            obj = ((Timestamp) obj).getTime();
+            //去除上一层OutputRowtimeProcessFunction 调用时区导致的影响
+            obj = ((Timestamp) obj).getTime() + (long)LOCAL_TZ.getOffset(((Timestamp) obj).getTime());
         }
         return obj;
     }
diff --git a/rdb/rdb-side/src/main/java/com/dtstack/flink/sql/side/rdb/async/RdbAsyncReqRow.java b/rdb/rdb-side/src/main/java/com/dtstack/flink/sql/side/rdb/async/RdbAsyncReqRow.java
@@ -48,6 +48,7 @@
 import java.time.Instant;
 import java.util.List;
 import java.util.Map;
+import java.util.TimeZone;
 import java.util.concurrent.*;
 import java.util.concurrent.atomic.AtomicBoolean;
 import java.util.concurrent.atomic.AtomicLong;
@@ -63,6 +64,8 @@ public class RdbAsyncReqRow extends BaseAsyncReqRow {
 
     private static final long serialVersionUID = 2098635244857937720L;
 
+    private static final TimeZone LOCAL_TZ = TimeZone.getDefault();
+
     private static final Logger LOG = LoggerFactory.getLogger(RdbAsyncReqRow.class);
 
     public final static int DEFAULT_VERTX_EVENT_LOOP_POOL_SIZE = 1;
@@ -227,7 +230,8 @@ public Row fillData(Row input, Object line) {
             Object obj = input.getField(entry.getValue());
             boolean isTimeIndicatorTypeInfo = TimeIndicatorTypeInfo.class.isAssignableFrom(sideInfo.getRowTypeInfo().getTypeAt(entry.getValue()).getClass());
             if (obj instanceof Timestamp && isTimeIndicatorTypeInfo) {
-                obj = ((Timestamp) obj).getTime();
+                //去除上一层OutputRowtimeProcessFunction 调用时区导致的影响
+                obj = ((Timestamp) obj).getTime() + (long)LOCAL_TZ.getOffset(((Timestamp) obj).getTime());
             }
 
             row.setField(entry.getKey(), obj);
diff --git a/redis5/redis5-side/redis-side-core/src/main/java/com/dtstack/flink/sql/side/redis/table/RedisSideReqRow.java b/redis5/redis5-side/redis-side-core/src/main/java/com/dtstack/flink/sql/side/redis/table/RedisSideReqRow.java
@@ -28,6 +28,7 @@
 import java.sql.Date;
 import java.sql.Timestamp;
 import java.util.Map;
+import java.util.TimeZone;
 
 /**
  * redis fill row data
@@ -40,6 +41,8 @@ public class RedisSideReqRow implements ISideReqRow, Serializable {
 
     private static final long serialVersionUID = 3751171828444748982L;
 
+    private static final TimeZone LOCAL_TZ = TimeZone.getDefault();
+
     private BaseSideInfo sideInfo;
 
     public RedisSideReqRow(BaseSideInfo sideInfo){
@@ -55,7 +58,8 @@ public Row fillData(Row input, Object sideInput) {
             boolean isTimeIndicatorTypeInfo = TimeIndicatorTypeInfo.class.isAssignableFrom(sideInfo.getRowTypeInfo().getTypeAt(entry.getValue()).getClass());
 
             if(obj instanceof Timestamp && isTimeIndicatorTypeInfo){
-                obj = ((Timestamp)obj).getTime();
+                //去除上一层OutputRowtimeProcessFunction 调用时区导致的影响
+                obj = ((Timestamp) obj).getTime() + (long)LOCAL_TZ.getOffset(((Timestamp) obj).getTime());
             }
             row.setField(entry.getKey(), obj);
         }

Original file line number	Diff line number	Diff line change
`@@ -106,9 +106,11 @@ public Row fillData(Row input, Object sideInput) {`
`106`	`106`
`107`	`107`	`//Type information for indicating event or processing time. However, it behaves like a regular SQL timestamp but is serialized as Long.`
`108`	`108`	`if (obj instanceof Timestamp && isTimeIndicatorTypeInfo) {`
`109`		`- obj = ((Timestamp) obj).getTime();`
	`109`	`+ //去除上一层OutputRowtimeProcessFunction 调用时区导致的影响`
	`110`	`+ obj = ((Timestamp) obj).getTime() + (long)LOCAL_TZ.getOffset(((Timestamp) obj).getTime());`
`110`	`111`	`}`
`111`	`112`
	`113`	`+`
`112`	`114`	`row.setField(entry.getKey(), obj);`
`113`	`115`	`}`
`114`	`116`
Original file line number	Diff line number	Diff line change
`@@ -98,8 +98,10 @@ public Row fillData(Row input, Object sideInput) {`
`98`	`98`
`99`	`99`	`//Type information for indicating event or processing time. However, it behaves like a regular SQL timestamp but is serialized as Long.`
`100`	`100`	`if(obj instanceof Timestamp && isTimeIndicatorTypeInfo){`
`101`		`- obj = ((Timestamp)obj).getTime();`
	`101`	`+ //去除上一层OutputRowtimeProcessFunction 调用时区导致的影响`
	`102`	`+ obj = ((Timestamp) obj).getTime() + (long)LOCAL_TZ.getOffset(((Timestamp) obj).getTime());`
`102`	`103`	`}`
	`104`	`+`
`103`	`105`	`row.setField(entry.getKey(), obj);`
`104`	`106`	`}`
`105`	`107`
Original file line number	Diff line number	Diff line change
`@@ -76,8 +76,10 @@ public Row fillData(Row input, Object sideInput) {`
`76`	`76`
`77`	`77`	`//Type information for indicating event or processing time. However, it behaves like a regular SQL timestamp but is serialized as Long.`
`78`	`78`	`if (obj instanceof Timestamp && isTimeIndicatorTypeInfo) {`
`79`		`- obj = ((Timestamp) obj).getTime();`
	`79`	`+ //去除上一层OutputRowtimeProcessFunction 调用时区导致的影响`
	`80`	`+ obj = ((Timestamp) obj).getTime() + (long)LOCAL_TZ.getOffset(((Timestamp) obj).getTime());`
`80`	`81`	`}`
	`82`	`+`
`81`	`83`	`row.setField(entry.getKey(), obj);`
`82`	`84`	`}`
`83`	`85`
Original file line number	Diff line number	Diff line change
`@@ -88,8 +88,10 @@ public Row fillData(Row input, Object sideInput) {`
`88`	`88`
`89`	`89`	`//Type information for indicating event or processing time. However, it behaves like a regular SQL timestamp but is serialized as Long.`
`90`	`90`	`if (obj instanceof Timestamp && isTimeIndicatorTypeInfo) {`
`91`		`- obj = ((Timestamp) obj).getTime();`
	`91`	`+ //去除上一层OutputRowtimeProcessFunction 调用时区导致的影响`
	`92`	`+ obj = ((Timestamp) obj).getTime() + (long)LOCAL_TZ.getOffset(((Timestamp) obj).getTime());`
`92`	`93`	`}`
	`94`	`+`
`93`	`95`	`row.setField(entry.getKey(), obj);`
`94`	`96`	`}`
`95`	`97`