优化 output format 的第一条数据输入和第一条脏数据输出

Author: zoudaokoulife

cassandra/cassandra-sink/src/main/java/com/dtstack/flink/sql/sink/cassandra/CassandraOutputFormat.java

@@ -216,7 +216,7 @@ private void insertWrite(Row row) {
             }
         } catch (Exception e) {
             if(outDirtyRecords.getCount() % DIRTY_PRINT_FREQUENCY == 0){
-                LOG.error("record insert failed ..", row.toString().substring(0, 100));
+                LOG.error("record insert failed, total dirty num:{}, current record:{}", outDirtyRecords.getCount(), row.toString());
                 LOG.error("", e);
             }
 

kudu/kudu-sink/src/main/java/com/dtstack/flink/sql/sink/kudu/KuduOutputFormat.java

@@ -108,7 +108,7 @@ public void writeRecord(Tuple2 record) throws IOException {
 
         if (row.getArity() != fieldNames.length) {
             if(outDirtyRecords.getCount() % DIRTY_PRINT_FREQUENCY == 0) {
-                LOG.error("record insert failed ..", row.toString());
+                LOG.error("record insert failed:{}", row.toString());
                 LOG.error("cause by row.getArity() != fieldNames.length");
             }
 
@@ -131,7 +131,7 @@ public void writeRecord(Tuple2 record) throws IOException {
         } catch (KuduException e) {
 
             if(outDirtyRecords.getCount() % DIRTY_PRINT_FREQUENCY == 0){
-                LOG.error("record insert failed ..", row.toString().substring(0, 100));
+                LOG.error("record insert failed, total dirty record:{} current row:{}", outDirtyRecords.getCount(), row.toString());
                 LOG.error("", e);
             }
 

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

@@ -194,7 +194,15 @@ private void insertWrite(Row row) {
                 }
             }
         } catch (SQLException e) {
-            LOG.error("", e);
+            if (outDirtyRecords.getCount() % DIRTY_PRINT_FREQUENCY == 0 || LOG.isDebugEnabled()) {
+                outDirtyRecords.inc(batchNum == 1 ? batchNum : rows.size());
+                LOG.error("record insert failed,dirty record num:{}, current row:{}", outDirtyRecords.getCount(), row.toString());
+                LOG.error("", e);
+            } else {
+                outDirtyRecords.inc(batchNum == 1 ? batchNum : rows.size());
+            }
+
+
         }
 
     }