[Fix][Connector-V2] Fixed the problem of complex path reading of ftp

docs/en/connector-v2/source/FtpFile.md

@@ -263,6 +263,15 @@ The target ftp connection mode , default is active mode, supported as the follow
 
 `active_local` `passive_local`
 
+### control_encoding [string]
+
+Character encoding for FTP control connection. Default is `UTF-8`.
+
+When file paths contain special characters (such as `$`, spaces, Chinese characters, etc.),
+this should be set to `UTF-8` to ensure paths can be parsed correctly.
+
+For example: `/data/whale_ops/share/$Fund-Product/DA - SANY （三一）/Daily/2025.08.18/file.xlsx`
+
 ### delimiter/field_delimiter [string]
 
 **delimiter** parameter will deprecate after version 2.3.5, please use **field_delimiter** instead.

docs/zh/connector-v2/source/FtpFile.md

@@ -241,6 +241,14 @@ schema {
 
 `active_local` `passive_local`
 
+### control_encoding [string]
+
+FTP 控制连接的字符编码。默认为 `UTF-8`。
+
+当文件路径包含特殊字符（如 `$`、空格、中文字符等）时，需要设置为 `UTF-8` 以确保路径能够正确解析。
+
+例如：`/data/whale_ops/share/$Fund-Product/DA - SANY （三一）/Daily/2025.08.18/file.xlsx`
+
 ### delimiter/field_delimiter [string]
 
 **delimiter** 参数将在 2.3.5 版本后弃用，请使用 **field_delimiter** 代替。

seatunnel-connectors-v2/connector-file/connector-file-ftp/src/main/java/org/apache/seatunnel/connectors/seatunnel/file/ftp/config/FtpConf.java

@@ -61,6 +61,8 @@ public static HadoopConf buildWithConfig(ReadonlyConfig config) {
         ftpOptions.put(
                 "fs.ftp.remote.verification.enabled",
                 String.valueOf(config.get(FtpFileBaseOptions.FTP_REMOTE_VERIFICATION_ENABLED)));
+        ftpOptions.put(
+                "fs.ftp.control.encoding", config.get(FtpFileBaseOptions.FTP_CONTROL_ENCODING));
         hadoopConf.setExtraOptions(ftpOptions);
         return hadoopConf;
     }

seatunnel-connectors-v2/connector-file/connector-file-ftp/src/main/java/org/apache/seatunnel/connectors/seatunnel/file/ftp/config/FtpFileBaseOptions.java

@@ -50,4 +50,10 @@ public class FtpFileBaseOptions extends FileBaseOptions {
                     .defaultValue(true)
                     .withDescription(
                             "Whether to enable remote host verification for FTP data channels (enabled by default)");
+    public static final Option<String> FTP_CONTROL_ENCODING =
+            Options.key("control_encoding")
+                    .stringType()
+                    .defaultValue("UTF-8")
+                    .withDescription(
+                            "Character encoding for FTP control connection. Use UTF-8 to support special characters in file paths");
 }

seatunnel-connectors-v2/connector-file/connector-file-ftp/src/main/java/org/apache/seatunnel/connectors/seatunnel/file/ftp/sink/FtpFileSinkFactory.java

@@ -122,6 +122,7 @@ public OptionRule optionRule() {
                 .optional(FileBaseSinkOptions.CREATE_EMPTY_FILE_WHEN_NO_DATA)
                 .optional(FileBaseSinkOptions.FILENAME_EXTENSION)
                 .optional(FtpFileSourceOptions.FTP_REMOTE_VERIFICATION_ENABLED)
+                .optional(FtpFileSourceOptions.FTP_CONTROL_ENCODING)
                 .build();
     }
 

seatunnel-connectors-v2/connector-file/connector-file-ftp/src/main/java/org/apache/seatunnel/connectors/seatunnel/file/ftp/source/FtpFileSourceFactory.java

@@ -98,6 +98,7 @@ public OptionRule optionRule() {
                 .optional(FileBaseSourceOptions.FILENAME_EXTENSION)
                 .optional(FileBaseSourceOptions.READ_COLUMNS)
                 .optional(FtpFileSourceOptions.FTP_REMOTE_VERIFICATION_ENABLED)
+                .optional(FtpFileSourceOptions.FTP_CONTROL_ENCODING)
                 .build();
     }
 

seatunnel-connectors-v2/connector-file/connector-file-ftp/src/main/java/org/apache/seatunnel/connectors/seatunnel/file/ftp/system/SeaTunnelFTPFileSystem.java

@@ -70,6 +70,7 @@ public class SeaTunnelFTPFileSystem extends FileSystem {
     public static final String FS_FTP_CONNECTION_MODE = "fs.ftp.connection.mode";
     public static final String FS_FTP_REMOTE_VERIFICATION_ENABLED =
             "fs.ftp.remote.verification.enabled";
+    public static final String FS_FTP_CONTROL_ENCODING = "fs.ftp.control.encoding";
 
     public static final String E_SAME_DIRECTORY_ONLY = "only same directory renames are supported";
 
@@ -137,6 +138,10 @@ private FTPClient connect() throws IOException {
         String connectionMode =
                 conf.get(FS_FTP_CONNECTION_MODE, FtpConnectionMode.ACTIVE_LOCAL.getMode());
 
+        // Set control encoding BEFORE connecting - this is critical for special characters
+        String controlEncoding = conf.get(FS_FTP_CONTROL_ENCODING, "UTF-8");
+        client.setControlEncoding(controlEncoding);
+
         // Check if remote verification is enabled
         boolean remoteVerificationEnabled =
                 conf.getBoolean(

seatunnel-e2e/seatunnel-connector-v2-e2e/connector-file-ftp-e2e/src/test/java/org/apache/seatunnel/e2e/connector/file/ftp/FtpFileIT.java

@@ -257,6 +257,57 @@ public void testFtpFileReadAndWrite(TestContainer container)
         Assertions.assertEquals(getFileListFromContainer(homePath + sink02).size(), 1);
     }
 
+    @TestTemplate
+    public void testFtpFileWithSpecialCharactersPath(TestContainer container)
+            throws IOException, InterruptedException {
+        TestHelper helper = new TestHelper(container);
+
+        // Create test file with spaces in path - simpler test to avoid Docker memory issues
+        String specialPath = "/tmp/seatunnel/test spaces";
+        String fileName = "file with spaces.txt";
+        String fullPath = specialPath + "/" + fileName;
+        String homePath = "/home/vsftpd/seatunnel";
+        String containerPath = homePath + fullPath;
+
+        try {
+            // Create directory structure with special characters
+            Container.ExecResult mkdirResult =
+                    ftpContainer.execInContainer("mkdir", "-p", homePath + specialPath);
+            log.info(
+                    "mkdir result: exit code {}, stdout: {}, stderr: {}",
+                    mkdirResult.getExitCode(),
+                    mkdirResult.getStdout(),
+                    mkdirResult.getStderr());
+
+            // Create test file with content
+            String testContent = "name,age,city\nJohn,30,NYC\nJane,25,LA\n";
+            Container.ExecResult createResult =
+                    ftpContainer.execInContainer(
+                            "sh", "-c", "echo '" + testContent + "' > '" + containerPath + "'");
+            log.info(
+                    "create file result: exit code {}, stdout: {}, stderr: {}",
+                    createResult.getExitCode(),
+                    createResult.getStdout(),
+                    createResult.getStderr());
+
+            // Verify file was created
+            Container.ExecResult lsResult =
+                    ftpContainer.execInContainer("ls", "-la", containerPath);
+            Assertions.assertEquals(
+                    0,
+                    lsResult.getExitCode(),
+                    "Failed to create test file with special characters: " + lsResult.getStderr());
+            log.info("File created successfully: {}", lsResult.getStdout());
+
+            // Test reading file with special characters in path using UTF-8 control encoding
+            helper.execute("/text/ftp_special_characters_path_to_assert.conf");
+
+        } finally {
+            // Clean up
+            deleteFileFromContainer(homePath + "/tmp/seatunnel/test\\ spaces");
+        }
+    }
+
     @TestTemplate
     public void testMultipleTableAndSaveMode(TestContainer container)
             throws IOException, InterruptedException {

seatunnel-e2e/seatunnel-connector-v2-e2e/connector-file-ftp-e2e/src/test/resources/text/ftp_special_characters_path_to_assert.conf

@@ -0,0 +1,91 @@
+#
+# 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.
+#
+
+env {
+  parallelism = 1
+  job.mode = "BATCH"
+}
+
+source {
+  FtpFile {
+    host = "ftp"
+    port = 21
+    user = seatunnel
+    password = pass
+    # Test path with spaces
+    path = """/tmp/seatunnel/test spaces/file with spaces.txt"""
+    file_format_type = "text"
+    # Key configuration: UTF-8 control encoding to support special characters in paths
+    control_encoding = "UTF-8"
+    plugin_output = "ftp"
+    schema = {
+      fields {
+        name = string
+        age = int
+        city = string
+      }
+    }
+    field_delimiter = ","
+    skip_header_row_number = 1
+  }
+}
+
+sink {
+  Assert {
+    rules {
+      row_rules = [
+        {
+          rule_type = MAX_ROW
+          rule_value = 2
+        },
+        {
+          rule_type = MIN_ROW
+          rule_value = 2
+        }
+      ],
+      field_rules = [
+        {
+          field_name = name
+          field_type = string
+          field_value = [
+            {
+              rule_type = NOT_NULL
+            }
+          ]
+        },
+        {
+          field_name = age
+          field_type = int
+          field_value = [
+            {
+              rule_type = NOT_NULL
+            }
+          ]
+        },
+        {
+          field_name = city
+          field_type = string
+          field_value = [
+            {
+              rule_type = NOT_NULL
+            }
+          ]
+        }
+      ]
+    }
+  }
+}