Merge remote-tracking branch 'origin/v1.5.0_dev' into 1.8_merge_1.5

# Conflicts:
#	kafka09/kafka09-sink/src/main/java/com/dtstack/flink/sql/sink/kafka/KafkaSink.java
#	kafka10/kafka10-sink/src/main/java/com/dtstack/flink/sql/sink/kafka/KafkaSink.java
#	kafka11/kafka11-sink/src/main/java/com/dtstack/flink/sql/sink/kafka/KafkaSink.java

Author: todd5167

core/src/main/java/com/dtstack/flink/sql/metric/MetricConstant.java

@@ -45,6 +45,8 @@ public class MetricConstant {
 
     public static final String DT_NUM_RECORDS_OUT = "dtNumRecordsOut";
 
+    public static final String DT_NUM_DIRTY_RECORDS_OUT = "dtNumDirtyRecordsOut";
+
     public static final String DT_NUM_RECORDS_OUT_RATE = "dtNumRecordsOutRate";
 
     public static final String DT_EVENT_DELAY_GAUGE = "dtEventDelay";

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

@@ -32,10 +32,13 @@ public abstract  class MetricOutputFormat extends  RichOutputFormat<Tuple2>{
 
      protected  transient Counter outRecords;
 
+     protected  transient Counter outDirtyRecords;
+
      protected transient Meter outRecordsRate;
 
      public void initMetric() {
         outRecords = getRuntimeContext().getMetricGroup().counter(MetricConstant.DT_NUM_RECORDS_OUT);
+        outDirtyRecords = getRuntimeContext().getMetricGroup().counter(MetricConstant.DT_NUM_DIRTY_RECORDS_OUT);
         outRecordsRate = getRuntimeContext().getMetricGroup().meter(MetricConstant.DT_NUM_RECORDS_OUT_RATE, new MeterView(outRecords, 20));
      }
 

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

@@ -32,6 +32,7 @@
 import java.sql.DriverManager;
 import java.sql.PreparedStatement;
 import java.sql.SQLException;
+import java.util.ArrayList;
 import java.util.Arrays;
 import java.util.List;
 import java.util.Map;
@@ -66,9 +67,11 @@ public class RetractJDBCOutputFormat extends MetricOutputFormat {
     private String insertQuery;
     public int[] typesArray;
 
+    /** 存储用于批量写入的数据 */
+    protected List<Row> rows = new ArrayList();
+
     private Connection dbConn;
     private PreparedStatement upload;
-    private AtomicInteger batchCount = new AtomicInteger(0);
     private transient ScheduledThreadPoolExecutor timerService;
 
 
@@ -95,7 +98,7 @@ public void configure(Configuration parameters) {
     public void open(int taskNumber, int numTasks) throws IOException {
         try {
             LOG.info("PreparedStatement execute batch num is {}", batchNum);
-            establishConnection();
+            dbConn = establishConnection();
             initMetric();
 
             if (dbConn.getMetaData().getTables(null, null, tableName, null).next()) {
@@ -107,7 +110,7 @@ public void open(int taskNumber, int numTasks) throws IOException {
                 throw new SQLException("Table " + tableName + " doesn't exist");
             }
 
-            if (batchWaitInterval > 0) {
+            if (batchWaitInterval > 0 && batchNum > 1) {
                 LOG.info("open batch wait interval scheduled, interval is {} ms", batchWaitInterval);
 
                 timerService = new ScheduledThreadPoolExecutor(1);
@@ -127,13 +130,16 @@ public void open(int taskNumber, int numTasks) throws IOException {
     }
 
 
-    private void establishConnection() throws SQLException, ClassNotFoundException {
+    private Connection establishConnection() throws SQLException, ClassNotFoundException {
+        Connection connection ;
         Class.forName(drivername);
         if (username == null) {
-            dbConn = DriverManager.getConnection(dbURL);
+            connection = DriverManager.getConnection(dbURL);
         } else {
-            dbConn = DriverManager.getConnection(dbURL, username, password);
+            connection = DriverManager.getConnection(dbURL, username, password);
         }
+        connection.setAutoCommit(false);
+        return connection;
     }
 
     /**
@@ -150,35 +156,52 @@ private void establishConnection() throws SQLException, ClassNotFoundException {
      * @see PreparedStatement
      */
     @Override
-    public void writeRecord(Tuple2 tuple2) throws IOException {
+    public void writeRecord(Tuple2 tuple2)  {
 
         Tuple2<Boolean, Row> tupleTrans = tuple2;
         Boolean retract = tupleTrans.getField(0);
         Row row = tupleTrans.getField(1);
 
-
         if (typesArray != null && typesArray.length > 0 && typesArray.length != row.getArity()) {
             LOG.warn("Column SQL types array doesn't match arity of passed Row! Check the passed array...");
         }
+
+        if (retract) {
+            insertWrite(row);
+            outRecords.inc();
+        } else {
+            //do nothing
+        }
+    }
+
+
+    private void insertWrite(Row row) {
+        checkConnectionOpen(dbConn);
         try {
-            if (retract) {
-                insertWrite(row);
-                outRecords.inc();
+            if (batchNum == 1) {
+                writeSingleRecord(row);
             } else {
-                //do nothing
+                updatePreparedStmt(row, upload);
+                rows.add(row);
+                upload.addBatch();
+                if (rows.size() >= batchNum) {
+                    submitExecuteBatch();
+                }
             }
-        } catch (SQLException | IllegalArgumentException e) {
-            throw new IllegalArgumentException("writeRecord() failed", e);
+        } catch (SQLException e) {
+            LOG.error("", e);
         }
-    }
 
+    }
 
-    private void insertWrite(Row row) throws SQLException {
-        updatePreparedStmt(row, upload);
-        upload.addBatch();
-        batchCount.incrementAndGet();
-        if (batchCount.get() >= batchNum) {
-            submitExecuteBatch();
+    private void writeSingleRecord(Row row) {
+        try {
+            updatePreparedStmt(row, upload);
+            upload.execute();
+        } catch (SQLException e) {
+            outDirtyRecords.inc();
+            LOG.error("record insert failed ..", row.toString());
+            LOG.error("", e);
         }
     }
 
@@ -276,9 +299,31 @@ private synchronized void submitExecuteBatch() {
         try {
             LOG.info("submitExecuteBatch start......");
             this.upload.executeBatch();
-            this.batchCount.set(0);
+            dbConn.commit();
         } catch (SQLException e) {
-            LOG.error("", e);
+            try {
+                dbConn.rollback();
+            } catch (SQLException e1) {
+                LOG.error("rollback  data error !", e);
+            }
+
+            rows.forEach(this::writeSingleRecord);
+        } finally {
+            rows.clear();
+        }
+    }
+
+    private void checkConnectionOpen(Connection dbConn) {
+        try {
+            if (dbConn.isClosed()) {
+                LOG.info("db connection reconnect..");
+                dbConn= establishConnection();
+                upload = dbConn.prepareStatement(insertQuery);
+            }
+        } catch (SQLException e) {
+            LOG.error("check connection open failed..", e);
+        } catch (ClassNotFoundException e) {
+            LOG.error("load jdbc class error when reconnect db..", e);
         }
     }
 
@@ -302,7 +347,7 @@ public void close() throws IOException {
             LOG.info("Inputformat couldn't be closed - ", se);
         } finally {
             upload = null;
-            batchCount.set(0);
+            rows.clear();
         }
 
         try {