Merge branch '1.8_merge_1.5' into 'v1.8.0_dev'

1.8 merge 1.5



See merge request !59

Author: zoudaokoulife

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));
      }
 

kafka09/kafka09-sink/src/main/java/com/dtstack/flink/sql/sink/kafka/KafkaSink.java

@@ -109,7 +109,6 @@ public void emitDataStream(DataStream<Tuple2<Boolean, Row>> dataStream) {
 				serializationSchema
 		);
 
-
 		DataStream<Row> ds = dataStream.map((Tuple2<Boolean, Row> record) -> {
 			return record.f1;
 		}).returns(getOutputType().getTypeAt(1));

kafka10/kafka10-sink/src/main/java/com/dtstack/flink/sql/sink/kafka/KafkaSink.java

@@ -112,7 +112,6 @@ public void emitDataStream(DataStream<Tuple2<Boolean, Row>> dataStream) {
                 serializationSchema
         );
 
-
         DataStream<Row> ds = dataStream.map((Tuple2<Boolean, Row> record) -> {
             return record.f1;
         }).returns(getOutputType().getTypeAt(1));

kafka11/kafka11-sink/src/main/java/com/dtstack/flink/sql/sink/kafka/KafkaSink.java

@@ -92,7 +92,6 @@ public KafkaSink genStreamSink(TargetTableInfo targetTableInfo) {
             schemaBuilder.field(fieldNames[i], fieldTypes[i]);
         }
         this.schema = schemaBuilder.build();
-
         this.serializationSchema = new CustomerJsonRowSerializationSchema(getOutputType().getTypeAt(1));
         return this;
     }
@@ -111,8 +110,6 @@ public void emitDataStream(DataStream<Tuple2<Boolean, Row>> dataStream) {
                 partitioner,
                 serializationSchema
         );
-
-
         DataStream<Row> ds = dataStream.map((Tuple2<Boolean, Row> record) -> {
             return record.f1;
         }).returns(getOutputType().getTypeAt(1));
@@ -122,7 +119,7 @@ public void emitDataStream(DataStream<Tuple2<Boolean, Row>> dataStream) {
 
     @Override
     public TupleTypeInfo<Tuple2<Boolean, Row>> getOutputType() {
-       return new TupleTypeInfo(org.apache.flink.table.api.Types.BOOLEAN(), new RowTypeInfo(fieldTypes, fieldNames));
+        return new TupleTypeInfo(org.apache.flink.table.api.Types.BOOLEAN(), new RowTypeInfo(fieldTypes, fieldNames));
     }
 
     @Override
@@ -136,7 +133,7 @@ public TypeInformation<?>[] getFieldTypes() {
     }
 
     @Override
-    public TableSink<Tuple2<Boolean, Row>> configure(String[] fieldNames, TypeInformation<?>[] fieldTypes) {
+    public TableSink<Tuple2<Boolean, Row>>configure(String[] fieldNames, TypeInformation<?>[] fieldTypes) {
         this.fieldNames = fieldNames;
         this.fieldTypes = fieldTypes;
         return this;

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,53 @@ 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();
+            dbConn.commit();
+        } catch (SQLException e) {
+            outDirtyRecords.inc();
+            LOG.error("record insert failed ..", row.toString());
+            LOG.error("", e);
         }
     }
 
@@ -276,9 +300,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 +348,7 @@ public void close() throws IOException {
             LOG.info("Inputformat couldn't be closed - ", se);
         } finally {
             upload = null;
-            batchCount.set(0);
+            rows.clear();
         }
 
         try {