Merge branch 'hotfix_1.8_3.10.x_31248' into 1.8_release_3.10.x

Author: gituser

flinkx-ftp/flinkx-ftp-core/src/main/java/com/dtstack/flinkx/ftp/FtpHandler.java

@@ -143,28 +143,52 @@ public boolean isFileExist(String filePath) {
     @Override
     public List<String> getFiles(String path) {
         List<String> sources = new ArrayList<>();
-        if(isDirExist(path)) {
+        if(isFileExist(path)) {
+            sources.add(path);
+            return sources;
+        }else{
+            path = path + SP;
+        }
+        try {
+            ftpClient.enterLocalPassiveMode();
+            FTPFile[] ftpFiles = ftpClient.listFiles(new String(path.getBytes(StandardCharsets.UTF_8),FTP.DEFAULT_CONTROL_ENCODING));
+            if(ftpFiles != null) {
+                for(FTPFile ftpFile : ftpFiles) {
+                    sources.addAll(getFiles(path + ftpFile.getName(), ftpFile));
+                }
+            }
+        } catch (IOException e) {
+            LOG.error("", e);
+            throw new RuntimeException(e);
+        }
+        return sources;
+    }
+
+    /**
+     * 递归获取指定路径下的所有文件(暂无过滤)
+     * isDirExist()、isFileExist()方法在Windows系统下判断有误
+     * @param path
+     * @param file
+     * @return
+     * @throws IOException
+     */
+    private List<String> getFiles(String path, FTPFile file)throws IOException {
+        List<String> sources = new ArrayList<>();
+        if(file.isDirectory()){
             if(!path.endsWith(SP)) {
                 path = path + SP;
             }
-            try {
-                ftpClient.enterLocalPassiveMode();
-                FTPFile[] ftpFiles = ftpClient.listFiles(new String(path.getBytes(StandardCharsets.UTF_8),FTP.DEFAULT_CONTROL_ENCODING));
-                if(ftpFiles != null) {
-                    for(FTPFile ftpFile : ftpFiles) {
-                        sources.addAll(getFiles(path + ftpFile.getName()));
-                    }
+            ftpClient.enterLocalPassiveMode();
+            FTPFile[] ftpFiles = ftpClient.listFiles(new String(path.getBytes(StandardCharsets.UTF_8),FTP.DEFAULT_CONTROL_ENCODING));
+            if(ftpFiles != null) {
+                for(FTPFile ftpFile : ftpFiles) {
+                    sources.addAll(getFiles(path + ftpFile.getName(), ftpFile));
                 }
-            } catch (IOException e) {
-                LOG.error("", e);
-                throw new RuntimeException(e);
             }
-
-        } else if(isFileExist(path)) {
+        }else{
             sources.add(path);
-            return sources;
         }
-
+        LOG.info("path = {}, FTPFile is directory = {}", path, file.isDirectory());
         return sources;
     }
 

flinkx-ftp/flinkx-ftp-reader/src/main/java/com/dtstack/flinkx/ftp/reader/FtpInputFormat.java

@@ -19,17 +19,16 @@
 package com.dtstack.flinkx.ftp.reader;
 
 import com.dtstack.flinkx.constants.ConstantValue;
-import com.dtstack.flinkx.ftp.EProtocol;
 import com.dtstack.flinkx.ftp.FtpConfig;
-import com.dtstack.flinkx.ftp.FtpHandler;
 import com.dtstack.flinkx.ftp.FtpHandlerFactory;
 import com.dtstack.flinkx.ftp.IFtpHandler;
-import com.dtstack.flinkx.ftp.SftpHandler;
 import com.dtstack.flinkx.inputformat.BaseRichInputFormat;
 import com.dtstack.flinkx.reader.MetaColumn;
+import com.dtstack.flinkx.util.GsonUtil;
 import com.dtstack.flinkx.util.StringUtil;
 import org.apache.flink.core.io.InputSplit;
 import org.apache.flink.types.Row;
+
 import java.io.IOException;
 import java.util.ArrayList;
 import java.util.List;
@@ -77,7 +76,7 @@ public InputSplit[] createInputSplitsInternal(int minNumSplits) throws Exception
                 files.addAll(ftpHandler.getFiles(p.trim()));
             }
         }
-
+        LOG.info("FTP files = {}", GsonUtil.GSON.toJson(files));
         int numSplits = (Math.min(files.size(), minNumSplits));
         FtpInputSplit[] ftpInputSplits = new FtpInputSplit[numSplits];
         for(int index = 0; index < numSplits; ++index) {

flinkx-ftp/flinkx-ftp-writer/src/main/java/com/dtstack/flinkx/ftp/writer/FtpOutputFormat.java

@@ -18,6 +18,7 @@
 
 package com.dtstack.flinkx.ftp.writer;
 
+import com.dtstack.flinkx.constants.ConstantValue;
 import com.dtstack.flinkx.exception.WriteRecordException;
 import com.dtstack.flinkx.ftp.FtpConfig;
 import com.dtstack.flinkx.ftp.FtpHandlerFactory;
@@ -31,7 +32,7 @@
 
 import java.io.IOException;
 import java.io.OutputStream;
-import java.util.Arrays;
+import java.util.Collections;
 import java.util.List;
 import java.util.function.Predicate;
 
@@ -56,8 +57,6 @@ public class FtpOutputFormat extends BaseFileOutputFormat {
 
     private transient OutputStream os;
 
-    private static final String DOT = ".";
-
     private static final String FILE_SUFFIX = ".csv";
 
     private static final String OVERWRITE_MODE = "overwrite";
@@ -133,7 +132,7 @@ protected void nextBlock(){
 
     @Override
     public void moveTemporaryDataBlockFileToDirectory(){
-        if (currentBlockFileName == null || !currentBlockFileName.startsWith(DOT)){
+        if (currentBlockFileName == null || !currentBlockFileName.startsWith(ConstantValue.POINT_SYMBOL)){
             return;
         }
 
@@ -242,9 +241,9 @@ protected void waitForAllTasksToFinish(){
 
     @Override
     protected void coverageData(){
-        boolean cleanPath = restoreConfig.isRestore() && OVERWRITE_MODE.equalsIgnoreCase(ftpConfig.getWriteMode()) && !SP.equals(ftpConfig.getPath());
+        boolean cleanPath = OVERWRITE_MODE.equalsIgnoreCase(ftpConfig.getWriteMode()) && !SP.equals(ftpConfig.getPath());
         if(cleanPath){
-            ftpHandler.deleteAllFilesInDir(ftpConfig.getPath(), Arrays.asList(tmpPath));
+            ftpHandler.deleteAllFilesInDir(ftpConfig.getPath(), Collections.singletonList(tmpPath));
         }
     }
 
@@ -271,7 +270,7 @@ protected void moveAllTemporaryDataFileToDirectory() throws IOException {
             List<String> files = ftpHandler.getFiles(tmpPath);
             for (String file : files) {
                 String fileName = file.substring(file.lastIndexOf(SP) + 1);
-                if (fileName.endsWith(FILE_SUFFIX) && !fileName.startsWith(DOT)){
+                if (fileName.endsWith(FILE_SUFFIX) && !fileName.startsWith(ConstantValue.POINT_SYMBOL)){
                     String newPath = ftpConfig.getPath() + SP + fileName;
                     LOG.info("Move file {} to path {}", file, newPath);
                     ftpHandler.rename(file, newPath);