Merge remote-tracking branch 'origin/1.5_v3.6.1' into 1.5_merge_3.6.1

# Conflicts:
#	core/src/main/java/com/dtstack/flink/sql/side/SideSQLParser.java
#	core/src/main/java/com/dtstack/flink/sql/side/SideSqlExec.java
#	rdb/rdb-sink/src/main/java/com/dtstack/flink/sql/sink/rdb/format/RetractJDBCOutputFormat.java

Author: todd5167

core/src/main/java/com/dtstack/flink/sql/sink/MetricOutputFormat.java

@@ -36,6 +36,8 @@ public abstract  class MetricOutputFormat extends  RichOutputFormat<Tuple2>{
 
      protected transient Meter outRecordsRate;
 
+    protected  transient Counter outDirtyRecords;
+
      public void initMetric() {
         outRecords = getRuntimeContext().getMetricGroup().counter(MetricConstant.DT_NUM_RECORDS_OUT);
         outDirtyRecords = getRuntimeContext().getMetricGroup().counter(MetricConstant.DT_NUM_DIRTY_RECORDS_OUT);

core/src/main/java/com/dtstack/flink/sql/util/JDBCUtils.java

@@ -0,0 +1,34 @@
+package com.dtstack.flink.sql.util;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.sql.DriverManager;
+
+public class JDBCUtils {
+
+    private static final Logger LOG = LoggerFactory.getLogger(ClassUtil.class);
+
+    public final static String lock_str = "jdbc_lock_str";
+
+    public static void forName(String clazz, ClassLoader classLoader)  {
+        synchronized (lock_str){
+            try {
+                Class.forName(clazz, true, classLoader);
+                DriverManager.setLoginTimeout(10);
+            } catch (Exception e) {
+                throw new RuntimeException(e);
+            }
+        }
+    }
+
+
+    public synchronized static void forName(String clazz) {
+        try {
+            Class<?> driverClass = Class.forName(clazz);
+            driverClass.newInstance();
+        } catch (Exception e) {
+            throw new RuntimeException(e);
+        }
+    }
+}