This is an automated email from the ASF dual-hosted git repository.
pvillard 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 2493d3cef9a NIFI-15345 Avoided calling SFTP stat when attributes not
configured
2493d3cef9a is described below
commit 2493d3cef9a525394329d2088904fa652b07894e
Author: exceptionfactory <[email protected]>
AuthorDate: Mon Dec 15 13:00:29 2025 -0600
NIFI-15345 Avoided calling SFTP stat when attributes not configured
- Refactored SFTPTransfer.put() to avoid getting remote file attributes
when the PutSFTP Processor is not configured to change attributes
Signed-off-by: Pierre Villard <[email protected]>
This closes #10646.
---
.../processors/standard/util/SFTPTransfer.java | 122 +++++++++++++++------
.../nifi/processors/standard/TestPutSFTP.java | 32 +++++-
2 files changed, 121 insertions(+), 33 deletions(-)
diff --git
a/nifi-extension-bundles/nifi-standard-bundle/nifi-standard-processors/src/main/java/org/apache/nifi/processors/standard/util/SFTPTransfer.java
b/nifi-extension-bundles/nifi-standard-bundle/nifi-standard-processors/src/main/java/org/apache/nifi/processors/standard/util/SFTPTransfer.java
index 45507944281..99d526e8fd6 100644
---
a/nifi-extension-bundles/nifi-standard-bundle/nifi-standard-processors/src/main/java/org/apache/nifi/processors/standard/util/SFTPTransfer.java
+++
b/nifi-extension-bundles/nifi-standard-bundle/nifi-standard-processors/src/main/java/org/apache/nifi/processors/standard/util/SFTPTransfer.java
@@ -747,44 +747,13 @@ public class SFTPTransfer implements FileTransfer {
}
final String tempPath = buildFullPath(path, tempFilename);
- final SftpClient.Attributes attributes;
try {
sftpClient.put(content, tempPath);
- attributes = sftpClient.stat(tempPath);
} catch (final SftpException e) {
throw new IOException("Failed to transfer content to
[%s]".formatted(fullPath), e);
}
- final String permissions =
ctx.getProperty(PERMISSIONS).evaluateAttributeExpressions(flowFile).getValue();
- if (StringUtils.isNotEmpty(permissions)) {
- final int perms = numberPermissions(permissions);
- attributes.setPermissions(perms);
- }
-
- final String lastModifiedTime =
ctx.getProperty(LAST_MODIFIED_TIME).evaluateAttributeExpressions(flowFile).getValue();
- if (lastModifiedTime != null && !lastModifiedTime.isBlank()) {
- final DateTimeFormatter dateTimeFormatter =
DateTimeFormatter.ofPattern(FILE_MODIFY_DATE_ATTR_FORMAT, Locale.US);
- final OffsetDateTime offsetDateTime =
OffsetDateTime.parse(lastModifiedTime, dateTimeFormatter);
- final FileTime modifyTime =
FileTime.from(offsetDateTime.toInstant());
- attributes.setModifyTime(modifyTime);
- }
-
- final String owner =
ctx.getProperty(REMOTE_OWNER).evaluateAttributeExpressions(flowFile).getValue();
- if (StringUtils.isNotEmpty(owner)) {
- attributes.setOwner(owner);
- }
-
- final String group =
ctx.getProperty(REMOTE_GROUP).evaluateAttributeExpressions(flowFile).getValue();
- if (StringUtils.isNotEmpty(group)) {
- attributes.setGroup(group);
- }
-
- // Set Attributes on temporary file path to avoid potential timing
issues with retrieval and removal of transferred files
- try {
- sftpClient.setStat(tempPath, attributes);
- } catch (final SftpException e) {
- logger.warn("Failed to set attributes on Remote File [{}]",
tempPath, e);
- }
+ setAttributes(flowFile, tempPath);
if (!filename.equals(tempFilename)) {
try {
@@ -809,6 +778,44 @@ public class SFTPTransfer implements FileTransfer {
return fullPath;
}
+ private void setAttributes(final FlowFile flowFile, final String
remotePath) {
+ final AttributesRequested attributesRequested = new
AttributesRequested();
+
+ final String permissions =
ctx.getProperty(PERMISSIONS).evaluateAttributeExpressions(flowFile).getValue();
+ if (StringUtils.isNotEmpty(permissions)) {
+ final int perms = numberPermissions(permissions);
+ attributesRequested.setPermissions(perms);
+ }
+
+ final String lastModifiedTime =
ctx.getProperty(LAST_MODIFIED_TIME).evaluateAttributeExpressions(flowFile).getValue();
+ if (StringUtils.isNotBlank(lastModifiedTime)) {
+ final DateTimeFormatter dateTimeFormatter =
DateTimeFormatter.ofPattern(FILE_MODIFY_DATE_ATTR_FORMAT, Locale.US);
+ final OffsetDateTime offsetDateTime =
OffsetDateTime.parse(lastModifiedTime, dateTimeFormatter);
+ final FileTime modifyTime =
FileTime.from(offsetDateTime.toInstant());
+ attributesRequested.setModifyTime(modifyTime);
+ }
+
+ final String owner =
ctx.getProperty(REMOTE_OWNER).evaluateAttributeExpressions(flowFile).getValue();
+ if (StringUtils.isNotEmpty(owner)) {
+ attributesRequested.setOwner(owner);
+ }
+
+ final String group =
ctx.getProperty(REMOTE_GROUP).evaluateAttributeExpressions(flowFile).getValue();
+ if (StringUtils.isNotEmpty(group)) {
+ attributesRequested.setGroup(group);
+ }
+
+ if (attributesRequested.isConfigured()) {
+ try {
+ final SftpClient.Attributes attributes =
sftpClient.stat(remotePath);
+ attributesRequested.setAttributes(attributes);
+ sftpClient.setStat(remotePath, attributes);
+ } catch (final IOException e) {
+ logger.warn("Failed to set attributes on Remote File [{}] for
{}", remotePath, flowFile, e);
+ }
+ }
+ }
+
@Override
public void rename(final FlowFile flowFile, final String source, final
String target) throws IOException {
final SftpClient sftpClient = getSFTPClient(flowFile);
@@ -868,4 +875,55 @@ public class SFTPTransfer implements FileTransfer {
}
return number;
}
+
+ private static class AttributesRequested {
+ private boolean configured;
+
+ private Integer permissions;
+
+ private FileTime modifyTime;
+
+ private String owner;
+
+ private String group;
+
+ void setPermissions(final int permissions) {
+ this.permissions = permissions;
+ this.configured = true;
+ }
+
+ void setModifyTime(final FileTime modifyTime) {
+ this.modifyTime = modifyTime;
+ this.configured = true;
+ }
+
+ void setOwner(final String owner) {
+ this.owner = owner;
+ this.configured = true;
+ }
+
+ void setGroup(final String group) {
+ this.group = group;
+ this.configured = true;
+ }
+
+ boolean isConfigured() {
+ return configured;
+ }
+
+ void setAttributes(final SftpClient.Attributes attributes) {
+ if (permissions != null) {
+ attributes.setPermissions(permissions);
+ }
+ if (modifyTime != null) {
+ attributes.setModifyTime(modifyTime);
+ }
+ if (StringUtils.isNotBlank(owner)) {
+ attributes.setOwner(owner);
+ }
+ if (StringUtils.isNotBlank(group)) {
+ attributes.setGroup(group);
+ }
+ }
+ }
}
diff --git
a/nifi-extension-bundles/nifi-standard-bundle/nifi-standard-processors/src/test/java/org/apache/nifi/processors/standard/TestPutSFTP.java
b/nifi-extension-bundles/nifi-standard-bundle/nifi-standard-processors/src/test/java/org/apache/nifi/processors/standard/TestPutSFTP.java
index 797b645a543..b2671a213f6 100644
---
a/nifi-extension-bundles/nifi-standard-bundle/nifi-standard-processors/src/test/java/org/apache/nifi/processors/standard/TestPutSFTP.java
+++
b/nifi-extension-bundles/nifi-standard-bundle/nifi-standard-processors/src/test/java/org/apache/nifi/processors/standard/TestPutSFTP.java
@@ -31,9 +31,15 @@ import org.junit.jupiter.api.io.TempDir;
import java.io.IOException;
import java.nio.file.Files;
import java.nio.file.Path;
+import java.nio.file.attribute.FileTime;
+import java.time.Instant;
+import java.time.OffsetDateTime;
+import java.time.ZoneId;
+import java.time.format.DateTimeFormatter;
import java.util.Collections;
import java.util.List;
+import static org.junit.jupiter.api.Assertions.assertEquals;
import static org.junit.jupiter.api.Assertions.assertTrue;
import static org.junit.jupiter.api.Assertions.assertFalse;
@@ -50,14 +56,19 @@ class TestPutSFTP {
private static final String TRANSIT_URI_FORMAT = "sftp://%s";;
+ private static final OffsetDateTime LAST_MODIFIED_TIME_CONFIGURED =
OffsetDateTime.ofInstant(Instant.ofEpochSecond(30), ZoneId.systemDefault());
+
private SSHTestServer sshTestServer;
private TestRunner runner;
+ private Path serverRootPath;
+
@BeforeEach
void setRunner(@TempDir final Path rootPath) throws IOException {
sshTestServer = new SSHTestServer(rootPath);
sshTestServer.startServer();
+ serverRootPath = rootPath;
runner = TestRunners.newTestRunner(PutSFTP.class);
runner.setProperty(SFTPTransfer.HOSTNAME, sshTestServer.getHost());
@@ -192,8 +203,27 @@ class TestPutSFTP {
final List<ProvenanceEventRecord> records =
runner.getProvenanceEvents();
assertFalse(records.isEmpty());
- final ProvenanceEventRecord record = records.iterator().next();
+ final ProvenanceEventRecord record = records.getFirst();
final String firstTransitUri = String.format(TRANSIT_URI_FORMAT,
sshTestServer.getHost());
assertTrue(record.getTransitUri().startsWith(firstTransitUri),
"Transit URI not found");
}
+
+ @Test
+ void testRunSetLastModifiedTime() throws IOException {
+ final DateTimeFormatter formatter =
DateTimeFormatter.ofPattern(SFTPTransfer.FILE_MODIFY_DATE_ATTR_FORMAT);
+ final String lastModifiedTimeConfigured =
formatter.format(LAST_MODIFIED_TIME_CONFIGURED);
+
+ runner.setProperty(SFTPTransfer.LAST_MODIFIED_TIME,
lastModifiedTimeConfigured);
+ runner.enqueue(FLOW_FILE_CONTENTS,
Collections.singletonMap(CoreAttributes.FILENAME.key(), FIRST_FILENAME));
+ runner.run();
+
+ runner.assertAllFlowFilesTransferred(PutSFTP.REL_SUCCESS);
+
+ final Path serverPath =
serverRootPath.resolve(REMOTE_DIRECTORY).resolve(FIRST_FILENAME);
+ assertTrue(Files.exists(serverPath), "Server file not found
[%s]".formatted(serverPath));
+
+ final FileTime lastModifiedTime =
Files.getLastModifiedTime(serverPath);
+ final FileTime expectedLastModifiedTime =
FileTime.from(LAST_MODIFIED_TIME_CONFIGURED.toInstant());
+ assertEquals(expectedLastModifiedTime, lastModifiedTime);
+ }
}
