This is an automated email from the ASF dual-hosted git repository.
exceptionfactory pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/nifi.git
The following commit(s) were added to refs/heads/main by this push:
new 429f9e7348b NIFI-15683 Added Move Confliction Resolution to FetchSFTP
(#11281)
429f9e7348b is described below
commit 429f9e7348bd7bdb10966a7ed4c389a63eae490b
Author: Rakesh Kumar Singh <[email protected]>
AuthorDate: Thu Aug 6 19:52:05 2026 +0530
NIFI-15683 Added Move Confliction Resolution to FetchSFTP (#11281)
Co-authored-by: David Handermann <[email protected]>
Signed-off-by: David Handermann <[email protected]>
---
.../util/file/transfer/FetchFileTransfer.java | 56 +++++++++-
.../apache/nifi/processors/standard/FetchSFTP.java | 1 +
.../nifi/processors/standard/FetchSFTPTest.java | 115 +++++++++++++++++++++
3 files changed, 170 insertions(+), 2 deletions(-)
diff --git
a/nifi-extension-bundles/nifi-extension-utils/nifi-file-transfer/src/main/java/org/apache/nifi/processor/util/file/transfer/FetchFileTransfer.java
b/nifi-extension-bundles/nifi-extension-utils/nifi-file-transfer/src/main/java/org/apache/nifi/processor/util/file/transfer/FetchFileTransfer.java
index 36dfac1fe85..ec429cf9503 100644
---
a/nifi-extension-bundles/nifi-extension-utils/nifi-file-transfer/src/main/java/org/apache/nifi/processor/util/file/transfer/FetchFileTransfer.java
+++
b/nifi-extension-bundles/nifi-extension-utils/nifi-file-transfer/src/main/java/org/apache/nifi/processor/util/file/transfer/FetchFileTransfer.java
@@ -21,6 +21,7 @@ import org.apache.commons.lang3.StringUtils;
import org.apache.nifi.annotation.lifecycle.OnScheduled;
import org.apache.nifi.annotation.lifecycle.OnStopped;
import org.apache.nifi.components.AllowableValue;
+import org.apache.nifi.components.DescribedValue;
import org.apache.nifi.components.PropertyDescriptor;
import org.apache.nifi.expression.ExpressionLanguageScope;
import org.apache.nifi.flowfile.FlowFile;
@@ -55,6 +56,37 @@ import java.util.concurrent.TimeUnit;
*/
public abstract class FetchFileTransfer extends AbstractProcessor {
+ public enum MoveConflictResolution implements DescribedValue {
+ FAIL("Fail", "Leave the source file unchanged and log failure status"),
+ REPLACE("Replace", """
+ When the destination file already exists, delete it and rename
the source file to the destination.
+ This option supports SFTP servers such as OpenSSH that do not
implement atomic overwrite on rename.
+ """);
+
+ MoveConflictResolution(final String displayName, final String
description) {
+ this.displayName = displayName;
+ this.description = description;
+ }
+
+ private final String displayName;
+ private final String description;
+
+ @Override
+ public String getValue() {
+ return name();
+ }
+
+ @Override
+ public String getDisplayName() {
+ return displayName;
+ }
+
+ @Override
+ public String getDescription() {
+ return description;
+ }
+ }
+
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");
@@ -91,8 +123,7 @@ public abstract class FetchFileTransfer extends
AbstractProcessor {
.build();
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.")
+ .description("Specifies what to do with the original file on the
server once it has been fetched")
.expressionLanguageSupported(ExpressionLanguageScope.NONE)
.allowableValues(COMPLETION_NONE, COMPLETION_MOVE, COMPLETION_DELETE)
.defaultValue(COMPLETION_NONE.getValue())
@@ -116,6 +147,18 @@ public abstract class FetchFileTransfer extends
AbstractProcessor {
.required(false)
.build();
+ public static final PropertyDescriptor MOVE_CONFLICT_RESOLUTION = new
PropertyDescriptor.Builder()
+ .name("Move Conflict Resolution")
+ .description("""
+ Specifies how to handle the case when the destination file
already exists when Completion Strategy is set to Move files
+ """)
+ .expressionLanguageSupported(ExpressionLanguageScope.NONE)
+ .allowableValues(MoveConflictResolution.class)
+ .defaultValue(MoveConflictResolution.FAIL)
+ .dependsOn(COMPLETION_STRATEGY, COMPLETION_MOVE)
+ .required(true)
+ .build();
+
public static final PropertyDescriptor FILE_NOT_FOUND_LOG_LEVEL = new
PropertyDescriptor.Builder()
.name("Log Level When File Not Found")
.description("Log level to use in case the file does not exist when
the processor is triggered")
@@ -354,6 +397,7 @@ public abstract class FetchFileTransfer extends
AbstractProcessor {
} else if
(COMPLETION_MOVE.getValue().equalsIgnoreCase(completionStrategy)) {
final String targetDir =
context.getProperty(MOVE_DESTINATION_DIR).evaluateAttributeExpressions(flowFile).getValue();
final String simpleFilename =
StringUtils.substringAfterLast(filename, "/");
+ final MoveConflictResolution moveConflictResolution =
context.getProperty(MOVE_CONFLICT_RESOLUTION).asAllowableValue(MoveConflictResolution.class);
try {
final String absoluteTargetDirPath =
transfer.getAbsolutePath(flowFile, targetDir);
@@ -363,6 +407,14 @@ public abstract class FetchFileTransfer extends
AbstractProcessor {
}
final String destinationPath = String.format("%s/%s",
absoluteTargetDirPath, simpleFilename);
+ if (MoveConflictResolution.REPLACE == moveConflictResolution) {
+ final FileInfo existingDestination =
transfer.getRemoteFileInfo(flowFile, absoluteTargetDirPath, simpleFilename);
+ if (existingDestination != null) {
+ getLogger().debug("Destination [{}] already exists:
removing before rename based on conflict resolution strategy", destinationPath);
+ transfer.deleteFile(flowFile, null, destinationPath);
+ }
+ }
+
transfer.rename(flowFile, filename, destinationPath);
} catch (final IOException ioe) {
diff --git
a/nifi-extension-bundles/nifi-standard-bundle/nifi-standard-processors/src/main/java/org/apache/nifi/processors/standard/FetchSFTP.java
b/nifi-extension-bundles/nifi-standard-bundle/nifi-standard-processors/src/main/java/org/apache/nifi/processors/standard/FetchSFTP.java
index 1e21adef703..ae37cdb123c 100644
---
a/nifi-extension-bundles/nifi-standard-bundle/nifi-standard-processors/src/main/java/org/apache/nifi/processors/standard/FetchSFTP.java
+++
b/nifi-extension-bundles/nifi-standard-bundle/nifi-standard-processors/src/main/java/org/apache/nifi/processors/standard/FetchSFTP.java
@@ -108,6 +108,7 @@ public class FetchSFTP extends FetchFileTransfer {
COMPLETION_STRATEGY,
MOVE_DESTINATION_DIR,
MOVE_CREATE_DIRECTORY,
+ MOVE_CONFLICT_RESOLUTION,
DISABLE_DIRECTORY_LISTING,
SFTPTransfer.CONNECTION_TIMEOUT,
SFTPTransfer.DATA_TIMEOUT,
diff --git
a/nifi-extension-bundles/nifi-standard-bundle/nifi-standard-processors/src/test/java/org/apache/nifi/processors/standard/FetchSFTPTest.java
b/nifi-extension-bundles/nifi-standard-bundle/nifi-standard-processors/src/test/java/org/apache/nifi/processors/standard/FetchSFTPTest.java
new file mode 100644
index 00000000000..1795ea410be
--- /dev/null
+++
b/nifi-extension-bundles/nifi-standard-bundle/nifi-standard-processors/src/test/java/org/apache/nifi/processors/standard/FetchSFTPTest.java
@@ -0,0 +1,115 @@
+/*
+ * 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.nifi.processors.standard;
+
+import org.apache.nifi.processor.util.file.transfer.FetchFileTransfer;
+import org.apache.nifi.processors.standard.util.SFTPTransfer;
+import org.apache.nifi.processors.standard.util.SSHTestServer;
+import org.apache.nifi.util.TestRunner;
+import org.apache.nifi.util.TestRunners;
+import org.junit.jupiter.api.AfterEach;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.io.TempDir;
+
+import java.io.IOException;
+import java.nio.charset.StandardCharsets;
+import java.nio.file.Files;
+import java.nio.file.Path;
+
+import static org.junit.jupiter.api.Assertions.assertFalse;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
+class FetchSFTPTest {
+
+ private static final String SOURCE_FILENAME = "test.txt";
+ private static final String SOURCE_CONTENTS = "source content";
+ private static final String DESTINATION_DIR = "/completed";
+
+ private SSHTestServer sshTestServer;
+ private TestRunner runner;
+ private Path serverRootPath;
+
+ @BeforeEach
+ void setup(@TempDir final Path rootPath) throws IOException {
+ sshTestServer = new SSHTestServer(rootPath);
+ sshTestServer.startServer();
+ serverRootPath = rootPath;
+
+ runner = TestRunners.newTestRunner(FetchSFTP.class);
+ runner.setProperty(FetchFileTransfer.HOSTNAME,
sshTestServer.getHost());
+ runner.setProperty(FetchFileTransfer.UNDEFAULTED_PORT,
Integer.toString(sshTestServer.getSSHPort()));
+ runner.setProperty(SFTPTransfer.USERNAME, sshTestServer.getUsername());
+ runner.setProperty(SFTPTransfer.PASSWORD, sshTestServer.getPassword());
+ runner.setProperty(SFTPTransfer.USE_KEEPALIVE_ON_TIMEOUT,
Boolean.FALSE.toString());
+ runner.setProperty(SFTPTransfer.STRICT_HOST_KEY_CHECKING,
Boolean.FALSE.toString());
+ runner.setProperty(SFTPTransfer.DATA_TIMEOUT, "30 sec");
+ }
+
+ @AfterEach
+ void stopServer() throws IOException {
+ sshTestServer.stopServer();
+ }
+
+ @Test
+ void testMoveCompletionStrategyCleansUpSourceFile() throws IOException {
+ final Path sourceDir =
Files.createDirectory(serverRootPath.resolve("source"));
+ final Path sourceFile = sourceDir.resolve(SOURCE_FILENAME);
+ Files.writeString(sourceFile, SOURCE_CONTENTS, StandardCharsets.UTF_8);
+
+ final Path completedDir =
Files.createDirectory(serverRootPath.resolve("completed"));
+
+ runner.setProperty(FetchFileTransfer.REMOTE_FILENAME, "/source/" +
SOURCE_FILENAME);
+ runner.setProperty(FetchFileTransfer.COMPLETION_STRATEGY,
FetchFileTransfer.COMPLETION_MOVE.getValue());
+ runner.setProperty(FetchFileTransfer.MOVE_DESTINATION_DIR,
DESTINATION_DIR);
+ runner.setProperty(FetchFileTransfer.MOVE_CREATE_DIRECTORY,
Boolean.FALSE.toString());
+
+ runner.enqueue(new byte[0]);
+ runner.run();
+
+ runner.assertTransferCount(FetchFileTransfer.REL_SUCCESS, 1);
+ assertFalse(Files.exists(sourceFile), "Source file should have been
moved");
+ assertTrue(Files.exists(completedDir.resolve(SOURCE_FILENAME)), "File
should exist in destination directory");
+ }
+
+ @Test
+ void testMoveCompletionStrategyOverwritesExistingDestinationFile() throws
IOException {
+ // When conflict resolution is set to Replace, FetchSFTP should
proactively delete the destination
+ // before renaming. This mirrors the behaviour needed for SFTP servers
(e.g. OpenSSH) that return
+ // SSH_FX_FAILURE on rename when the destination already exists.
+ final Path sourceDir =
Files.createDirectory(serverRootPath.resolve("source"));
+ final Path sourceFile = sourceDir.resolve(SOURCE_FILENAME);
+ Files.writeString(sourceFile, SOURCE_CONTENTS, StandardCharsets.UTF_8);
+
+ final Path completedDir =
Files.createDirectory(serverRootPath.resolve("completed"));
+ final Path existingDestination = completedDir.resolve(SOURCE_FILENAME);
+ Files.writeString(existingDestination, "old content",
StandardCharsets.UTF_8);
+
+ runner.setProperty(FetchFileTransfer.REMOTE_FILENAME, "/source/" +
SOURCE_FILENAME);
+ runner.setProperty(FetchFileTransfer.COMPLETION_STRATEGY,
FetchFileTransfer.COMPLETION_MOVE.getValue());
+ runner.setProperty(FetchFileTransfer.MOVE_DESTINATION_DIR,
DESTINATION_DIR);
+ runner.setProperty(FetchFileTransfer.MOVE_CREATE_DIRECTORY,
Boolean.FALSE.toString());
+ runner.setProperty(FetchFileTransfer.MOVE_CONFLICT_RESOLUTION,
FetchFileTransfer.MoveConflictResolution.REPLACE.getValue());
+
+ runner.enqueue(new byte[0]);
+ runner.run();
+
+ runner.assertTransferCount(FetchFileTransfer.REL_SUCCESS, 1);
+ assertFalse(Files.exists(sourceFile), "Source file should have been
moved");
+ assertTrue(Files.exists(existingDestination), "Destination file should
exist after move");
+ }
+}
