Merge branch 'v1.5.0_release'

ysq

Author: [email protected]

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

@@ -98,7 +98,7 @@ public Row deserialize(byte[] message) throws IOException {
 
         try {
             numInRecord.inc();
-            numInBytes.inc(message.length);
+            if(message!=null){numInBytes.inc(message.length);}
 
             JsonNode root = objectMapper.readTree(message);
             Row row = new Row(fieldNames.length);

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

@@ -98,7 +98,7 @@ public Row deserialize(byte[] message) throws IOException {
 
         try {
             numInRecord.inc();
-            numInBytes.inc(message.length);
+            if(message!=null){numInBytes.inc(message.length);}
 
             JsonNode root = objectMapper.readTree(message);
             Row row = new Row(fieldNames.length);

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

@@ -100,7 +100,7 @@ public Row deserialize(byte[] message) throws IOException {
 
         try {
             numInRecord.inc();
-            numInBytes.inc(message.length);
+            if(message!=null){numInBytes.inc(message.length);}
 
             JsonNode root = objectMapper.readTree(message);
             Row row = new Row(fieldNames.length);