You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@nifi.apache.org by jo...@apache.org on 2022/05/13 14:47:11 UTC
[nifi] branch support/nifi-1.16 updated (b883b9c356 -> 98d5c22efe)
This is an automated email from the ASF dual-hosted git repository.
joewitt pushed a change to branch support/nifi-1.16
in repository https://gitbox.apache.org/repos/asf/nifi.git
from b883b9c356 NIFI-10026 This closes #6043. Added jakarta.activation runtime dependency to bootstrap
new 462af16c81 NIFI-9470 This closes #6027. Allow creation of Parameter Context without any Inherited Parameter Contexts
new aa7a06e33d NIFI-9966 Corrects the registry loading of large flowfiles from git
new 98d5c22efe NIFI-9990 This closes #6030. Improved FTP 550 file unavailable handling
The 3 revisions listed above as "new" are entirely new to this
repository and will be described in separate emails. The revisions
listed as "add" were already present in the repository and have only
been added to this reference.
Summary of changes:
.../web/dao/impl/StandardParameterContextDAO.java | 6 +--
.../nifi/processors/standard/util/FTPTransfer.java | 26 +++++++---
.../apache/nifi/processors/standard/TestFTP.java | 58 ++++++++++++++++++----
.../provider/flow/git/GitFlowMetaData.java | 5 +-
.../flow/git/TestGitFlowPersistenceProvider.java | 29 +++++++++++
5 files changed, 102 insertions(+), 22 deletions(-)
[nifi] 03/03: NIFI-9990 This closes #6030. Improved FTP 550 file unavailable handling
Posted by jo...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
joewitt pushed a commit to branch support/nifi-1.16
in repository https://gitbox.apache.org/repos/asf/nifi.git
commit 98d5c22efecc9cd732d4df21b782c40b3a4efa7e
Author: exceptionfactory <ex...@apache.org>
AuthorDate: Tue May 10 10:08:14 2022 -0500
NIFI-9990 This closes #6030. Improved FTP 550 file unavailable handling
- Improved File Not Found reply detection
- Added Permission Denied reply handling
Signed-off-by: Joe Witt <jo...@apache.org>
---
.../nifi/processors/standard/util/FTPTransfer.java | 26 +++++++---
.../apache/nifi/processors/standard/TestFTP.java | 58 ++++++++++++++++++----
2 files changed, 66 insertions(+), 18 deletions(-)
diff --git a/nifi-nar-bundles/nifi-standard-bundle/nifi-standard-processors/src/main/java/org/apache/nifi/processors/standard/util/FTPTransfer.java b/nifi-nar-bundles/nifi-standard-bundle/nifi-standard-processors/src/main/java/org/apache/nifi/processors/standard/util/FTPTransfer.java
index 7eac3b3564..2d3e7adb2f 100644
--- a/nifi-nar-bundles/nifi-standard-bundle/nifi-standard-processors/src/main/java/org/apache/nifi/processors/standard/util/FTPTransfer.java
+++ b/nifi-nar-bundles/nifi-standard-bundle/nifi-standard-processors/src/main/java/org/apache/nifi/processors/standard/util/FTPTransfer.java
@@ -135,6 +135,10 @@ public class FTPTransfer implements FileTransfer {
.addValidator(StandardValidators.BOOLEAN_VALIDATOR)
.build();
+ private static final int REPLY_CODE_FILE_UNAVAILABLE = 550;
+
+ private static final Pattern NOT_FOUND_MESSAGE_PATTERN = Pattern.compile("(no such)|(not exist)|(not found)", Pattern.CASE_INSENSITIVE);
+
private static final FTPClientProvider FTP_CLIENT_PROVIDER = new StandardFTPClientProvider();
private static final ProxySpec[] PROXY_SPECS = {ProxySpec.HTTP_AUTH, ProxySpec.SOCKS_AUTH};
@@ -321,13 +325,22 @@ public class FTPTransfer implements FileTransfer {
FlowFile resultFlowFile;
try (InputStream in = client.retrieveFileStream(remoteFileName)) {
if (in == null) {
- final String response = client.getReplyString();
- // FTPClient doesn't throw exception if file not found.
- // Instead, response string will contain: "550 Can't open <absolute_path>: No such file or directory"
- if (response != null && response.trim().endsWith("No such file or directory")) {
- throw new FileNotFoundException(response);
+ final String reply = client.getReplyString();
+ if (reply == null) {
+ throw new IOException("Retrieve File Failed: FTP server response not found");
}
- throw new IOException(response);
+
+ // Get reply code after checking for reply string
+ final int replyCode = client.getReplyCode();
+ if (REPLY_CODE_FILE_UNAVAILABLE == replyCode) {
+ if (NOT_FOUND_MESSAGE_PATTERN.matcher(reply).find()) {
+ throw new FileNotFoundException(reply);
+ } else {
+ throw new PermissionDeniedException(reply);
+ }
+ }
+
+ throw new IOException(reply);
}
resultFlowFile = session.write(origFlowFile, out -> StreamUtils.copy(in, out));
client.completePendingCommand();
@@ -606,5 +619,4 @@ public class FTPTransfer implements FileTransfer {
return componentProxyConfig;
};
}
-
}
diff --git a/nifi-nar-bundles/nifi-standard-bundle/nifi-standard-processors/src/test/java/org/apache/nifi/processors/standard/TestFTP.java b/nifi-nar-bundles/nifi-standard-bundle/nifi-standard-processors/src/test/java/org/apache/nifi/processors/standard/TestFTP.java
index 1993aad6ea..418b1528d6 100644
--- a/nifi-nar-bundles/nifi-standard-bundle/nifi-standard-processors/src/test/java/org/apache/nifi/processors/standard/TestFTP.java
+++ b/nifi-nar-bundles/nifi-standard-bundle/nifi-standard-processors/src/test/java/org/apache/nifi/processors/standard/TestFTP.java
@@ -16,7 +16,7 @@
*/
package org.apache.nifi.processors.standard;
-import static org.junit.Assert.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertEquals;
import java.io.FileInputStream;
import java.io.IOException;
@@ -45,6 +45,7 @@ import org.mockftpserver.fake.UserAccount;
import org.mockftpserver.fake.filesystem.DirectoryEntry;
import org.mockftpserver.fake.filesystem.FileEntry;
import org.mockftpserver.fake.filesystem.FileSystem;
+import org.mockftpserver.fake.filesystem.Permissions;
import org.mockftpserver.fake.filesystem.WindowsFakeFileSystem;
public class TestFTP {
@@ -80,7 +81,6 @@ public class TestFTP {
ProcessContext pc;
/* Set the basic required values */
- results = new HashSet<>();
runner.setProperty(FTPTransfer.USERNAME, "${el-username}");
runner.setProperty(FTPTransfer.HOSTNAME, "static-hostname");
runner.setProperty(FTPTransfer.PORT, "${el-portNumber}");
@@ -132,8 +132,8 @@ public class TestFTP {
runner.setProperty(FTPTransfer.USERNAME, username);
runner.setProperty(FTPTransfer.PASSWORD, password);
runner.setProperty(FTPTransfer.PORT, Integer.toString(ftpPort));
- try (FileInputStream fis = new FileInputStream("src/test/resources/randombytes-1");) {
- Map<String, String> attributes = new HashMap<String, String>();
+ try (FileInputStream fis = new FileInputStream("src/test/resources/randombytes-1")) {
+ Map<String, String> attributes = new HashMap<>();
attributes.put(CoreAttributes.FILENAME.key(), "randombytes-1");
runner.enqueue(fis, attributes);
runner.run();
@@ -157,14 +157,14 @@ public class TestFTP {
// Get two flowfiles to test by running data
try (FileInputStream fis = new FileInputStream("src/test/resources/randombytes-1")) {
- Map<String, String> attributes = new HashMap<String, String>();
+ Map<String, String> attributes = new HashMap<>();
attributes.put(CoreAttributes.FILENAME.key(), "randombytes-1");
attributes.put("transfer-host", "localhost");
runner.enqueue(fis, attributes);
runner.run();
}
try (FileInputStream fis = new FileInputStream("src/test/resources/hello.txt")) {
- Map<String, String> attributes = new HashMap<String, String>();
+ Map<String, String> attributes = new HashMap<>();
attributes.put(CoreAttributes.FILENAME.key(), "hello.txt");
attributes.put("transfer-host", "127.0.0.1");
runner.enqueue(fis, attributes);
@@ -198,7 +198,7 @@ public class TestFTP {
}
@Test
- public void basicFileGet() throws IOException {
+ public void basicFileGet() {
FileSystem results = fakeFtpServer.getFileSystem();
FileEntry sampleFile = new FileEntry("c:\\data\\randombytes-2");
@@ -222,7 +222,7 @@ public class TestFTP {
}
@Test
- public void basicFileFetch() throws IOException {
+ public void basicFileFetch() {
FileSystem results = fakeFtpServer.getFileSystem();
FileEntry sampleFile = new FileEntry("c:\\data\\randombytes-2");
@@ -242,7 +242,7 @@ public class TestFTP {
runner.setProperty(FetchFTP.MOVE_DESTINATION_DIR, "data");
- Map<String, String> attrs = new HashMap<String, String>();
+ Map<String, String> attrs = new HashMap<>();
attrs.put("host", "localhost");
attrs.put("username", username);
attrs.put("port", Integer.toString(ftpPort));
@@ -254,10 +254,46 @@ public class TestFTP {
retrievedFile.assertContentEquals("Just some random test test test chocolate");
}
+ @Test
+ public void testFetchFileNotFound() {
+ final TestRunner runner = TestRunners.newTestRunner(FetchFTP.class);
+ runner.setProperty(FetchFTP.HOSTNAME, "127.0.0.1");
+ runner.setProperty(FetchFTP.USERNAME, username);
+ runner.setProperty(FTPTransfer.PASSWORD, password);
+ runner.setProperty(FTPTransfer.PORT, Integer.toString(ftpPort));
+ runner.setProperty(FetchFTP.REMOTE_FILENAME, "remote-file-not-found");
+
+ runner.enqueue(new byte[0]);
+ runner.run();
+
+ runner.assertAllFlowFilesTransferred(FetchFTP.REL_NOT_FOUND);
+ }
+
+ @Test
+ public void testFetchFilePermissionDenied() {
+ final FileSystem fs = fakeFtpServer.getFileSystem();
+
+ final FileEntry restrictedFileEntry = new FileEntry("c:\\data\\restricted");
+ restrictedFileEntry.setPermissions(Permissions.NONE);
+ fs.add(restrictedFileEntry);
+
+ final TestRunner runner = TestRunners.newTestRunner(FetchFTP.class);
+ runner.setProperty(FetchFTP.HOSTNAME, "127.0.0.1");
+ runner.setProperty(FetchFTP.USERNAME, username);
+ runner.setProperty(FTPTransfer.PASSWORD, password);
+ runner.setProperty(FTPTransfer.PORT, Integer.toString(ftpPort));
+ runner.setProperty(FetchFTP.REMOTE_FILENAME, "restricted");
+
+ runner.enqueue(new byte[0]);
+ runner.run();
+
+ runner.assertAllFlowFilesTransferred(FetchFTP.REL_PERMISSION_DENIED);
+ }
+
@Test
@EnabledIfSystemProperty(named = "file.encoding", matches = "UTF-8",
disabledReason = "org.mockftpserver does not support specification of charset")
- public void basicFileFetchWithUTF8FileName() throws IOException {
+ public void basicFileFetchWithUTF8FileName() {
FileSystem fs = fakeFtpServer.getFileSystem();
FileEntry sampleFile = new FileEntry("c:\\data\\őűőű.txt");
@@ -284,7 +320,7 @@ public class TestFTP {
}
@Test
- public void basicFileList() throws IOException, InterruptedException {
+ public void basicFileList() throws InterruptedException {
FileSystem results = fakeFtpServer.getFileSystem();
FileEntry sampleFile = new FileEntry("c:\\data\\randombytes-2");
[nifi] 02/03: NIFI-9966 Corrects the registry loading of large flowfiles from git
Posted by jo...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
joewitt pushed a commit to branch support/nifi-1.16
in repository https://gitbox.apache.org/repos/asf/nifi.git
commit aa7a06e33dd853e242fc1f0dad0fb489c9a1d7a8
Author: Matthieu Bertin <be...@gmail.com>
AuthorDate: Tue May 3 21:16:46 2022 +0200
NIFI-9966 Corrects the registry loading of large flowfiles from git
This closes #6012
Signed-off-by: Chris Sampson <ch...@gmail.com>
---
.../provider/flow/git/GitFlowMetaData.java | 5 +++-
.../flow/git/TestGitFlowPersistenceProvider.java | 29 ++++++++++++++++++++++
2 files changed, 33 insertions(+), 1 deletion(-)
diff --git a/nifi-registry/nifi-registry-core/nifi-registry-framework/src/main/java/org/apache/nifi/registry/provider/flow/git/GitFlowMetaData.java b/nifi-registry/nifi-registry-core/nifi-registry-framework/src/main/java/org/apache/nifi/registry/provider/flow/git/GitFlowMetaData.java
index 8ed146dfd2..4bac3284b3 100644
--- a/nifi-registry/nifi-registry-core/nifi-registry-framework/src/main/java/org/apache/nifi/registry/provider/flow/git/GitFlowMetaData.java
+++ b/nifi-registry/nifi-registry-core/nifi-registry-framework/src/main/java/org/apache/nifi/registry/provider/flow/git/GitFlowMetaData.java
@@ -16,6 +16,7 @@
*/
package org.apache.nifi.registry.provider.flow.git;
+import org.apache.commons.io.IOUtils;
import org.apache.commons.lang3.concurrent.BasicThreadFactory;
import org.eclipse.jgit.api.Git;
import org.eclipse.jgit.api.LsRemoteCommand;
@@ -24,6 +25,7 @@ import org.eclipse.jgit.api.Status;
import org.eclipse.jgit.api.errors.GitAPIException;
import org.eclipse.jgit.api.errors.NoHeadException;
import org.eclipse.jgit.lib.ObjectId;
+import org.eclipse.jgit.lib.ObjectStream;
import org.eclipse.jgit.lib.Ref;
import org.eclipse.jgit.lib.Repository;
import org.eclipse.jgit.lib.RepositoryCache;
@@ -522,7 +524,8 @@ class GitFlowMetaData {
byte[] getContent(String objectId) throws IOException {
final ObjectId flowSnapshotObjectId = gitRepo.resolve(objectId);
- return gitRepo.newObjectReader().open(flowSnapshotObjectId).getBytes();
+ final ObjectStream objStream = gitRepo.newObjectReader().open(flowSnapshotObjectId).openStream();
+ return IOUtils.toByteArray(objStream);
}
}
diff --git a/nifi-registry/nifi-registry-core/nifi-registry-framework/src/test/java/org/apache/nifi/registry/provider/flow/git/TestGitFlowPersistenceProvider.java b/nifi-registry/nifi-registry-core/nifi-registry-framework/src/test/java/org/apache/nifi/registry/provider/flow/git/TestGitFlowPersistenceProvider.java
index 45351abc14..a8b050a37e 100644
--- a/nifi-registry/nifi-registry-core/nifi-registry-framework/src/test/java/org/apache/nifi/registry/provider/flow/git/TestGitFlowPersistenceProvider.java
+++ b/nifi-registry/nifi-registry-core/nifi-registry-framework/src/test/java/org/apache/nifi/registry/provider/flow/git/TestGitFlowPersistenceProvider.java
@@ -16,6 +16,7 @@
*/
package org.apache.nifi.registry.provider.flow.git;
+import org.apache.commons.lang3.RandomUtils;
import org.apache.nifi.registry.flow.FlowPersistenceException;
import org.apache.nifi.registry.provider.ProviderConfigurationContext;
import org.apache.nifi.registry.provider.ProviderCreationException;
@@ -39,6 +40,7 @@ import java.util.Map;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.function.Consumer;
+import static org.junit.Assert.assertArrayEquals;
import static org.junit.Assert.assertEquals;
import static org.junit.Assert.fail;
@@ -287,4 +289,31 @@ public class TestGitFlowPersistenceProvider {
}
}, true);
}
+
+ @Test
+ public void testLoadLargeFlow() throws GitAPIException, IOException {
+ final Map<String, String> properties = new HashMap<>();
+ final byte[] largeByteContent = RandomUtils.nextBytes(60_000_000);
+ properties.put(GitFlowPersistenceProvider.FLOW_STORAGE_DIR_PROP, "target/repo-with-large-flow");
+
+ assertProvider(properties, g -> {}, p -> {
+ // Create some Flows and keep the directory.
+ final StandardFlowSnapshotContext.Builder contextBuilder = new StandardFlowSnapshotContext.Builder()
+ .bucketId("bucket-id-A")
+ .bucketName("C'est/Bucket A/です。")
+ .flowId("flow-id-1")
+ .flowName("テスト_用/フロー#1\\[contains invalid chars]")
+ .author("unit-test-user")
+ .comments("Initial commit.")
+ .snapshotTimestamp(new Date().getTime())
+ .version(1);
+ p.saveFlowContent(contextBuilder.build(), largeByteContent);
+ }, false);
+
+ assertProvider(properties, g -> {}, p -> {
+ // Should be able to load flow from commit histories.
+ final byte[] fromRepo = p.getFlowContent("bucket-id-A", "flow-id-1", 1);
+ assertArrayEquals(largeByteContent, fromRepo);
+ }, true);
+ }
}
[nifi] 01/03: NIFI-9470 This closes #6027. Allow creation of Parameter Context without any Inherited Parameter Contexts
Posted by jo...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
joewitt pushed a commit to branch support/nifi-1.16
in repository https://gitbox.apache.org/repos/asf/nifi.git
commit 462af16c81bf5fe8385c5b6af450165ba5edef5f
Author: Chris Sampson <ch...@gmail.com>
AuthorDate: Mon May 9 20:45:34 2022 +0100
NIFI-9470 This closes #6027. Allow creation of Parameter Context without any Inherited Parameter Contexts
Signed-off-by: Joe Witt <jo...@apache.org>
---
.../org/apache/nifi/web/dao/impl/StandardParameterContextDAO.java | 6 +++---
1 file changed, 3 insertions(+), 3 deletions(-)
diff --git a/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-web/nifi-web-api/src/main/java/org/apache/nifi/web/dao/impl/StandardParameterContextDAO.java b/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-web/nifi-web-api/src/main/java/org/apache/nifi/web/dao/impl/StandardParameterContextDAO.java
index 18616feae4..3c7b5b5bf2 100644
--- a/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-web/nifi-web-api/src/main/java/org/apache/nifi/web/dao/impl/StandardParameterContextDAO.java
+++ b/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-web/nifi-web-api/src/main/java/org/apache/nifi/web/dao/impl/StandardParameterContextDAO.java
@@ -82,9 +82,9 @@ public class StandardParameterContextDAO implements ParameterContextDAO {
final AtomicReference<ParameterContext> parameterContextReference = new AtomicReference<>();
flowManager.withParameterContextResolution(() -> {
- final List<String> referencedIds = parameterContextDto.getInheritedParameterContexts().stream()
- .map(ParameterContextReferenceEntity::getId)
- .collect(Collectors.toList());
+ final List<String> referencedIds = parameterContextDto.getInheritedParameterContexts() == null
+ ? new ArrayList<>(0)
+ : parameterContextDto.getInheritedParameterContexts().stream().map(ParameterContextReferenceEntity::getId).collect(Collectors.toList());
final ParameterContext parameterContext = flowManager.createParameterContext(parameterContextDto.getId(), parameterContextDto.getName(), parameters, referencedIds);
if (parameterContextDto.getDescription() != null) {