Merge remote-tracking branch 'origin/v1.5.0_dev' into v1.8.0_dev

# Conflicts:
#	core/src/main/java/com/dtstack/flink/sql/side/SideSqlExec.java

Author: todd5167

core/src/main/java/com/dtstack/flink/sql/side/SideSqlExec.java

@@ -17,6 +17,7 @@
  */
 
 
+
 package com.dtstack.flink.sql.side;
 
 import com.dtstack.flink.sql.enums.ECacheType;

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

@@ -148,7 +148,7 @@ public Row deserialize(byte[] message) throws IOException {
             return row;
         } catch (Exception e) {
             //add metric of dirty data
-            if (dirtyDataCounter.getCount() % dirtyDataFrequency == 0) {
+            if (dirtyDataCounter.getCount() % dirtyDataFrequency == 0 || LOG.isDebugEnabled()) {
                 LOG.info("dirtyData: " + new String(message));
                 LOG.error(" ", e);
             }