NIFI-12627 Extract nifi-file-transfer from nifi-standard-processors

This closes apache#8262 Signed-off-by: David Handermann <exceptionfactory@apache.org>
jrsteinebrey · Jan 18, 2024 · 359ffa5 · 359ffa5
1 parent 53bb995
commit 359ffa5
Show file tree

Hide file tree

Showing 31 changed files with 137 additions and 52 deletions.
diff --git a/nifi-code-coverage/pom.xml b/nifi-code-coverage/pom.xml
@@ -1002,6 +1002,11 @@
             <artifactId>nifi-event-transport</artifactId>
             <version>2.0.0-SNAPSHOT</version>
         </dependency>
+        <dependency>
+            <groupId>org.apache.nifi</groupId>
+            <artifactId>nifi-file-transfer</artifactId>
+            <version>2.0.0-SNAPSHOT</version>
+        </dependency>
         <dependency>
             <groupId>org.apache.nifi</groupId>
             <artifactId>nifi-hadoop-utils</artifactId>

diff --git a/nifi-nar-bundles/nifi-extension-utils/nifi-file-transfer/pom.xml b/nifi-nar-bundles/nifi-extension-utils/nifi-file-transfer/pom.xml
@@ -0,0 +1,59 @@
+<?xml version="1.0"?>
+<!--
+  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 https://maven.apache.org/xsd/maven-4.0.0.xsd">
+    <modelVersion>4.0.0</modelVersion>
+    <parent>
+        <groupId>org.apache.nifi</groupId>
+        <artifactId>nifi-extension-utils</artifactId>
+        <version>2.0.0-SNAPSHOT</version>
+    </parent>
+    <artifactId>nifi-file-transfer</artifactId>
+    <packaging>jar</packaging>
+    <dependencies>
+        <dependency>
+            <groupId>org.apache.nifi</groupId>
+            <artifactId>nifi-api</artifactId>
+        </dependency>
+        <dependency>
+            <groupId>org.apache.nifi</groupId>
+            <artifactId>nifi-listed-entity</artifactId>
+            <version>2.0.0-SNAPSHOT</version>
+        </dependency>
+        <dependency>
+            <groupId>org.apache.nifi</groupId>
+            <artifactId>nifi-property-utils</artifactId>
+            <version>2.0.0-SNAPSHOT</version>
+        </dependency>
+        <dependency>
+            <groupId>org.apache.nifi</groupId>
+            <artifactId>nifi-record</artifactId>
+            <version>2.0.0-SNAPSHOT</version>
+        </dependency>
+        <dependency>
+            <groupId>org.apache.nifi</groupId>
+            <artifactId>nifi-utils</artifactId>
+            <version>2.0.0-SNAPSHOT</version>
+        </dependency>
+        <dependency>
+            <groupId>commons-io</groupId>
+            <artifactId>commons-io</artifactId>
+        </dependency>
+        <dependency>
+            <groupId>org.apache.commons</groupId>
+            <artifactId>commons-lang3</artifactId>
+        </dependency>
+    </dependencies>
+</project>
diff --git a/...rocessors/standard/FetchFileTransfer.java → ...util/file/transfer/FetchFileTransfer.java b/...rocessors/standard/FetchFileTransfer.java → ...util/file/transfer/FetchFileTransfer.java
@@ -15,7 +15,7 @@
  * limitations under the License.
  */
 
-package org.apache.nifi.processors.standard;
+package org.apache.nifi.processor.util.file.transfer;
 
 import org.apache.commons.lang3.StringUtils;
 import org.apache.nifi.annotation.lifecycle.OnScheduled;
@@ -32,8 +32,6 @@
 import org.apache.nifi.processor.Relationship;
 import org.apache.nifi.processor.exception.ProcessException;
 import org.apache.nifi.processor.util.StandardValidators;
-import org.apache.nifi.processors.standard.util.FileTransfer;
-import org.apache.nifi.processors.standard.util.PermissionDeniedException;
 import org.apache.nifi.util.StopWatch;
 import org.apache.nifi.util.Tuple;
 
