Merge branch '1.10_release_4.0.x' into 1.10_release_4.1.x

Author: FlechazoW

core/src/main/java/com/dtstack/flink/sql/side/BaseAllReqRow.java

@@ -37,6 +37,7 @@
 import java.time.LocalDateTime;
 import java.util.Map;
 import java.util.TimeZone;
+import java.util.TimeZone;
 import java.util.concurrent.ScheduledExecutorService;
 import java.util.concurrent.ScheduledThreadPoolExecutor;
 import java.util.concurrent.TimeUnit;
@@ -45,6 +46,7 @@
  * Reason:
  * Date: 2018/9/18
  * Company: www.dtstack.com
+ *
  * @author xuchao
  */
 
@@ -86,7 +88,8 @@ protected Object convertTimeIndictorTypeInfo(Integer index, Object obj) {
 
         //Type information for indicating event or processing time. However, it behaves like a regular SQL timestamp but is serialized as Long.
         if (obj instanceof LocalDateTime && isTimeIndicatorTypeInfo) {
-            obj = Timestamp.valueOf(((LocalDateTime) obj));
+            //去除上一层OutputRowtimeProcessFunction 调用时区导致的影响
+            obj = ((Timestamp) obj).getTime() + (long)LOCAL_TZ.getOffset(((Timestamp) obj).getTime());
         }
         return obj;
     }

core/src/main/java/com/dtstack/flink/sql/side/BaseAsyncReqRow.java

@@ -52,6 +52,7 @@
 import java.util.Collections;
 import java.util.List;
 import java.util.Map;
+import java.util.TimeZone;
 import java.util.concurrent.ScheduledFuture;
 
 /**
@@ -71,6 +72,7 @@ public abstract class BaseAsyncReqRow extends RichAsyncFunction<Row, BaseRow> im
     private int timeOutNum = 0;
     protected BaseSideInfo sideInfo;
     protected transient Counter parseErrorRecords;
+    private static final TimeZone LOCAL_TZ = TimeZone.getDefault();
 
     public BaseAsyncReqRow(BaseSideInfo sideInfo) {
         this.sideInfo = sideInfo;
@@ -117,7 +119,8 @@ protected Object convertTimeIndictorTypeInfo(Integer index, Object obj) {
 
         //Type information for indicating event or processing time. However, it behaves like a regular SQL timestamp but is serialized as Long.
         if (obj instanceof LocalDateTime && isTimeIndicatorTypeInfo) {
-            obj = Timestamp.valueOf(((LocalDateTime) obj));
+            //去除上一层OutputRowtimeProcessFunction 调用时区导致的影响
+            obj = ((Timestamp) obj).getTime() + (long)LOCAL_TZ.getOffset(((Timestamp) obj).getTime());
         }
         return obj;
     }

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

@@ -16,14 +16,18 @@
  * limitations under the License.
  */
 
- 
+
 
 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:
@@ -36,6 +40,10 @@ 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;
@@ -101,4 +109,23 @@ 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,6 +29,8 @@
 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
@@ -51,6 +53,8 @@ public abstract class AbstractCustomerWaterMarker<T> extends BoundedOutOfOrderne
 
     protected long lastTime = 0;
 
+    protected TimeZone timezone;
+
     public AbstractCustomerWaterMarker(Time maxOutOfOrderness) {
         super(maxOutOfOrderness);
     }
@@ -98,7 +102,8 @@ public void setFromSourceTag(String fromSourceTag) {
 
     protected long getExtractTimestamp(Long extractTime){
 
-        lastTime = extractTime;
+        lastTime = extractTime + timezone.getOffset(extractTime);
+
         eventDelayGauge.setDelayTime(MathUtil.getIntegerVal((System.currentTimeMillis() - extractTime)/1000));
 
         return lastTime;

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

@@ -26,6 +26,8 @@
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
+import java.util.TimeZone;
+
 /**
  * Custom watermark --- for eventtime
  * Date: 2017/12/28
@@ -37,9 +39,10 @@ public class CustomerWaterMarkerForLong extends AbstractCustomerWaterMarker<Row>
 
     private static final Logger logger = LoggerFactory.getLogger(CustomerWaterMarkerForLong.class);
 
-    public CustomerWaterMarkerForLong(Time maxOutOfOrderness, int pos) {
+    public CustomerWaterMarkerForLong(Time maxOutOfOrderness, int pos, String timezone) {
         super(maxOutOfOrderness);
         this.pos = pos;
+        this.timezone= TimeZone.getTimeZone(timezone);
     }
 
     @Override

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

@@ -16,7 +16,7 @@
  * limitations under the License.
  */
 
- 
+
 
 package com.dtstack.flink.sql.watermarker;
 
@@ -39,9 +39,10 @@ public class CustomerWaterMarkerForTimeStamp extends AbstractCustomerWaterMarker
 
     private static final Logger logger = LoggerFactory.getLogger(CustomerWaterMarkerForTimeStamp.class);
 
-    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

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

@@ -48,6 +48,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();
 
@@ -69,9 +71,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);
+            waterMarker = new CustomerWaterMarkerForTimeStamp(Time.milliseconds(maxOutOrderness), pos, timeZone);
         }else if(fieldType.getTypeClass().isAssignableFrom(Long.class)){
-            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).");
         }

