Merge branch '1.5_v3.7.1' into 1.5_v3.7.2

Author: todd5167

rdb/rdb-sink/src/main/java/com/dtstack/flink/sql/sink/rdb/format/RetractJDBCOutputFormat.java

@@ -63,7 +63,7 @@ public class RetractJDBCOutputFormat extends MetricOutputFormat {
     private long batchWaitInterval = 10000l;
 
     // batchNum
-    private int batchInterval = 5000;
+    private int batchNum = 100;
     private String insertQuery;
     public int[] typesArray;
 
@@ -98,7 +98,7 @@ public void configure(Configuration parameters) {
     @Override
     public void open(int taskNumber, int numTasks) throws IOException {
         try {
-            LOG.info("PreparedStatement execute batch num is {}", batchInterval);
+            LOG.info("PreparedStatement execute batch num is {}", batchNum);
             dbConn = establishConnection();
             initMetric();
             if (dbConn.getMetaData().getTables(null, null, tableName, null).next()) {
@@ -110,7 +110,7 @@ public void open(int taskNumber, int numTasks) throws IOException {
                 throw new SQLException("Table " + tableName + " doesn't exist");
             }
 
-            if (batchWaitInterval > 0 && batchInterval > 1) {
+            if (batchWaitInterval > 0 && batchNum > 1) {
                 LOG.info("open batch wait interval scheduled, interval is {} ms", batchWaitInterval);
 
                 timerService = new ScheduledThreadPoolExecutor(1);
@@ -182,13 +182,13 @@ public void writeRecord(Tuple2 tuple2) {
     private void insertWrite(Row row) {
         checkConnectionOpen(dbConn);
         try {
-            if (batchInterval == 1) {
+            if (batchNum == 1) {
                 writeSingleRecord(row);
             } else {
                 updatePreparedStmt(row, upload);
                 rows.add(row);
                 upload.addBatch();
-                if (rows.size() >= batchInterval) {
+                if (rows.size() >= batchNum) {
                     submitExecuteBatch();
                 }
             }
@@ -413,9 +413,6 @@ public void setDbSink(RdbSink dbSink) {
         this.dbSink = dbSink;
     }
 
-    public void setBatchInterval(int batchInterval) {
-        this.batchInterval = batchInterval;
-    }
 
     public void setInsertQuery(String insertQuery) {
         this.insertQuery = insertQuery;
@@ -445,6 +442,15 @@ public Map<String, List<String>> getRealIndexes() {
         return realIndexes;
     }
 
+
+    public void setBatchNum(int batchNum) {
+        this.batchNum = batchNum;
+    }
+
+    public void setBatchWaitInterval(long batchWaitInterval) {
+        this.batchWaitInterval = batchWaitInterval;
+    }
+
     public List<String> getFullField() {
         return fullField;
     }