@@ -58,19 +56,19 @@
  */
 public abstract class FetchFileTransfer extends AbstractProcessor {
 
-    static final AllowableValue COMPLETION_NONE = new AllowableValue("None", "None", "Leave the file as-is");
-    static final AllowableValue COMPLETION_MOVE = new AllowableValue("Move File", "Move File", "Move the file to the directory specified by the <Move Destination Directory> property");
-    static final AllowableValue COMPLETION_DELETE = new AllowableValue("Delete File", "Delete File", "Deletes the original file from the remote system");
-    static final String FAILURE_REASON_ATTRIBUTE = "fetch.failure.reason";
+    public static final AllowableValue COMPLETION_NONE = new AllowableValue("None", "None", "Leave the file as-is");
+    public static final AllowableValue COMPLETION_MOVE = new AllowableValue("Move File", "Move File", "Move the file to the directory specified by the <Move Destination Directory> property");
+    public static final AllowableValue COMPLETION_DELETE = new AllowableValue("Delete File", "Delete File", "Deletes the original file from the remote system");
+    public static final String FAILURE_REASON_ATTRIBUTE = "fetch.failure.reason";
 
-    static final PropertyDescriptor HOSTNAME = new PropertyDescriptor.Builder()
+    public static final PropertyDescriptor HOSTNAME = new PropertyDescriptor.Builder()
         .name("Hostname")
         .description("The fully-qualified hostname or IP address of the host to fetch the data from")
         .addValidator(StandardValidators.NON_EMPTY_VALIDATOR)
         .expressionLanguageSupported(ExpressionLanguageScope.FLOWFILE_ATTRIBUTES)
         .required(true)
         .build();
-    static final PropertyDescriptor UNDEFAULTED_PORT = new PropertyDescriptor.Builder()
+    public static final PropertyDescriptor UNDEFAULTED_PORT = new PropertyDescriptor.Builder()
         .name("Port")
         .description("The port to connect to on the remote host to fetch the data from")
         .addValidator(StandardValidators.NON_EMPTY_VALIDATOR)
@@ -91,7 +89,7 @@ public abstract class FetchFileTransfer extends AbstractProcessor {
         .addValidator(StandardValidators.NON_EMPTY_VALIDATOR)
         .expressionLanguageSupported(ExpressionLanguageScope.FLOWFILE_ATTRIBUTES)
         .build();
-    static final PropertyDescriptor COMPLETION_STRATEGY = new PropertyDescriptor.Builder()
+    public static final PropertyDescriptor COMPLETION_STRATEGY = new PropertyDescriptor.Builder()
         .name("Completion Strategy")
         .description("Specifies what to do with the original file on the server once it has been pulled into NiFi. If the Completion Strategy fails, a warning will be "
             + "logged but the data will still be transferred.")
@@ -100,14 +98,14 @@ public abstract class FetchFileTransfer extends AbstractProcessor {
         .defaultValue(COMPLETION_NONE.getValue())
         .required(true)
         .build();
-    static final PropertyDescriptor MOVE_CREATE_DIRECTORY = new PropertyDescriptor.Builder()
+    public static final PropertyDescriptor MOVE_CREATE_DIRECTORY = new PropertyDescriptor.Builder()
             .fromPropertyDescriptor(FileTransfer.CREATE_DIRECTORY).description(String.format("Used when '%s' is '%s'. %s",
                     COMPLETION_STRATEGY.getDisplayName(),
                     COMPLETION_MOVE.getDisplayName(),
                     FileTransfer.CREATE_DIRECTORY.getDescription()))
             .required(false)
             .build();
-    static final PropertyDescriptor MOVE_DESTINATION_DIR = new PropertyDescriptor.Builder()
+    public static final PropertyDescriptor MOVE_DESTINATION_DIR = new PropertyDescriptor.Builder()
         .name("Move Destination Directory")
         .description(String.format("The directory on the remote server to move the original file to once it has been ingested into NiFi. "
             + "This property is ignored unless the %s is set to '%s'. The specified directory must already exist on "
@@ -118,7 +116,7 @@ public abstract class FetchFileTransfer extends AbstractProcessor {
         .required(false)
         .build();
 
-    static final PropertyDescriptor FILE_NOT_FOUND_LOG_LEVEL = new PropertyDescriptor.Builder()
+    public static final PropertyDescriptor FILE_NOT_FOUND_LOG_LEVEL = new PropertyDescriptor.Builder()
         .displayName("Log level when file not found")
         .name("fetchfiletransfer-notfound-loglevel")
         .description("Log level to use in case the file does not exist when the processor is triggered")
@@ -131,15 +129,15 @@ public abstract class FetchFileTransfer extends AbstractProcessor {
         .name("success")
         .description("All FlowFiles that are received are routed to success")
         .build();
-    static final Relationship REL_COMMS_FAILURE = new Relationship.Builder()
+    public static final Relationship REL_COMMS_FAILURE = new Relationship.Builder()
         .name("comms.failure")
         .description("Any FlowFile that could not be fetched from the remote server due to a communications failure will be transferred to this Relationship.")
         .build();
-    static final Relationship REL_NOT_FOUND = new Relationship.Builder()
+    public static final Relationship REL_NOT_FOUND = new Relationship.Builder()
         .name("not.found")
         .description("Any FlowFile for which we receive a 'Not Found' message from the remote server will be transferred to this Relationship.")
         .build();
-    static final Relationship REL_PERMISSION_DENIED = new Relationship.Builder()
+    public static final Relationship REL_PERMISSION_DENIED = new Relationship.Builder()
         .name("permission.denied")
         .description("Any FlowFile that could not be fetched from the remote server due to insufficient permissions will be transferred to this Relationship.")
         .build();

diff --git a/...fi/processors/standard/util/FileInfo.java → ...rocessor/util/file/transfer/FileInfo.java b/...fi/processors/standard/util/FileInfo.java → ...rocessor/util/file/transfer/FileInfo.java
@@ -14,7 +14,7 @@
  * See the License for the specific language governing permissions and
  * limitations under the License.
  */
-package org.apache.nifi.processors.standard.util;
+package org.apache.nifi.processor.util.file.transfer;
 
 import java.io.File;
 import java.io.Serializable;

diff --git a/...rocessors/standard/util/FileTransfer.java → ...ssor/util/file/transfer/FileTransfer.java b/...rocessors/standard/util/FileTransfer.java → ...ssor/util/file/transfer/FileTransfer.java
@@ -14,7 +14,7 @@
  * See the License for the specific language governing permissions and
  * limitations under the License.
  */
-package org.apache.nifi.processors.standard.util;
+package org.apache.nifi.processor.util.file.transfer;
 
 import java.io.Closeable;
 import java.io.File;

diff --git a/.../processors/standard/GetFileTransfer.java → ...r/util/file/transfer/GetFileTransfer.java b/.../processors/standard/GetFileTransfer.java → ...r/util/file/transfer/GetFileTransfer.java
@@ -14,7 +14,7 @@
  * See the License for the specific language governing permissions and
  * limitations under the License.
  */
-package org.apache.nifi.processors.standard;
+package org.apache.nifi.processor.util.file.transfer;
 
 import org.apache.nifi.annotation.lifecycle.OnScheduled;
 import org.apache.nifi.flowfile.FlowFile;
@@ -25,8 +25,6 @@
 import org.apache.nifi.processor.ProcessSession;
 import org.apache.nifi.processor.Relationship;
 import org.apache.nifi.processor.exception.FlowFileAccessException;
-import org.apache.nifi.processors.standard.util.FileInfo;
-import org.apache.nifi.processors.standard.util.FileTransfer;
 import org.apache.nifi.util.StopWatch;
 
 import java.io.File;
@@ -67,6 +65,7 @@ public abstract class GetFileTransfer extends AbstractProcessor {
     public static final String FILE_OWNER_ATTRIBUTE = "file.owner";
     public static final String FILE_GROUP_ATTRIBUTE = "file.group";
     public static final String FILE_PERMISSIONS_ATTRIBUTE = "file.permissions";
+    public static final String FILE_SIZE_ATTRIBUTE = "file.size";
     public static final String FILE_MODIFY_DATE_ATTR_FORMAT = "yyyy-MM-dd'T'HH:mm:ssZ";
     protected static final DateTimeFormatter DATE_TIME_FORMATTER = DateTimeFormatter.ofPattern(FILE_MODIFY_DATE_ATTR_FORMAT);
 

diff --git a/...processors/standard/ListFileTransfer.java → .../util/file/transfer/ListFileTransfer.java b/...processors/standard/ListFileTransfer.java → .../util/file/transfer/ListFileTransfer.java
@@ -15,7 +15,7 @@
  * limitations under the License.
  */
 
-package org.apache.nifi.processors.standard;
+package org.apache.nifi.processor.util.file.transfer;
 
 import org.apache.commons.io.IOUtils;
 import org.apache.nifi.components.PropertyDescriptor;
@@ -24,8 +24,6 @@
 import org.apache.nifi.processor.ProcessContext;
 import org.apache.nifi.processor.util.StandardValidators;
 import org.apache.nifi.processor.util.list.AbstractListProcessor;
-import org.apache.nifi.processors.standard.util.FileInfo;
-import org.apache.nifi.processors.standard.util.FileTransfer;
 import org.apache.nifi.serialization.record.RecordSchema;
 
 import java.io.IOException;
@@ -46,7 +44,7 @@ public abstract class ListFileTransfer extends AbstractListProcessor<FileInfo> {
         .required(true)
         .expressionLanguageSupported(ExpressionLanguageScope.ENVIRONMENT)
         .build();
-    static final PropertyDescriptor UNDEFAULTED_PORT = new PropertyDescriptor.Builder()
+    public static final PropertyDescriptor UNDEFAULTED_PORT = new PropertyDescriptor.Builder()
         .name("Port")
         .description("The port to connect to on the remote host to fetch the data from")
         .addValidator(StandardValidators.NON_EMPTY_VALIDATOR)
@@ -76,15 +74,15 @@ public abstract class ListFileTransfer extends AbstractListProcessor<FileInfo> {
     @Override
     protected Map<String, String> createAttributes(final FileInfo fileInfo, final ProcessContext context) {
         final Map<String, String> attributes = new HashMap<>();
-        final DateTimeFormatter dateTimeFormatter = DateTimeFormatter.ofPattern(ListFile.FILE_MODIFY_DATE_ATTR_FORMAT, Locale.US);
+        final DateTimeFormatter dateTimeFormatter = DateTimeFormatter.ofPattern(GetFileTransfer.FILE_MODIFY_DATE_ATTR_FORMAT, Locale.US);
         attributes.put(getProtocolName() + ".remote.host", context.getProperty(HOSTNAME).evaluateAttributeExpressions().getValue());
         attributes.put(getProtocolName() + ".remote.port", context.getProperty(UNDEFAULTED_PORT).evaluateAttributeExpressions().getValue());
         attributes.put(getProtocolName() + ".listing.user", context.getProperty(USERNAME).evaluateAttributeExpressions().getValue());
-        attributes.put(ListFile.FILE_LAST_MODIFY_TIME_ATTRIBUTE, dateTimeFormatter.format(Instant.ofEpochMilli(fileInfo.getLastModifiedTime()).atZone(ZoneId.systemDefault())));
-        attributes.put(ListFile.FILE_PERMISSIONS_ATTRIBUTE, fileInfo.getPermissions());
-        attributes.put(ListFile.FILE_OWNER_ATTRIBUTE, fileInfo.getOwner());
-        attributes.put(ListFile.FILE_GROUP_ATTRIBUTE, fileInfo.getGroup());
-        attributes.put(ListFile.FILE_SIZE_ATTRIBUTE, Long.toString(fileInfo.getSize()));
+        attributes.put(GetFileTransfer.FILE_LAST_MODIFY_TIME_ATTRIBUTE, dateTimeFormatter.format(Instant.ofEpochMilli(fileInfo.getLastModifiedTime()).atZone(ZoneId.systemDefault())));
+        attributes.put(GetFileTransfer.FILE_PERMISSIONS_ATTRIBUTE, fileInfo.getPermissions());
+        attributes.put(GetFileTransfer.FILE_OWNER_ATTRIBUTE, fileInfo.getOwner());
+        attributes.put(GetFileTransfer.FILE_GROUP_ATTRIBUTE, fileInfo.getGroup());
+        attributes.put(GetFileTransfer.FILE_SIZE_ATTRIBUTE, Long.toString(fileInfo.getSize()));
         attributes.put(CoreAttributes.FILENAME.key(), fileInfo.getFileName());
         final String fullPath = fileInfo.getFullPathFileName();
         if (fullPath != null) {

diff --git a/...ndard/util/PermissionDeniedException.java → ...e/transfer/PermissionDeniedException.java b/...ndard/util/PermissionDeniedException.java → ...e/transfer/PermissionDeniedException.java
@@ -15,7 +15,7 @@
  * limitations under the License.
  */
 
-package org.apache.nifi.processors.standard.util;
+package org.apache.nifi.processor.util.file.transfer;
 
 import java.io.IOException;
 

diff --git a/.../processors/standard/PutFileTransfer.java → ...r/util/file/transfer/PutFileTransfer.java b/.../processors/standard/PutFileTransfer.java → ...r/util/file/transfer/PutFileTransfer.java
@@ -14,7 +14,7 @@
  * See the License for the specific language governing permissions and
  * limitations under the License.
  */
-package org.apache.nifi.processors.standard;
+package org.apache.nifi.processor.util.file.transfer;
 
 import org.apache.nifi.flowfile.FlowFile;
 import org.apache.nifi.flowfile.attributes.CoreAttributes;
@@ -26,9 +26,6 @@
 import org.apache.nifi.processor.exception.FlowFileAccessException;
 import org.apache.nifi.processor.exception.ProcessException;
 import org.apache.nifi.processor.io.InputStreamCallback;
-import org.apache.nifi.processors.standard.util.FileInfo;
-import org.apache.nifi.processors.standard.util.FileTransfer;
-import org.apache.nifi.processors.standard.util.SFTPTransfer;
 import org.apache.nifi.util.StopWatch;
 import org.apache.nifi.util.StringUtils;
 
@@ -127,7 +124,7 @@ public void onTrigger(final ProcessContext context, final ProcessSession session
                         @Override
                         public void process(final InputStream in) throws IOException {
                             try (final InputStream bufferedIn = new BufferedInputStream(in)) {
-                                if (workingDirPath != null && context.getProperty(SFTPTransfer.CREATE_DIRECTORY).asBoolean()) {
+                                if (workingDirPath != null && context.getProperty(FileTransfer.CREATE_DIRECTORY).asBoolean()) {
                                     transfer.ensureDirectoryExists(flowFileToTransfer, new File(workingDirPath));
                                 }
 

diff --git a/...rocessors/standard/util/TestFileInfo.java → ...ssor/util/file/transfer/TestFileInfo.java b/...rocessors/standard/util/TestFileInfo.java → ...ssor/util/file/transfer/TestFileInfo.java
@@ -15,7 +15,7 @@
  * limitations under the License.
  */
 
-package org.apache.nifi.processors.standard.util;
+package org.apache.nifi.processor.util.file.transfer;
 
 import org.junit.jupiter.api.Test;
 

diff --git a/nifi-nar-bundles/nifi-extension-utils/pom.xml b/nifi-nar-bundles/nifi-extension-utils/pom.xml
@@ -34,6 +34,7 @@
         <module>nifi-event-listen</module>
         <module>nifi-event-put</module>
         <module>nifi-event-transport</module>
+        <module>nifi-file-transfer</module>
         <module>nifi-hadoop-utils</module>
         <module>nifi-kerberos-test-utils</module>
         <module>nifi-listed-entity</module>

diff --git a/nifi-nar-bundles/nifi-standard-bundle/nifi-standard-processors/pom.xml b/nifi-nar-bundles/nifi-standard-bundle/nifi-standard-processors/pom.xml
@@ -67,6 +67,11 @@
             <artifactId>nifi-event-transport</artifactId>
             <version>2.0.0-SNAPSHOT</version>
         </dependency>
+        <dependency>
+            <groupId>org.apache.nifi</groupId>
+            <artifactId>nifi-file-transfer</artifactId>
+            <version>2.0.0-SNAPSHOT</version>
+        </dependency>
         <dependency>
             <groupId>org.apache.nifi</groupId>
             <artifactId>nifi-syslog-utils</artifactId>

diff --git a/.../nifi-standard-processors/src/main/java/org/apache/nifi/processors/standard/FetchFTP.java b/.../nifi-standard-processors/src/main/java/org/apache/nifi/processors/standard/FetchFTP.java
@@ -33,8 +33,9 @@
 import org.apache.nifi.components.ValidationContext;
 import org.apache.nifi.components.ValidationResult;
 import org.apache.nifi.processor.ProcessContext;
+import org.apache.nifi.processor.util.file.transfer.FetchFileTransfer;
 import org.apache.nifi.processors.standard.util.FTPTransfer;
-import org.apache.nifi.processors.standard.util.FileTransfer;
+import org.apache.nifi.processor.util.file.transfer.FileTransfer;
 
 // Note that we do not use @SupportsBatching annotation. This processor cannot support batching because it must ensure that session commits happen before remote files are deleted.
 

diff --git a/...nifi-standard-processors/src/main/java/org/apache/nifi/processors/standard/FetchSFTP.java b/...nifi-standard-processors/src/main/java/org/apache/nifi/processors/standard/FetchSFTP.java
@@ -33,8 +33,9 @@
 import org.apache.nifi.components.ValidationContext;
 import org.apache.nifi.components.ValidationResult;
 import org.apache.nifi.processor.ProcessContext;
+import org.apache.nifi.processor.util.file.transfer.FetchFileTransfer;
 import org.apache.nifi.processors.standard.util.FTPTransfer;
-import org.apache.nifi.processors.standard.util.FileTransfer;
+import org.apache.nifi.processor.util.file.transfer.FileTransfer;
 import org.apache.nifi.processors.standard.util.SFTPTransfer;
 
 // Note that we do not use @SupportsBatching annotation. This processor cannot support batching because it must ensure that session commits happen before remote files are deleted.

diff --git a/...le/nifi-standard-processors/src/main/java/org/apache/nifi/processors/standard/GetFTP.java b/...le/nifi-standard-processors/src/main/java/org/apache/nifi/processors/standard/GetFTP.java
@@ -33,8 +33,9 @@
 import org.apache.nifi.components.ValidationResult;
 import org.apache.nifi.processor.ProcessContext;
 import org.apache.nifi.processor.ProcessorInitializationContext;
+import org.apache.nifi.processor.util.file.transfer.GetFileTransfer;
 import org.apache.nifi.processors.standard.util.FTPTransfer;
-import org.apache.nifi.processors.standard.util.FileTransfer;
+import org.apache.nifi.processor.util.file.transfer.FileTransfer;
 
 @InputRequirement(Requirement.INPUT_FORBIDDEN)
 @Tags({"FTP", "get", "retrieve", "files", "fetch", "remote", "ingest", "source", "input"})

diff --git a/...e/nifi-standard-processors/src/main/java/org/apache/nifi/processors/standard/GetSFTP.java b/...e/nifi-standard-processors/src/main/java/org/apache/nifi/processors/standard/GetSFTP.java
@@ -33,8 +33,9 @@
 import org.apache.nifi.components.ValidationResult;
 import org.apache.nifi.processor.ProcessContext;
 import org.apache.nifi.processor.ProcessorInitializationContext;
+import org.apache.nifi.processor.util.file.transfer.GetFileTransfer;
 import org.apache.nifi.processors.standard.util.FTPTransfer;
-import org.apache.nifi.processors.standard.util.FileTransfer;
+import org.apache.nifi.processor.util.file.transfer.FileTransfer;
 import org.apache.nifi.processors.standard.util.SFTPTransfer;
 
 @InputRequirement(Requirement.INPUT_FORBIDDEN)