elasticsearch6/elasticsearch6-side/elasticsearch6-all-side/src/main/java/com/dtstack/flink/sql/side/elasticsearch6/Elasticsearch6AllReqRow.java

@@ -30,6 +30,7 @@
 import org.apache.commons.lang3.StringUtils;
 import org.apache.flink.api.java.typeutils.RowTypeInfo;
 import org.apache.flink.table.dataformat.BaseRow;
+import org.apache.flink.table.typeutils.TimeIndicatorTypeInfo;
 import org.apache.flink.types.Row;
 import org.apache.flink.util.Collector;
 import org.elasticsearch.action.search.SearchRequest;
@@ -45,6 +46,7 @@
 
 import java.io.IOException;
 import java.io.Serializable;
+import java.sql.Timestamp;
 import java.util.Calendar;
 import java.util.List;
 import java.util.Map;
@@ -93,6 +95,34 @@ public void flatMap(Row value, Collector<BaseRow> out) throws Exception {
         }
     }
 
+    @Override
+    public Row fillData(Row input, Object sideInput) {
+        Map<String, Object> cacheInfo = (Map<String, Object>) sideInput;
+        Row row = new Row(sideInfo.getOutFieldInfoList().size());
+        for (Map.Entry<Integer, Integer> entry : sideInfo.getInFieldIndex().entrySet()) {
+            Object obj = input.getField(entry.getValue());
+            boolean isTimeIndicatorTypeInfo = TimeIndicatorTypeInfo.class.isAssignableFrom(sideInfo.getRowTypeInfo().getTypeAt(entry.getValue()).getClass());
+
+            //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) {
+                //去除上一层OutputRowtimeProcessFunction 调用时区导致的影响
+                obj = ((Timestamp) obj).getTime() + (long)LOCAL_TZ.getOffset(((Timestamp) obj).getTime());
+            }
+
+            row.setField(entry.getKey(), obj);
+        }
+
+        for (Map.Entry<Integer, String> entry : sideInfo.getSideFieldNameIndex().entrySet()) {
+            if (cacheInfo == null) {
+                row.setField(entry.getKey(), null);
+            } else {
+                row.setField(entry.getKey(), cacheInfo.get(entry.getValue()));
+            }
+        }
+
+        return row;
+    }
+
     private String buildKey(List<Object> equalValList) {
         StringBuilder sb = new StringBuilder("");
         for (Object equalVal : equalValList) {

hbase/hbase-side/hbase-all-side/src/main/java/com/dtstack/flink/sql/side/hbase/HbaseAllReqRow.java

@@ -99,7 +99,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 LocalDateTime && isTimeIndicatorTypeInfo) {
-                obj = Timestamp.valueOf(((LocalDateTime) obj));
+                //去除上一层OutputRowtimeProcessFunction 调用时区导致的影响
+                obj = ((Timestamp) obj).getTime() + (long)LOCAL_TZ.getOffset(((Timestamp) obj).getTime());
             }
 
             row.setField(entry.getKey(), obj);

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

@@ -56,6 +56,7 @@ public AbstractTableInfo getTableInfo(String tableName, String fieldsInfo, Map<S
         kafkaSourceTableInfo.setSchemaString(MathUtil.getString(props.get(KafkaSourceTableInfo.SCHEMA_STRING_KEY.toLowerCase())));
         kafkaSourceTableInfo.setFieldDelimiter(MathUtil.getString(props.getOrDefault(KafkaSourceTableInfo.CSV_FIELD_DELIMITER_KEY.toLowerCase(), "|")));
         kafkaSourceTableInfo.setSourceDataType(MathUtil.getString(props.getOrDefault(KafkaSourceTableInfo.SOURCE_DATA_TYPE_KEY.toLowerCase(), FormatType.DT_NEST.name())));
+        kafkaSourceTableInfo.setTimeZone(MathUtil.getString(props.get(KafkaSourceTableInfo.TIME_ZONE_KEY.toLowerCase())));
 
         if(props.containsKey(KafkaSourceTableInfo.TIMESTAMP_OFFSET.toLowerCase())){
             kafkaSourceTableInfo.setTimestampOffset(MathUtil.getLongVal(props.getOrDefault(KafkaSourceTableInfo.TIMESTAMP_OFFSET.toLowerCase(), System.currentTimeMillis())));