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: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
