[Feature][Connector-V2][File] Support skip corrupt files

Author: TyrantLucifer

Diff for: seatunnel-connectors-v2/connector-file/connector-file-base/src/main/java/org/apache/seatunnel/connectors/seatunnel/file/config/BaseSourceConfig.java

@@ -126,4 +126,11 @@ public class BaseSourceConfig {
                     .enumType(CompressFormat.class)
                     .defaultValue(CompressFormat.NONE)
                     .withDescription("Compression codec");
+
+    public static final Option<Boolean> IGNORE_CORRUPT_FILES =
+            Options.key("ignore_corrupt_files")
+                    .booleanType()
+                    .defaultValue(true)
+                    .withDescription(
+                            "Whether corrupted [orc/parquet] files are ignored during synchronization");
 }

Diff for: seatunnel-connectors-v2/connector-file/connector-file-base/src/main/java/org/apache/seatunnel/connectors/seatunnel/file/source/reader/AbstractReadStrategy.java

@@ -75,6 +75,7 @@ public abstract class AbstractReadStrategy implements ReadStrategy {
     protected List<String> readColumns = new ArrayList<>();
     protected boolean isMergePartition = true;
     protected long skipHeaderNumber = BaseSourceConfig.SKIP_HEADER_ROW_NUMBER.defaultValue();
+    protected boolean ignoreCorruptFiles = BaseSourceConfig.IGNORE_CORRUPT_FILES.defaultValue();
     protected transient boolean isKerberosAuthorization = false;
 
     protected Pattern pattern;
@@ -177,6 +178,10 @@ public void setPluginConfig(Config pluginConfig) {
                     pluginConfig.getString(BaseSourceConfig.FILE_FILTER_PATTERN.key());
             this.pattern = Pattern.compile(Matcher.quoteReplacement(filterPattern));
         }
+        if (pluginConfig.hasPath(BaseSourceConfig.IGNORE_CORRUPT_FILES.key())) {
+            this.ignoreCorruptFiles =
+                    pluginConfig.getBoolean(BaseSourceConfig.IGNORE_CORRUPT_FILES.key());
+        }
     }
 
     @Override

Diff for: seatunnel-connectors-v2/connector-file/connector-file-base/src/main/java/org/apache/seatunnel/connectors/seatunnel/file/source/reader/OrcReadStrategy.java

@@ -77,6 +77,10 @@ public class OrcReadStrategy extends AbstractReadStrategy {
     public void read(String path, Collector<SeaTunnelRow> output)
             throws FileConnectorException, IOException {
         if (Boolean.FALSE.equals(checkFileType(path))) {
+            if (ignoreCorruptFiles) {
+                log.warn("This file [{}] is corrupted, it will be ignored", path);
+                return;
+            }
             String errorMsg =
                     String.format(
                             "This file [%s] is not a orc file, please check the format of this file",

Diff for: seatunnel-connectors-v2/connector-file/connector-file-base/src/main/java/org/apache/seatunnel/connectors/seatunnel/file/source/reader/ParquetReadStrategy.java

@@ -84,6 +84,10 @@ public class ParquetReadStrategy extends AbstractReadStrategy {
     public void read(String path, Collector<SeaTunnelRow> output)
             throws FileConnectorException, IOException {
         if (Boolean.FALSE.equals(checkFileType(path))) {
+            if (ignoreCorruptFiles) {
+                log.warn("This file [{}] is corrupted, it will be ignored", path);
+                return;
+            }
             String errorMsg =
                     String.format(
                             "This file [%s] is not a parquet file, please check the format of this file",