You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@seatunnel.apache.org by GitBox <gi...@apache.org> on 2022/08/20 15:46:22 UTC
[GitHub] [incubator-seatunnel] TyrantLucifer commented on a diff in pull request #2483: add ftp file sink
TyrantLucifer commented on code in PR #2483:
URL: https://github.com/apache/incubator-seatunnel/pull/2483#discussion_r950708486
##########
seatunnel-connectors-v2/connector-file/connector-file-base/src/main/java/org/apache/seatunnel/connectors/seatunnel/file/sink/TransactionStateFileSinkWriter.java:
##########
@@ -50,7 +50,7 @@ public class TransactionStateFileSinkWriter implements SinkWriter<SeaTunnelRow,
public TransactionStateFileSinkWriter(@NonNull SeaTunnelRowType seaTunnelRowTypeInfo,
@NonNull Config pluginConfig,
- @NonNull SinkWriter.Context context,
Review Comment:
The same as above.
##########
seatunnel-connectors-v2/connector-file/connector-file-base/src/main/java/org/apache/seatunnel/connectors/seatunnel/file/sink/FileSinkWriterWithTransaction.java:
##########
@@ -89,7 +89,7 @@ public FileSinkWriterWithTransaction(@NonNull SeaTunnelRowType seaTunnelRowTypeI
public FileSinkWriterWithTransaction(@NonNull SeaTunnelRowType seaTunnelRowTypeInfo,
@NonNull Config pluginConfig,
- @NonNull SinkWriter.Context context,
Review Comment:
The same as above.
##########
seatunnel-connectors-v2/connector-file/connector-file-ftp/src/main/java/org/apache/seatunnel/connectors/seatunnel/file/sink/ftp/FtpFileSink.java:
##########
@@ -0,0 +1,32 @@
+package org.apache.seatunnel.connectors.seatunnel.file.sink.ftp;
Review Comment:
Please add license header in your code file.
##########
seatunnel-connectors-v2/connector-file/connector-file-ftp/pom.xml:
##########
@@ -0,0 +1,57 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<!--
+
+ 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.
+
+-->
+<project xmlns="http://maven.apache.org/POM/4.0.0"
+ xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
+ xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
+ <parent>
+ <artifactId>connector-file</artifactId>
+ <groupId>org.apache.seatunnel</groupId>
+ <version>${revision}</version>
+ </parent>
+
+ <modelVersion>4.0.0</modelVersion>
+ <artifactId>connector-file-ftp</artifactId>
+
+ <dependencies>
+ <dependency>
+ <groupId>org.apache.seatunnel</groupId>
+ <artifactId>connector-file-base</artifactId>
+ <version>${project.version}</version>
+ </dependency>
+ <dependency>
+ <groupId>commons-net</groupId>
+ <artifactId>commons-net</artifactId>
+ <version>3.6</version>
Review Comment:
Please add this dependency in main pom file and use dependencyManager to manage it.
##########
seatunnel-connectors-v2/connector-file/connector-file-ftp/src/main/java/org/apache/seatunnel/connectors/seatunnel/file/sink/ftp/util/FtpFileUtils.java:
##########
@@ -0,0 +1,176 @@
+/*
+ * 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.
+ */
+
+package org.apache.seatunnel.connectors.seatunnel.file.sink.ftp.util;
+
+import lombok.NonNull;
+import org.apache.commons.net.ftp.FTPClient;
+import org.apache.commons.net.ftp.FTPFile;
+import org.apache.commons.net.ftp.FTPReply;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.io.ByteArrayInputStream;
+import java.io.IOException;
+import java.io.InputStream;
+import java.io.OutputStream;
+import java.net.SocketException;
+import java.util.StringTokenizer;
+
+public class FtpFileUtils {
+ private static final Logger LOGGER = LoggerFactory.getLogger(FtpFileUtils.class);
+ public static FTPClient FTPCLIENT;
+ public static String FTP_PASSWORD;
+ public static String FTP_USERNAME;
+ public static String FTP_HOST;
+ public static Integer FTP_PORT;
+ public static final int FTP_CONNECT_MAX_TIMEOUT = 30000;
+
+ public static FTPClient getFTPClient(){
+ return getFTPClient(FTP_HOST, FTP_PORT, FTP_USERNAME, FTP_PASSWORD);
+ }
+
+ public static FTPClient getFTPClient(String ftpHost, int ftpPort, String ftpUserName, String ftpPassword) {
+ try {
+ FTPCLIENT = new FTPClient();
Review Comment:
I think `new FTPClient()` it should be instantiated only once because multiple instantiation can cause resource leakage.
So I think use `Singleton pattern` is best.
##########
seatunnel-connectors-v2/connector-file/connector-file-ftp/src/main/java/org/apache/seatunnel/connectors/seatunnel/file/sink/ftp/filesystem/FtpFileSystemCommitter.java:
##########
@@ -0,0 +1,60 @@
+/*
+ * 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.
+ */
+
+package org.apache.seatunnel.connectors.seatunnel.file.sink.ftp.filesystem;
+
+import org.apache.seatunnel.connectors.seatunnel.file.sink.FileAggregatedCommitInfo;
+import org.apache.seatunnel.connectors.seatunnel.file.sink.ftp.util.FtpFileUtils;
+import org.apache.seatunnel.connectors.seatunnel.file.sink.spi.FileSystemCommitter;
+
+import lombok.NonNull;
+
+import java.io.IOException;
+import java.util.Map;
+
+public class FtpFileSystemCommitter implements FileSystemCommitter {
+ @Override
+ public void commitTransaction(@NonNull FileAggregatedCommitInfo aggregateCommitInfo) throws IOException {
+ for (Map.Entry<String, Map<String, String>> entry : aggregateCommitInfo.getTransactionMap().entrySet()) {
+ for (Map.Entry<String, String> mvFileEntry : entry.getValue().entrySet()) {
+ String key = mvFileEntry.getKey();
+ String value = mvFileEntry.getValue();
+ FtpFileUtils.renameFile(key, value);
+ }
+ // delete the transaction dir
+ FtpFileUtils.deleteFiles(entry.getKey());
+
Review Comment:
Please remove redundant line.
##########
seatunnel-connectors-v2/connector-file/connector-file-ftp/src/main/java/org/apache/seatunnel/connectors/seatunnel/file/sink/ftp/config/Config.java:
##########
@@ -0,0 +1,16 @@
+package org.apache.seatunnel.connectors.seatunnel.file.sink.ftp.config;
+
+public class Config {
+
+ public static final String FTP_PASSWORD = "ftp_password";
+
+
Review Comment:
Please remove redundant line.
##########
seatunnel-connectors-v2/connector-file/connector-file-ftp/src/main/java/org/apache/seatunnel/connectors/seatunnel/file/sink/ftp/config/Config.java:
##########
@@ -0,0 +1,16 @@
+package org.apache.seatunnel.connectors.seatunnel.file.sink.ftp.config;
+
+public class Config {
+
+ public static final String FTP_PASSWORD = "ftp_password";
+
+
+ public static final String FTP_USERNAME = "ftp_username";
+
+
Review Comment:
The same as above.
##########
seatunnel-connectors-v2/connector-file/connector-file-base/src/main/java/org/apache/seatunnel/connectors/seatunnel/file/source/split/FileSourceSplitEnumerator.java:
##########
@@ -37,12 +37,12 @@ public class FileSourceSplitEnumerator implements SourceSplitEnumerator<FileSour
private Set<FileSourceSplit> assignedSplit;
private final List<String> filePaths;
- public FileSourceSplitEnumerator(SourceSplitEnumerator.Context<FileSourceSplit> context, List<String> filePaths) {
Review Comment:
The same as above.
##########
seatunnel-connectors-v2/connector-file/connector-file-ftp/pom.xml:
##########
@@ -0,0 +1,57 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<!--
+
+ 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.
+
+-->
+<project xmlns="http://maven.apache.org/POM/4.0.0"
+ xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
+ xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
+ <parent>
+ <artifactId>connector-file</artifactId>
+ <groupId>org.apache.seatunnel</groupId>
+ <version>${revision}</version>
+ </parent>
+
+ <modelVersion>4.0.0</modelVersion>
+ <artifactId>connector-file-ftp</artifactId>
+
+ <dependencies>
+ <dependency>
+ <groupId>org.apache.seatunnel</groupId>
+ <artifactId>connector-file-base</artifactId>
+ <version>${project.version}</version>
+ </dependency>
+ <dependency>
+ <groupId>commons-net</groupId>
+ <artifactId>commons-net</artifactId>
+ <version>3.6</version>
+ </dependency>
+ <dependency>
+ <groupId>org.apache.seatunnel</groupId>
+ <artifactId>seatunnel-api</artifactId>
+ <version>2.1.3-SNAPSHOT</version>
Review Comment:
Please use ${project.version} instead.
##########
seatunnel-connectors-v2/connector-file/connector-file-base/src/main/java/org/apache/seatunnel/connectors/seatunnel/file/source/split/FileSourceSplitEnumerator.java:
##########
@@ -37,12 +37,12 @@ public class FileSourceSplitEnumerator implements SourceSplitEnumerator<FileSour
private Set<FileSourceSplit> assignedSplit;
private final List<String> filePaths;
- public FileSourceSplitEnumerator(SourceSplitEnumerator.Context<FileSourceSplit> context, List<String> filePaths) {
+ public FileSourceSplitEnumerator(Context<FileSourceSplit> context, List<String> filePaths) {
this.context = context;
this.filePaths = filePaths;
}
- public FileSourceSplitEnumerator(SourceSplitEnumerator.Context<FileSourceSplit> context, List<String> filePaths,
Review Comment:
The same as above.
##########
seatunnel-connectors-v2/connector-file/connector-file-base/src/main/java/org/apache/seatunnel/connectors/seatunnel/file/sink/TransactionStateFileSinkWriter.java:
##########
@@ -89,7 +89,7 @@ public TransactionStateFileSinkWriter(@NonNull SeaTunnelRowType seaTunnelRowType
public TransactionStateFileSinkWriter(@NonNull SeaTunnelRowType seaTunnelRowTypeInfo,
@NonNull Config pluginConfig,
- @NonNull SinkWriter.Context context,
Review Comment:
The same as above.
##########
seatunnel-connectors-v2/connector-file/connector-file-base/src/main/java/org/apache/seatunnel/connectors/seatunnel/file/source/BaseFileSourceReader.java:
##########
@@ -34,10 +34,10 @@ public class BaseFileSourceReader implements SourceReader<SeaTunnelRow, FileSour
private static final long THREAD_WAIT_TIME = 500L;
private final ReadStrategy readStrategy;
private final HadoopConf hadoopConf;
- private final SourceReader.Context context;
+ private final Context context;
private final Set<FileSourceSplit> sourceSplits;
- public BaseFileSourceReader(ReadStrategy readStrategy, HadoopConf hadoopConf, SourceReader.Context context) {
Review Comment:
The same as above.
##########
seatunnel-connectors-v2/connector-file/connector-file-ftp/src/main/java/org/apache/seatunnel/connectors/seatunnel/file/sink/ftp/FtpFileSinkPlugin.java:
##########
@@ -0,0 +1,45 @@
+package org.apache.seatunnel.connectors.seatunnel.file.sink.ftp;
Review Comment:
Please add license header in your code file.
##########
seatunnel-connectors-v2/connector-file/connector-file-base/src/main/java/org/apache/seatunnel/connectors/seatunnel/file/source/BaseFileSourceReader.java:
##########
@@ -34,10 +34,10 @@ public class BaseFileSourceReader implements SourceReader<SeaTunnelRow, FileSour
private static final long THREAD_WAIT_TIME = 500L;
private final ReadStrategy readStrategy;
private final HadoopConf hadoopConf;
- private final SourceReader.Context context;
Review Comment:
The same as above.
##########
seatunnel-connectors-v2/connector-file/connector-file-base/src/main/java/org/apache/seatunnel/connectors/seatunnel/file/sink/FileSinkWriterWithTransaction.java:
##########
@@ -50,7 +50,7 @@ public class FileSinkWriterWithTransaction implements SinkWriter<SeaTunnelRow, F
public FileSinkWriterWithTransaction(@NonNull SeaTunnelRowType seaTunnelRowTypeInfo,
@NonNull Config pluginConfig,
- @NonNull SinkWriter.Context context,
Review Comment:
I think it's best to keep class prefix, because `Context` was contained both `SourceReader` and `SourceWriter`, keep class prefix we can distinguish which class the `Context` comes from.
##########
seatunnel-connectors-v2/connector-file/connector-file-ftp/src/main/java/org/apache/seatunnel/connectors/seatunnel/file/sink/ftp/filesystem/FtpFileSystemCommitter.java:
##########
@@ -0,0 +1,60 @@
+/*
+ * 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.
+ */
+
+package org.apache.seatunnel.connectors.seatunnel.file.sink.ftp.filesystem;
+
+import org.apache.seatunnel.connectors.seatunnel.file.sink.FileAggregatedCommitInfo;
+import org.apache.seatunnel.connectors.seatunnel.file.sink.ftp.util.FtpFileUtils;
+import org.apache.seatunnel.connectors.seatunnel.file.sink.spi.FileSystemCommitter;
+
+import lombok.NonNull;
+
+import java.io.IOException;
+import java.util.Map;
+
+public class FtpFileSystemCommitter implements FileSystemCommitter {
+ @Override
+ public void commitTransaction(@NonNull FileAggregatedCommitInfo aggregateCommitInfo) throws IOException {
+ for (Map.Entry<String, Map<String, String>> entry : aggregateCommitInfo.getTransactionMap().entrySet()) {
+ for (Map.Entry<String, String> mvFileEntry : entry.getValue().entrySet()) {
+ String key = mvFileEntry.getKey();
+ String value = mvFileEntry.getValue();
+ FtpFileUtils.renameFile(key, value);
+ }
+ // delete the transaction dir
+ FtpFileUtils.deleteFiles(entry.getKey());
+
+ }
+ }
+
+ @Override
+ public void abortTransaction(@NonNull FileAggregatedCommitInfo aggregateCommitInfo) throws IOException {
+ for (Map.Entry<String, Map<String, String>> entry : aggregateCommitInfo.getTransactionMap().entrySet()) {
+ // rollback the file
+ for (Map.Entry<String, String> mvFileEntry : entry.getValue().entrySet()) {
+ String oldFile = mvFileEntry.getKey();
+ String newFile = mvFileEntry.getValue();
+ if (FtpFileUtils.fileExist(newFile) && !FtpFileUtils.fileExist(oldFile)) {
+ FtpFileUtils.renameFile(mvFileEntry.getValue(), mvFileEntry.getKey());
+ }
+ }
+ // delete the transaction dir
+ FtpFileUtils.deleteFiles(entry.getKey());
+
Review Comment:
The same as above.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@seatunnel.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org