[Improve][Connector-V2] Ensure that the FTP connector behaves reliably during directory operation (#8959)

Author: corgy-w

seatunnel-connectors-v2/connector-file/connector-file-ftp/pom.xml

@@ -29,6 +29,10 @@
     <artifactId>connector-file-ftp</artifactId>
     <name>SeaTunnel : Connectors V2 : File : Ftp</name>
 
+    <properties>
+        <mockftpserver.version>3.1.0</mockftpserver.version>
+    </properties>
+
     <dependencies>
         <dependency>
             <groupId>org.apache.seatunnel</groupId>
@@ -48,6 +52,12 @@
                 </exclusion>
             </exclusions>
         </dependency>
+        <dependency>
+            <groupId>org.mockftpserver</groupId>
+            <artifactId>MockFtpServer</artifactId>
+            <version>${mockftpserver.version}</version>
+            <scope>test</scope>
+        </dependency>
     </dependencies>
 
 </project>

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

@@ -127,13 +127,19 @@ public void initialize(URI uri, Configuration conf) throws IOException { // get
      * @throws IOException IOException
      */
     private FTPClient connect() throws IOException {
-        FTPClient client = null;
+        FTPClient client = new FTPClient();
         Configuration conf = getConf();
+        // Get the connection mode from configuration, default to passive_local mode
+        String connectionMode =
+                conf.get(FS_FTP_CONNECTION_MODE, FtpConnectionMode.ACTIVE_LOCAL.getMode());
+
+        // Retrieve host, port, user, and password from configuration
         String host = conf.get(FS_FTP_HOST);
         int port = conf.getInt(FS_FTP_HOST_PORT, FTP.DEFAULT_PORT);
         String user = conf.get(FS_FTP_USER_PREFIX + host);
         String password = conf.get(FS_FTP_PASSWORD_PREFIX + host);
-        client = new FTPClient();
+
+        // Connect to the FTP server
         client.connect(host, port);
         int reply = client.getReplyCode();
         if (!FTPReply.isPositiveCompletion(reply)) {
@@ -143,23 +149,29 @@ private FTPClient connect() throws IOException {
                     NetUtils.UNKNOWN_HOST,
                     0,
                     new ConnectException("Server response " + reply));
-        } else if (client.login(user, password)) {
-            client.setFileTransferMode(FTP.BLOCK_TRANSFER_MODE);
-            client.setFileType(FTP.BINARY_FILE_TYPE);
-            client.setBufferSize(DEFAULT_BUFFER_SIZE);
-        } else {
+        }
+
+        // Log in to the FTP server
+        if (!client.login(user, password)) {
             throw new IOException(
-                    "Login failed on server - "
-                            + host
-                            + ", port - "
-                            + port
-                            + " as user '"
-                            + user
-                            + "'");
+                    String.format(
+                            "Login failed on server - %s, port - %d as user '%s', reply code: %d",
+                            host, port, user, client.getReplyCode()));
         }
 
-        setFsFtpConnectionMode(
-                client, conf.get(FS_FTP_CONNECTION_MODE, FtpConnectionMode.ACTIVE_LOCAL.getMode()));
+        // Set the file type to binary and buffer size
+        client.setFileType(FTP.BINARY_FILE_TYPE);
+        client.setBufferSize(DEFAULT_BUFFER_SIZE);
+        client.setFileTransferMode(FTP.BLOCK_TRANSFER_MODE);
+
+        // Set the connection mode
+        setFsFtpConnectionMode(client, connectionMode);
+
+        // Log successful connection information
+        LOG.info(
+                String.format(
+                        "Successfully connected to FTP server %s:%d in %s",
+                        host, port, connectionMode));
 
         return client;
     }
@@ -170,13 +182,39 @@ private FTPClient connect() throws IOException {
      * @param client FTPClient
      * @param mode mode
      */
-    private void setFsFtpConnectionMode(FTPClient client, String mode) {
-        switch (FtpConnectionMode.fromMode(mode)) {
+    private void setFsFtpConnectionMode(FTPClient client, String mode) throws IOException {
+        FtpConnectionMode connectionMode = FtpConnectionMode.fromMode(mode);
+        switch (connectionMode) {
             case PASSIVE_LOCAL:
                 client.enterLocalPassiveMode();
+                LOG.info("Using passive mode for FTP connection");
                 break;
             case ACTIVE_LOCAL:
-                client.enterLocalActiveMode();
+                // Create a test directory to check if active mode is working
+                String pathName = "/.ftptest" + System.currentTimeMillis();
+                try {
+                    client.enterLocalActiveMode();
+                    // test active mode is working or not
+                    boolean created = client.makeDirectory(pathName);
+                    if (!created) {
+                        LOG.warn("Active mode failed, switching to passive mode");
+                        throw new IOException("FTP connection active mode test failed");
+                    }
+
+                    LOG.info("Using active mode for FTP connection");
+                } catch (IOException e) {
+                    // if active mode failed, switch to passive mode
+                    client.enterLocalPassiveMode();
+                    // update the connection mode to passive mode
+                    getConf()
+                            .set(FS_FTP_CONNECTION_MODE, FtpConnectionMode.PASSIVE_LOCAL.getMode());
+                } finally {
+                    // delete the test directory if it was created
+                    FTPFile[] files = client.listFiles(pathName);
+                    if (files != null && files.length > 0) {
+                        client.deleteFile(pathName);
+                    }
+                }
                 break;
             default:
                 log.warn(
@@ -548,30 +586,50 @@ public boolean mkdirs(Path file, FsPermission permission) throws IOException {
      */
     private boolean mkdirs(FTPClient client, Path file, FsPermission permission)
             throws IOException {
-        boolean created = true;
         Path workDir = new Path(client.printWorkingDirectory());
         Path absolute = makeAbsolute(workDir, file);
-        String pathName = absolute.getName();
-        if (!exists(client, absolute)) {
-            Path parent = absolute.getParent();
-            created = parent == null || mkdirs(client, parent, FsPermission.getDirDefault());
-            if (created) {
-                String parentDir = parent.toUri().getPath();
-                client.changeWorkingDirectory(parentDir);
-                LOG.debug("Creating directory " + pathName);
-                created = client.makeDirectory(pathName);
+        // If directory already exists, return true
+        if (exists(client, absolute)) {
+            if (isFile(client, absolute)) {
+                throw new ParentNotDirectoryException(
+                        String.format(
+                                "Can't make directory for path %s since it is a file.", absolute));
             }
-        } else if (isFile(client, absolute)) {
-            throw new ParentNotDirectoryException(
+            return true;
+        }
+
+        // Create parent directories if they don't exist
+        Path parent = absolute.getParent();
+        if (parent != null && !exists(client, parent)) {
+            mkdirs(client, parent, FsPermission.getDirDefault());
+        }
+
+        // Create the directory
+        String pathName = absolute.getName();
+        String parentDir = parent != null ? parent.toUri().getPath() : "/";
+
+        // Change to parent directory
+        if (!client.changeWorkingDirectory(parentDir)) {
+            throw new IOException(
                     String.format(
-                            "Can't make directory for path %s since it is a file.", absolute));
-        } else {
-            LOG.debug("Skipping creation of existing directory " + file);
+                            "Failed to change working directory to %s, FTP reply code: %d, reply string: %s",
+                            parentDir, client.getReplyCode(), client.getReplyString()));
         }
+        // Create directory
+        boolean created = client.makeDirectory(pathName);
         if (!created) {
-            LOG.debug("Failed to create " + file);
+            // Double check if directory was actually created (some FTP servers don't return true)
+            if (!exists(client, absolute)) {
+                throw new IOException(
+                        String.format(
+                                "Failed to create directory %s in %s, FTP reply code: %d, reply string: %s",
+                                pathName,
+                                parentDir,
+                                client.getReplyCode(),
+                                client.getReplyString()));
+            }
         }
-        return created;
+        return true;
     }
 
     /**