Merge branch 'feat_1.10.4.2.x_fileDirtyData' into 1.10_test_4.2.x

Author: gituser

file/file-source/src/main/java/com/dtstack/flink/sql/source/file/FileSource.java

@@ -46,6 +46,7 @@
 import java.net.URI;
 import java.util.Locale;
 import java.util.concurrent.atomic.AtomicBoolean;
+import java.util.concurrent.atomic.AtomicInteger;
 
 /**
  * @author tiezhu
@@ -71,6 +72,10 @@ public class FileSource implements IStreamSourceGener<Table>, SourceFunction<Row
 
     private String charset;
 
+    private final AtomicInteger count = new AtomicInteger(0);
+
+    private final AtomicInteger errorCount = new AtomicInteger(0);
+
     @Override
     public Table genStreamSource(AbstractSourceTableInfo sourceTableInfo,
                                  StreamExecutionEnvironment env,
@@ -168,10 +173,19 @@ public void run(SourceContext<Row> ctx) throws Exception {
                 inputStream.close();
                 break;
             } else {
-                Row row = deserializationSchema.deserialize(line.getBytes());
-                ctx.collect(row);
+                try {
+                    Row row = deserializationSchema.deserialize(line.getBytes());
+                    if (row == null) {
+                        throw new IOException("Deserialized row is null");
+                    }
+                    ctx.collect(row);
+                    count.incrementAndGet();
+                } catch (IOException e) {
+                    errorCount.incrementAndGet();
+                }
             }
         }
+        printCount();
     }
 
     @Override
@@ -185,5 +199,11 @@ public void cancel() {
                 LOG.error("File input stream close error!");
             }
         }
+        printCount();
+    }
+
+    private void printCount() {
+        LOG.info("Read count: {}", count.get());
+        LOG.info("Error count: {}", errorCount.get());
     }
 }