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/12 19:51:35 UTC
[nifi] 02/04: NIFI-10023: Ensure that any files that are archived upon startup properly increment the archival count in the content repo
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 ad75c80f32c17d061650c64722a640086bfdc1ec
Author: Mark Payne <ma...@hotmail.com>
AuthorDate: Thu May 12 11:10:36 2022 -0400
NIFI-10023: Ensure that any files that are archived upon startup properly increment the archival count in the content repo
Signed-off-by: Matthew Burgess <ma...@apache.org>
This closes #6037
---
.../repository/FileSystemRepository.java | 29 +++++++++++--
.../repository/TestFileSystemRepository.java | 50 ++++++++++++++++++++++
.../src/test/resources/conf/nifi.properties | 2 +
3 files changed, 78 insertions(+), 3 deletions(-)
diff --git a/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/controller/repository/FileSystemRepository.java b/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/controller/repository/FileSystemRepository.java
index 4bebb44bbe..755bef032a 100644
--- a/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/controller/repository/FileSystemRepository.java
+++ b/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/controller/repository/FileSystemRepository.java
@@ -504,11 +504,11 @@ public class FileSystemRepository implements ContentRepository {
final ResourceClaim resourceClaim = resourceClaimManager.newResourceClaim(containerName, sectionName, id, false, false);
if (resourceClaimManager.getClaimantCount(resourceClaim) == 0) {
- removeIncompleteContent(fileToRemove);
+ removeIncompleteContent(fileToRemove, containerName);
}
}
- private void removeIncompleteContent(final Path fileToRemove) {
+ private void removeIncompleteContent(final Path fileToRemove, final String containerName) {
String fileDescription = null;
try {
fileDescription = fileToRemove.toFile().getAbsolutePath() + " (" + Files.size(fileToRemove) + " bytes)";
@@ -520,7 +520,16 @@ public class FileSystemRepository implements ContentRepository {
try {
if (archiveData) {
- archive(fileToRemove);
+ final boolean archived = archive(fileToRemove);
+
+ if (archived) {
+ final ContainerState containerState = containerStateMap.get(containerName);
+ if (containerState == null) {
+ LOG.warn("Failed to increment container's archive count for {} because container {} could not be found", fileToRemove.toFile(), containerName);
+ } else {
+ containerState.incrementArchiveCount();
+ }
+ }
} else {
Files.delete(fileToRemove);
}
@@ -531,6 +540,16 @@ public class FileSystemRepository implements ContentRepository {
}
}
+ // Visible for testing
+ long getArchiveCount(String containerName) {
+ final ContainerState containerState = containerStateMap.get(containerName);
+ if (containerState == null) {
+ throw new IllegalArgumentException("No container exists with name " + containerName);
+ }
+
+ return containerState.getArchiveCount();
+ }
+
@Override
public boolean isActiveResourceClaimsSupported() {
return true;
@@ -1769,6 +1788,10 @@ public class FileSystemRepository implements ContentRepository {
archivedFileCount.incrementAndGet();
}
+ public long getArchiveCount() {
+ return archivedFileCount.get();
+ }
+
public void decrementArchiveCount() {
archivedFileCount.decrementAndGet();
}
diff --git a/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/test/java/org/apache/nifi/controller/repository/TestFileSystemRepository.java b/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/test/java/org/apache/nifi/controller/repository/TestFileSystemRepository.java
index e05938ac93..db762bfa08 100644
--- a/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/test/java/org/apache/nifi/controller/repository/TestFileSystemRepository.java
+++ b/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/test/java/org/apache/nifi/controller/repository/TestFileSystemRepository.java
@@ -55,6 +55,7 @@ import java.util.Arrays;
import java.util.Collections;
import java.util.List;
import java.util.Locale;
+import java.util.Map;
import java.util.Random;
import java.util.concurrent.TimeUnit;
@@ -140,6 +141,55 @@ public class TestFileSystemRepository {
assertTrue(repository.isArchived(Paths.get("a/b/c/archive/1.txt")));
}
+ @Test
+ public void testUnreferencedFilesAreArchivedOnCleanup() throws IOException {
+ final Map<String, Path> containerPaths = nifiProperties.getContentRepositoryPaths();
+ assertTrue(containerPaths.size() > 0);
+
+ for (final Map.Entry<String, Path> entry : containerPaths.entrySet()) {
+ final String containerName = entry.getKey();
+ final Path containerPath = entry.getValue();
+
+ final Path section1 = containerPath.resolve("1");
+ final Path file1 = section1.resolve("file-1");
+ Files.write(file1, "hello".getBytes(), StandardOpenOption.CREATE);
+
+ // Should be nothing in the archive at this point
+ assertEquals(0, repository.getArchiveCount(containerName));
+
+ // When we cleanup, we should see one file moved to archive
+ repository.cleanup();
+ assertEquals(1, repository.getArchiveCount(containerName));
+ }
+ }
+
+ @Test
+ public void testAlreadyArchivedFilesCounted() throws IOException {
+ // We want to make sure that the initialization code counts files in archive, so we need to create a new FileSystemRepository to do this.
+ repository.shutdown();
+
+ final Map<String, Path> containerPaths = nifiProperties.getContentRepositoryPaths();
+ assertTrue(containerPaths.size() > 0);
+
+ for (final Path containerPath : containerPaths.values()) {
+ final Path section1 = containerPath.resolve("1");
+ final Path archive = section1.resolve("archive");
+ Files.createDirectories(archive);
+
+ for (int i=0; i < 3; i++) {
+ final Path file1 = archive.resolve("file-" + i);
+ Files.write(file1, "hello".getBytes(), StandardOpenOption.CREATE);
+ }
+ }
+
+ repository = new FileSystemRepository(nifiProperties);
+
+ for (final String containerName : containerPaths.keySet()) {
+ assertEquals(3, repository.getArchiveCount(containerName));
+ }
+ }
+
+
@Test
public void testContentNotFoundExceptionThrownIfResourceClaimTooShort() throws IOException {
final File contentFile = new File("target/content_repository/0/0.bin");
diff --git a/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/test/resources/conf/nifi.properties b/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/test/resources/conf/nifi.properties
index 3490ee59eb..1427ae8607 100644
--- a/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/test/resources/conf/nifi.properties
+++ b/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/test/resources/conf/nifi.properties
@@ -48,6 +48,8 @@ nifi.swap.out.threads=4
nifi.content.claim.max.appendable.size=1 MB
nifi.content.claim.max.flow.files=100
nifi.content.repository.directory.default=./target/content_repository
+nifi.content.repository.archive.enabled=true
+nifi.content.repository.archive.max.usage.percentage=90%
# Provenance Repository Properties
nifi.provenance.repository.storage.directory=./target/provenance_repository