修改异步维表情况下。如果字段选择了ROWTIME, 框架本身在输出的时候会对rowtime进行时区处理，导致window无法触发

xuchao · xuchao · commit 7fafc922a26c · 2020-06-08T19:27:55.000+08:00
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/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-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
@@ -48,6 +48,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;
@@ -63,6 +64,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
@@ -143,7 +146,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-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-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/async/RdbAsyncReqRow.java b/rdb/rdb-side/src/main/java/com/dtstack/flink/sql/side/rdb/async/RdbAsyncReqRow.java
@@ -47,6 +47,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;
@@ -62,6 +63,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;
@@ -211,7 +214,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);
         }
