[opt-36803][file] Adjust dirty data processing logic.

Author: FlechazoW

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

@@ -18,6 +18,7 @@
 
 package com.dtstack.flink.sql.source.file;
 
+import com.dtstack.flink.sql.source.file.throwable.LengthMismatchException;
 import org.apache.flink.annotation.PublicEvolving;
 import org.apache.flink.api.common.serialization.DeserializationSchema;
 import org.apache.flink.api.common.typeinfo.BasicArrayTypeInfo;
@@ -371,8 +372,9 @@ private static RuntimeConverter createByteArrayRuntimeConverter() {
 
     private static void validateArity(int expected, int actual) {
         if (expected != actual) {
-            LOG.warn("Row length mismatch. " + expected +
-                " fields expected but was " + actual + ".");
+            throw new LengthMismatchException(
+                    "Row length mismatch. " + expected +
+                            " fields expected but was " + actual + ".");
         }
     }
 }

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

@@ -18,6 +18,7 @@
 
 package com.dtstack.flink.sql.source.file;
 
+import com.dtstack.flink.sql.exception.ExceptionTrace;
 import com.dtstack.flink.sql.metric.MetricConstant;
 import com.dtstack.flink.sql.source.IStreamSourceGener;
 import com.dtstack.flink.sql.source.file.table.FileSourceTableInfo;
@@ -224,7 +225,7 @@ public void run(SourceContext<Row> ctx) throws Exception {
                 } catch (IOException e) {
                     if (errorCounter.getCount() % 1000 == 0) {
                         LOG.error("Deserialize error! Record: " + line);
-                        LOG.error("Cause: ", e);
+                        LOG.error("Cause: " + ExceptionTrace.traceOriginalCause(e));
                     }
                     errorCounter.inc();
                 }

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

@@ -0,0 +1,32 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package com.dtstack.flink.sql.source.file.throwable;
+
+import org.apache.flink.util.FlinkRuntimeException;
+
+/**
+ * @author tiezhu
+ * @since 2021/5/6 3:39 下午
+ */
+
+public class LengthMismatchException extends FlinkRuntimeException {
+    public LengthMismatchException(String message) {
+        super(message);
+    }
+}