add log

Author: yanxi0227

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

@@ -111,14 +111,14 @@ public Row deserialize(byte[] message) throws IOException {
         }
 
         try {
-            numInRecord.inc();
-            numInBytes.inc(message.length);
-
             JsonNode root = objectMapper.readTree(message);
-
-            if (numInResolveRecord.getCount()%rowLenth == 0){
+            if (numInRecord.getCount()%rowLenth == 0){
                 LOG.info(root.toString());
             }
+
+            numInRecord.inc();
+            numInBytes.inc(message.length);
+
             parseTree(root, null);
             Row row = new Row(fieldNames.length);
 
@@ -143,6 +143,9 @@ public Row deserialize(byte[] message) throws IOException {
             return row;
         } catch (Throwable t) {
             //add metric of dirty data
+            if (dirtyDataCounter.getCount()%rowLenth == 0){
+                LOG.info(objectMapper.readTree(message).toString());
+            }
             dirtyDataCounter.inc();
             return null;
         }finally {

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

@@ -110,14 +110,15 @@ public Row deserialize(byte[] message) throws IOException {
         }
 
         try {
-            numInRecord.inc();
-            numInBytes.inc(message.length);
-
             JsonNode root = objectMapper.readTree(message);
 
-            if (numInResolveRecord.getCount()%rowLenth == 0){
+            if (numInRecord.getCount()%rowLenth == 0){
                 LOG.info(root.toString());
             }
+
+            numInRecord.inc();
+            numInBytes.inc(message.length);
+
             parseTree(root, null);
             Row row = new Row(fieldNames.length);
 
@@ -142,6 +143,9 @@ public Row deserialize(byte[] message) throws IOException {
             return row;
         } catch (Throwable t) {
             //add metric of dirty data
+            if (dirtyDataCounter.getCount()%rowLenth == 0){
+                LOG.info(objectMapper.readTree(message).toString());
+            }
             dirtyDataCounter.inc();
             return null;
         }finally {

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

@@ -113,14 +113,15 @@ public Row deserialize(byte[] message) throws IOException {
         }
 
         try {
-            numInRecord.inc();
-            numInBytes.inc(message.length);
-
             JsonNode root = objectMapper.readTree(message);
 
-            if (numInResolveRecord.getCount()%rowLenth == 0){
+            if (numInRecord.getCount()%rowLenth == 0){
                 LOG.info(root.toString());
             }
+
+            numInRecord.inc();
+            numInBytes.inc(message.length);
+
             parseTree(root, null);
             Row row = new Row(fieldNames.length);
 
@@ -145,6 +146,9 @@ public Row deserialize(byte[] message) throws IOException {
             return row;
         } catch (Throwable t) {
             //add metric of dirty data
+            if (dirtyDataCounter.getCount()%rowLenth == 0){
+                LOG.info(objectMapper.readTree(message).toString());
+            }
             dirtyDataCounter.inc();
             return null;
         }finally {