[Fix][Connector-V2] Fix file binary format sync convert directory to …

…file (#7942)
apache · Oct 31, 2024 · 86ae927 · 86ae927
1 parent cb9c257
commit 86ae927
Show file tree

Hide file tree

Showing 5 changed files with 72 additions and 2 deletions.
diff --git a/.github/workflows/backend.yml b/.github/workflows/backend.yml
@@ -364,7 +364,7 @@ jobs:
       matrix:
         java: [ '8', '11' ]
         os: [ 'ubuntu-latest' ]
-    timeout-minutes: 120
+    timeout-minutes: 180
     steps:
       - uses: actions/checkout@v2
       - name: Set up JDK ${{ matrix.java }}

diff --git a/...ain/java/org/apache/seatunnel/connectors/seatunnel/file/hadoop/HadoopFileSystemProxy.java b/...ain/java/org/apache/seatunnel/connectors/seatunnel/file/hadoop/HadoopFileSystemProxy.java
@@ -64,6 +64,10 @@ public boolean fileExist(@NonNull String filePath) throws IOException {
         return execute(() -> getFileSystem().exists(new Path(filePath)));
     }
 
+    public boolean isFile(@NonNull String filePath) throws IOException {
+        return execute(() -> getFileSystem().getFileStatus(new Path(filePath)).isFile());
+    }
+
     public void createFile(@NonNull String filePath) throws IOException {
         execute(
                 () -> {

diff --git a/...java/org/apache/seatunnel/connectors/seatunnel/file/source/reader/BinaryReadStrategy.java b/...java/org/apache/seatunnel/connectors/seatunnel/file/source/reader/BinaryReadStrategy.java
@@ -55,7 +55,7 @@ public void read(String path, String tableId, Collector<SeaTunnelRow> output)
             throws IOException, FileConnectorException {
         try (InputStream inputStream = hadoopFileSystemProxy.getInputStream(path)) {
             String relativePath;
-            if (basePath.isFile()) {
+            if (hadoopFileSystemProxy.isFile(basePath.getAbsolutePath())) {
                 relativePath = basePath.getName();
             } else {
                 relativePath =

diff --git a/...tor-file-ftp-e2e/src/test/java/org/apache/seatunnel/e2e/connector/file/ftp/FtpFileIT.java b/...tor-file-ftp-e2e/src/test/java/org/apache/seatunnel/e2e/connector/file/ftp/FtpFileIT.java
@@ -187,6 +187,25 @@ public void testFtpFileReadAndWriteForPassive(TestContainer container)
         deleteFileFromContainer(homePath);
     }
 
+    @TestTemplate
+    public void testFtpToFtpForBinary(TestContainer container)
+            throws IOException, InterruptedException {
+
+        Container.ExecResult execResult = container.executeJob("/text/ftp_to_ftp_for_binary.conf");
+        Assertions.assertEquals(0, execResult.getExitCode(), execResult.getStderr());
+
+        String homePath = "/home/vsftpd/seatunnel/uploads/seatunnel";
+        Assertions.assertEquals(1, getFileListFromContainer(homePath).size());
+
+        // Confirm data is written correctly
+        Container.ExecResult resultExecResult =
+                ftpContainer.execInContainer(
+                        "sh", "-c", "awk 'END {print NR}' " + homePath + "/e2e.txt");
+        Assertions.assertEquals("5", resultExecResult.getStdout().trim());
+
+        deleteFileFromContainer(homePath);
+    }
+
     private void assertJobExecution(TestContainer container, String configPath, List<String> params)
             throws IOException, InterruptedException {
         Container.ExecResult execResult = container.executeJob(configPath, params);

diff --git a/...onnector-v2-e2e/connector-file-ftp-e2e/src/test/resources/text/ftp_to_ftp_for_binary.conf b/...onnector-v2-e2e/connector-file-ftp-e2e/src/test/resources/text/ftp_to_ftp_for_binary.conf
@@ -0,0 +1,47 @@
+#
+# 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
+    path= "/tmp/seatunnel/read/text/name=tyrantlucifer/hobby=coding/e2e.txt"
+    file_format_type= "binary"
+    encoding = "UTF-8"
+  }
+}
+
+
+sink {
+  FtpFile {
+    host = "ftp"
+    port = 21
+    user = seatunnel
+    password = pass
+    tmp_path = "/upload-tmp/seatunnel"
+    path= "/uploads/seatunnel"
+    file_format_type= "binary"
+    encoding="UTF-8"
+  }
+}