ignore miss field

Author: yanxi0227

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

@@ -150,6 +150,10 @@ public void setFailOnMissingField(boolean failOnMissingField) {
     private JsonNode getIgnoreCase(String key) {
         String nodeMappingKey = rowAndFieldMapping.getOrDefault(key, key);
         JsonNode node = nodeAndJsonNodeMapping.get(nodeMappingKey);
+        if(node == null){
+            return null;
+        }
+
         JsonNodeType nodeType = node.getNodeType();
 
         if (nodeType == JsonNodeType.ARRAY){

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

@@ -145,6 +145,10 @@ public Row deserialize(byte[] message) throws IOException {
     public JsonNode getIgnoreCase(String key) {
         String nodeMappingKey = rowAndFieldMapping.getOrDefault(key, key);
         JsonNode node = nodeAndJsonNodeMapping.get(nodeMappingKey);
+        if(node == null){
+            return null;
+        }
+
         JsonNodeType nodeType = node.getNodeType();
 
         if (nodeType==JsonNodeType.ARRAY){

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

@@ -148,6 +148,10 @@ public Row deserialize(byte[] message) throws IOException {
     public JsonNode getIgnoreCase(String key) {
         String nodeMappingKey = rowAndFieldMapping.getOrDefault(key, key);
         JsonNode node = nodeAndJsonNodeMapping.get(nodeMappingKey);
+        if(node == null){
+            return null;
+        }
+
         JsonNodeType nodeType = node.getNodeType();
 
         if (nodeType==JsonNodeType.ARRAY){