[Fix][Connector-File] Fix conflicting file_format_type requirement (#8823)

Co-authored-by: litiliu <[email protected]>
Co-authored-by: hailin0 <[email protected]>

Author: 3 people

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

@@ -46,7 +46,7 @@ public class BaseFileSinkConfig implements DelimiterConfig, Serializable {
     protected boolean singleFileMode = BaseSinkConfig.SINGLE_FILE_MODE.defaultValue();
     protected boolean createEmptyFileWhenNoData =
             BaseSinkConfig.CREATE_EMPTY_FILE_WHEN_NO_DATA.defaultValue();
-    protected FileFormat fileFormat = FileFormat.TEXT;
+    protected FileFormat fileFormat;
     protected String filenameExtension = BaseSinkConfig.FILENAME_EXTENSION.defaultValue();
     protected DateUtils.Formatter dateFormat = DateUtils.Formatter.YYYY_MM_DD;
     protected DateTimeUtils.Formatter datetimeFormat = DateTimeUtils.Formatter.YYYY_MM_DD_HH_MM_SS;
@@ -101,6 +101,9 @@ public BaseFileSinkConfig(@NonNull Config config) {
                     FileFormat.valueOf(
                             config.getString(BaseSinkConfig.FILE_FORMAT_TYPE.key())
                                     .toUpperCase(Locale.ROOT));
+        } else {
+            // fall back to the default
+            this.fileFormat = BaseSinkConfig.FILE_FORMAT_TYPE.defaultValue();
         }
 
         if (config.hasPath(BaseSinkConfig.FILENAME_EXTENSION.key())

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

@@ -209,9 +209,7 @@ public FileSinkConfig(@NonNull Config config, @NonNull SeaTunnelRowType seaTunne
             this.sheetName = config.getString(BaseSinkConfig.SHEET_NAME.key());
         }
 
-        if (FileFormat.XML
-                .name()
-                .equalsIgnoreCase(config.getString(BaseSinkConfig.FILE_FORMAT_TYPE.key()))) {
+        if (FileFormat.XML.equals(this.fileFormat)) {
             if (!config.hasPath(BaseSinkConfig.XML_USE_ATTR_FORMAT.key())) {
                 throw new FileConnectorException(
                         CommonErrorCodeDeprecated.ILLEGAL_ARGUMENT,
@@ -229,9 +227,7 @@ public FileSinkConfig(@NonNull Config config, @NonNull SeaTunnelRowType seaTunne
             }
         }
 
-        if (FileFormat.PARQUET
-                .name()
-                .equalsIgnoreCase(config.getString(BaseSinkConfig.FILE_FORMAT_TYPE.key()))) {
+        if (FileFormat.PARQUET.equals(this.fileFormat)) {
             if (config.hasPath(BaseSinkConfig.PARQUET_AVRO_WRITE_TIMESTAMP_AS_INT96.key())) {
                 this.parquetWriteTimestampAsInt96 =
                         config.getBoolean(
@@ -244,9 +240,7 @@ public FileSinkConfig(@NonNull Config config, @NonNull SeaTunnelRowType seaTunne
             }
         }
 
-        if (FileFormat.CSV
-                .name()
-                .equalsIgnoreCase(config.getString(BaseSinkConfig.FILE_FORMAT_TYPE.key()))) {
+        if (FileFormat.CSV.equals(this.fileFormat)) {
             if (config.hasPath(BaseSinkConfig.CSV_STRING_QUOTE_MODE.key())) {
                 this.csvStringQuoteMode =
                         CsvStringQuoteMode.valueOf(

Diff for: seatunnel-connectors-v2/connector-file/connector-file-base/src/test/java/org/apache/seatunnel/connectors/seatunnel/file/writer/FileSinkConfigTest.java

@@ -49,6 +49,20 @@ public void testConfigInit() throws Exception {
         Assertions.assertDoesNotThrow(() -> new FileSinkConfig(config, rowType));
     }
 
+    @Test
+    public void testConfigInitDefault() throws Exception {
+        URL conf = OrcReadStrategyTest.class.getResource("/test_write_hdfs_default_format.conf");
+        Assertions.assertNotNull(conf);
+        String confPath = Paths.get(conf.toURI()).toString();
+        Config config = ConfigFactory.parseFile(new File(confPath));
+
+        SeaTunnelRowType rowType =
+                new SeaTunnelRowType(
+                        new String[] {"data", "ts"},
+                        new SeaTunnelDataType[] {BasicType.STRING_TYPE, BasicType.STRING_TYPE});
+        Assertions.assertDoesNotThrow(() -> new FileSinkConfig(config, rowType));
+    }
+
     @Test
     public void testSinkColumnsGreaterThanSource() throws Exception {
         URL conf = OrcReadStrategyTest.class.getResource("/test_write_hive.conf");

Diff for: seatunnel-connectors-v2/connector-file/connector-file-base/src/test/resources/test_write_hdfs_default_format.conf

@@ -0,0 +1,27 @@
+#
+# 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.
+#
+
+{
+      fs.defaultFS = "hdfs://hadoop01:9000"
+      have_partition = true
+      partition_by = ["ts"]
+      partition_dir_expression = "${v0}"
+      is_partition_field_write_in_file = false
+      path = "/data/test"
+      batch_size=10
+}
+