format print log

Author: todd5167

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

@@ -53,6 +53,10 @@ public class RetractJDBCOutputFormat extends MetricOutputFormat {
 
     private static final Logger LOG = LoggerFactory.getLogger(RetractJDBCOutputFormat.class);
 
+    private static int dirtyDataPrintFrequency = 1000;
+
+    private static int receiveDataPrintFrequency = 1000;
+
     private String username;
     private String password;
     private String drivername;
@@ -167,8 +171,11 @@ public void writeRecord(Tuple2 tuple2)  {
         }
 
         if (retract) {
-            insertWrite(row);
             outRecords.inc();
+            if (outRecords.getCount() % receiveDataPrintFrequency == 0) {
+                LOG.info("Receive data : {}", row);
+            }
+            insertWrite(row);
         } else {
             //do nothing
         }
@@ -200,8 +207,10 @@ private void writeSingleRecord(Row row) {
             upload.execute();
         } catch (SQLException e) {
             outDirtyRecords.inc();
-            LOG.error("record insert failed ..", row.toString());
-            LOG.error("", e);
+            if (outDirtyRecords.getCount() % dirtyDataPrintFrequency == 0) {
+                LOG.error("record insert failed ..", row.toString());
+                LOG.error("", e);
+            }
         }
     }
 
@@ -297,7 +306,6 @@ private void updatePreparedStmt(Row row, PreparedStatement pstmt) throws SQLExce
 
     private synchronized void submitExecuteBatch() {
         try {
-            LOG.info("submitExecuteBatch start......");
             this.upload.executeBatch();
             dbConn.commit();
         } catch (SQLException e) {