handle json array type when get node

Author: todd5167

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

@@ -34,6 +34,7 @@
 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;
 import org.apache.flink.streaming.connectors.kafka.internal.KafkaConsumerThread;
 import org.apache.flink.streaming.connectors.kafka.internals.AbstractFetcher;
 import org.apache.flink.types.Row;
@@ -124,7 +125,7 @@ public Row deserialize(byte[] message)  {
             Row row = new Row(fieldNames.length);
 
             for (int i = 0; i < fieldNames.length; i++) {
-                JsonNode node = getIgnoreCase(root, fieldNames[i]);
+                JsonNode node = getIgnoreCase(fieldNames[i]);
 
                 if (node == null) {
                     if (failOnMissingField) {
@@ -168,9 +169,16 @@ public void setFailOnMissingField(boolean failOnMissingField) {
     }
 
 
-    public JsonNode getIgnoreCase(JsonNode jsonNode, String key) {
+    public JsonNode getIgnoreCase(String key) {
         String nodeMappingKey = rowAndFieldMapping.get(key);
-        return nodeAndJsonnodeMapping.get(nodeMappingKey);
+        JsonNode node = nodeAndJsonnodeMapping.get(nodeMappingKey);
+        JsonNodeType nodeType = node.getNodeType();
+
+        if (nodeType==JsonNodeType.ARRAY){
+            throw new IllegalStateException("Unsupported  type information  array .") ;
+        }
+
+        return node;
     }
 
     public void setFetcher(AbstractFetcher<Row, ?> fetcher) {
@@ -236,12 +244,7 @@ private Object convert(JsonNode node, TypeInformation<?> info) {
         } else if (info.getTypeClass().equals(Types.SQL_TIMESTAMP.getTypeClass())) {
              // local zone
             return Timestamp.valueOf(node.asText());
-        } else if (info instanceof ObjectArrayTypeInfo) {
-            throw new IllegalStateException("Unsupported type information '" + info + "' for node: " + node);
-        } else if (info instanceof BasicArrayTypeInfo) {
-            throw new IllegalStateException("Unsupported type information '" + info + "' for node: " + node);
-        } else if (info instanceof PrimitiveArrayTypeInfo &&
-                ((PrimitiveArrayTypeInfo) info).getComponentType() == Types.BYTE) {
+        }  else if (info.getTypeClass().equals(Types.BYTE.getTypeClass())){
             return convertByteArray(node);
         } else {
             // for types that were specified without JSON schema