modify dirtyDataFrequency name

Author: todd5167

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

@@ -67,7 +67,7 @@ public class CustomerJsonDeserialization extends AbsDeserialization<Row> {
 
     private static final long serialVersionUID = 2385115520960444192L;
 
-    private static int rowLenth = 1000;
+    private static int dirtyDataFrequency = 1000;
 
     private final ObjectMapper objectMapper = new ObjectMapper();
 
@@ -117,7 +117,7 @@ public Row deserialize(byte[] message) throws IOException {
 
         try {
             JsonNode root = objectMapper.readTree(message);
-            if (numInRecord.getCount()%rowLenth == 0){
+            if (numInRecord.getCount() % dirtyDataFrequency == 0) {
                 LOG.info(root.toString());
             }
 
@@ -148,7 +148,7 @@ public Row deserialize(byte[] message) throws IOException {
             return row;
         } catch (Exception e) {
             //add metric of dirty data
-            if (dirtyDataCounter.getCount()%rowLenth == 0){
+            if (dirtyDataCounter.getCount() % dirtyDataFrequency == 0) {
                 LOG.info("dirtyData: " + new String(message));
                 LOG.info(" " ,e);
             }

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

@@ -67,7 +67,7 @@ public class CustomerJsonDeserialization extends AbsDeserialization<Row> {
 
     private static final long serialVersionUID = 2385115520960444192L;
 
-    private static int rowLenth = 1000;
+    private static int dirtyDataFrequency = 1000;
 
     private final ObjectMapper objectMapper = new ObjectMapper();
 
@@ -117,7 +117,7 @@ public Row deserialize(byte[] message) throws IOException {
         try {
             JsonNode root = objectMapper.readTree(message);
 
-            if (numInRecord.getCount()%rowLenth == 0){
+            if (numInRecord.getCount() % dirtyDataFrequency == 0) {
                 LOG.info(root.toString());
             }
 
@@ -148,7 +148,7 @@ public Row deserialize(byte[] message) throws IOException {
             return row;
         } catch (Exception e) {
             //add metric of dirty data
-            if (dirtyDataCounter.getCount()%rowLenth == 0){
+            if (dirtyDataCounter.getCount() % dirtyDataFrequency == 0) {
                 LOG.info("dirtyData: " + new String(message));
                 LOG.error(" ", e);
             }

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

@@ -70,7 +70,7 @@ public class CustomerJsonDeserialization extends AbsDeserialization<Row> {
 
     private static final long serialVersionUID = 2385115520960444192L;
 
-    private static int rowLenth = 1000;
+    private static int dirtyDataFrequency = 1000;
 
     private final ObjectMapper objectMapper = new ObjectMapper();
 
@@ -121,7 +121,7 @@ public Row deserialize(byte[] message) throws IOException {
         try {
             JsonNode root = objectMapper.readTree(message);
 
-            if (numInRecord.getCount()%rowLenth == 0){
+            if (numInRecord.getCount() % dirtyDataFrequency == 0) {
                 LOG.info(root.toString());
             }
 
@@ -153,7 +153,7 @@ public Row deserialize(byte[] message) throws IOException {
             return row;
         } catch (Exception e) {
             //add metric of dirty data
-            if (dirtyDataCounter.getCount()%rowLenth == 0){
+            if (dirtyDataCounter.getCount() % dirtyDataFrequency == 0) {
                 LOG.info("dirtyData: " + new String(message));
                 LOG.error("" , e);
             }