convert json timestamp

Author: todd5167

kafka11/kafka11-source/src/main/java/com/dtstack/flink/sql/source/kafka/CustomerJsonDeserialization.java

@@ -24,10 +24,12 @@
 import com.dtstack.flink.sql.source.AbsDeserialization;
 import com.dtstack.flink.sql.source.kafka.metric.KafkaTopicPartitionLagMetric;
 import org.apache.flink.api.common.typeinfo.TypeInformation;
+import org.apache.flink.api.common.typeinfo.Types;
 import org.apache.flink.api.java.typeutils.RowTypeInfo;
 import org.apache.flink.calcite.shaded.com.google.common.base.Strings;
 import org.apache.flink.metrics.MetricGroup;
 import org.apache.flink.shaded.guava18.com.google.common.collect.Maps;
+import org.apache.flink.shaded.jackson2.com.fasterxml.jackson.core.JsonProcessingException;
 import org.apache.flink.shaded.jackson2.com.fasterxml.jackson.databind.JsonNode;
 import org.apache.flink.shaded.jackson2.com.fasterxml.jackson.databind.ObjectMapper;
 import org.apache.flink.shaded.jackson2.com.fasterxml.jackson.databind.node.JsonNodeType;
@@ -42,7 +44,11 @@
 import org.slf4j.LoggerFactory;
 
 import java.io.IOException;
+import java.lang.reflect.Array;
 import java.lang.reflect.Field;
+import java.sql.Date;
+import java.sql.Time;
+import java.sql.Timestamp;
 import java.util.Iterator;
 import java.util.Map;
 import java.util.Set;
@@ -137,7 +143,8 @@ public Row deserialize(byte[] message) throws IOException {
                     }
                 } else {
                     // Read the value as specified type
-                    Object value = objectMapper.treeToValue(node, fieldTypes[i].getTypeClass());
+
+                    Object value = convert(node, fieldTypes[i]);
                     row.setField(i, value);
                 }
             }
@@ -245,4 +252,29 @@ protected void registerPtMetric(AbstractFetcher<Row, ?> fetcher) throws Exceptio
     private static String partitionLagMetricName(TopicPartition tp) {
         return tp + ".records-lag";
     }
+
+    private Object convert(JsonNode node, TypeInformation<?> info) {
+        if (info.getTypeClass().equals(Types.BOOLEAN.getTypeClass())) {
+            return node.asBoolean();
+        } else if (info.getTypeClass().equals(Types.STRING.getTypeClass())) {
+            return node.asText();
+        }  else if (info.getTypeClass().equals(Types.SQL_DATE.getTypeClass())) {
+            return Date.valueOf(node.asText());
+        } else if (info.getTypeClass().equals(Types.SQL_TIME.getTypeClass())) {
+            // local zone
+            return Time.valueOf(node.asText());
+        } else if (info.getTypeClass().equals(Types.SQL_TIMESTAMP.getTypeClass())) {
+            // local zone
+            return Timestamp.valueOf(node.asText());
+        }  else {
+            // for types that were specified without JSON schema
+            // e.g. POJOs
+            try {
+                return objectMapper.treeToValue(node, info.getTypeClass());
+            } catch (JsonProcessingException e) {
+                throw new IllegalStateException("Unsupported type information '" + info + "' for node: " + node);
+            }
+        }
+    }
+
 }