[hotfix-34789][elasticsearch5&6][sink]support timestamp datatype.

Author: HiLany

core/src/main/java/com/dtstack/flink/sql/enums/ColumnType.java

@@ -102,6 +102,10 @@ public enum ColumnType {
      * timestamp
      */
     TIMESTAMP,
+    /**
+     * time eg: 23:59:59
+     */
+    TIME,
     /**
      * decimal
      */

core/src/main/java/com/dtstack/flink/sql/util/DtStringUtil.java

@@ -243,6 +243,9 @@ public static String col2string(Object column, String type) {
             case DATE:
                 result = DateUtil.dateToString((java.util.Date)column);
                 break;
+            case TIME:
+                result = DateUtil.getTimeFromStr(String.valueOf(column));
+                break;
             case TIMESTAMP:
                 result = DateUtil.timestampToString((java.util.Date)column);
                 break;

elasticsearch5/elasticsearch5-sink/src/main/java/com/dtstack/flink/sql/sink/elasticsearch/CustomerSinkFunc.java

@@ -20,6 +20,7 @@
 
 package com.dtstack.flink.sql.sink.elasticsearch;
 
+import com.dtstack.flink.sql.util.DateUtil;
 import org.apache.commons.lang3.StringUtils;
 import org.apache.flink.api.common.functions.RuntimeContext;
 import org.apache.flink.api.java.tuple.Tuple2;
@@ -32,6 +33,8 @@
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
+import java.sql.Date;
+import java.sql.Timestamp;
 import java.util.List;
 import java.util.Map;
 import java.util.stream.Collectors;
@@ -107,6 +110,14 @@ private IndexRequest createIndexRequest(Row element) {
         Map<String, Object> dataMap = EsUtil.rowToJsonMap(element,fieldNames,fieldTypes);
         int length = Math.min(element.getArity(), fieldNames.size());
         for(int i=0; i<length; i++){
+            if (element.getField(i) instanceof Date) {
+                dataMap.put(fieldNames.get(i), DateUtil.transformSqlDateToUtilDate((Date) element.getField(i)));
+                continue;
+            }
+            if (element.getField(i) instanceof Timestamp) {
+                dataMap.put(fieldNames.get(i), ((Timestamp) element.getField(i)).getTime());
+                continue;
+            }
             dataMap.put(fieldNames.get(i), element.getField(i));
         }
 

elasticsearch6/elasticsearch6-sink/src/main/java/com/dtstack/flink/sql/sink/elasticsearch/CustomerSinkFunc.java

@@ -33,6 +33,8 @@
 import org.slf4j.LoggerFactory;
 
 import java.sql.Date;
+import java.sql.Time;
+import java.sql.Timestamp;
 import java.util.List;
 import java.util.Map;
 import java.util.stream.Collectors;
@@ -115,6 +117,10 @@ private IndexRequest createIndexRequest(Row element) {
                 dataMap.put(fieldNames.get(i), DateUtil.transformSqlDateToUtilDate((Date) element.getField(i)));
                 continue;
             }
+            if (element.getField(i) instanceof Timestamp) {
+                dataMap.put(fieldNames.get(i), ((Timestamp) element.getField(i)).getTime());
+                continue;
+            }
             dataMap.put(fieldNames.get(i), element.getField(i));
         }
 

elasticsearch6/elasticsearch6-sink/src/main/java/com/dtstack/flink/sql/sink/elasticsearch/Es6Util.java

@@ -18,13 +18,11 @@
 
 package com.dtstack.flink.sql.sink.elasticsearch;
 
+import com.dtstack.flink.sql.util.DtStringUtil;
+import org.apache.flink.shaded.guava18.com.google.common.collect.Maps;
 import org.apache.flink.types.Row;
 import org.apache.flink.util.Preconditions;
 
-import org.apache.flink.shaded.guava18.com.google.common.collect.Maps;
-
-import com.dtstack.flink.sql.util.DtStringUtil;
-
 import java.util.HashMap;
 import java.util.List;
 import java.util.Map;