This is an automated email from the ASF dual-hosted git repository.
pvillard31 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 1a57774cab4 NIFI-16357: Use 12 sections per container for
FileSystemRepository in unit tests instead of 1024; this reduced run time from
2+ minutes to 24 seconds. Additionally, made a few minor tweaks to improve
timing-based reliability of tests. (#11691)
1a57774cab4 is described below
commit 1a57774cab455b04fac9f030b0ce8c26f4d0160c
Author: Mark Payne <[email protected]>
AuthorDate: Thu Sep 17 15:08:19 2026 -0400
NIFI-16357: Use 12 sections per container for FileSystemRepository in unit
tests instead of 1024; this reduced run time from 2+ minutes to 24 seconds.
Additionally, made a few minor tweaks to improve timing-based reliability of
tests. (#11691)
---
.../repository/FileSystemRepository.java | 35 +++++++++-----
.../repository/TestFileSystemRepository.java | 53 ++++++++++++----------
2 files changed, 54 insertions(+), 34 deletions(-)
diff --git
a/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/controller/repository/FileSystemRepository.java
b/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/controller/repository/FileSystemRepository.java
index 7a9fc1eda6f..031734d7f52 100644
---
a/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/controller/repository/FileSystemRepository.java
+++
b/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/controller/repository/FileSystemRepository.java
@@ -84,8 +84,7 @@ import java.util.regex.Pattern;
* Is thread safe
*/
public class FileSystemRepository implements ContentRepository {
-
- public static final int SECTIONS_PER_CONTAINER = 1024;
+ public static final int DEFAULT_SECTIONS_PER_CONTAINER = 1024;
public static final long MIN_CLEANUP_INTERVAL_MILLIS =
TimeUnit.SECONDS.toMillis(1L);
public static final long DEFAULT_CLEANUP_INTERVAL_MILLIS =
TimeUnit.MINUTES.toMillis(1L);
public static final String ARCHIVE_DIR_NAME = "archive";
@@ -138,9 +137,17 @@ public class FileSystemRepository implements
ContentRepository {
private final Map<String, BlockingQueue<ArchiveInfo>> archivedFiles = new
HashMap<>();
private final NiFiProperties nifiProperties;
+ private final AtomicLong claimTruncationRuns = new AtomicLong(0L);
+ private final int sectionsPerContainer;
public FileSystemRepository(final NiFiProperties nifiProperties) throws
IOException {
+ this(nifiProperties, DEFAULT_SECTIONS_PER_CONTAINER);
+ }
+
+ FileSystemRepository(final NiFiProperties nifiProperties, final int
sectionsPerContainer) throws IOException {
this.nifiProperties = nifiProperties;
+ this.sectionsPerContainer = sectionsPerContainer;
+
// determine the file repository paths and ensure they exist
final Map<String, Path> fileRepositoryPaths =
nifiProperties.getContentRepositoryPaths();
for (final Path path : fileRepositoryPaths.values()) {
@@ -256,7 +263,7 @@ public class FileSystemRepository implements
ContentRepository {
executor.scheduleWithFixedDelay(new
ArchiveOrDestroyDestructableClaims(), 1, 1, TimeUnit.SECONDS);
}
- final long cleanupMillis =
this.determineCleanupInterval(nifiProperties);
+ final long cleanupMillis = determineCleanupInterval(nifiProperties);
if (truncationEnabled) {
executor.scheduleWithFixedDelay(new TruncateClaims(),
cleanupMillis, cleanupMillis, TimeUnit.MILLISECONDS);
}
@@ -317,7 +324,7 @@ public class FileSystemRepository implements
ContentRepository {
}
// Ensure that the directory exists for the section, including the
archive directory, if configured to archive data.
- for (int i = 0; i < SECTIONS_PER_CONTAINER; i++) {
+ for (int i = 0; i < sectionsPerContainer; i++) {
final Path sectionPath = realPath.resolve(String.valueOf(i));
final Path toCreate = archiveData ?
sectionPath.resolve(ARCHIVE_DIR_NAME) : sectionPath;
Files.createDirectories(toCreate);
@@ -352,7 +359,7 @@ public class FileSystemRepository implements
ContentRepository {
private void scanArchiveDirectories(final String containerName, final File
containerDir, final ContainerState containerState) {
long archivedFilesFound = 0L;
- for (int i = 0; i < SECTIONS_PER_CONTAINER; i++) {
+ for (int i = 0; i < sectionsPerContainer; i++) {
final File sectionDir = new File(containerDir, String.valueOf(i));
final File archiveDir = new File(sectionDir, ARCHIVE_DIR_NAME);
if (!archiveDir.exists()) {
@@ -368,12 +375,12 @@ public class FileSystemRepository implements
ContentRepository {
archivedFilesFound += filenames.length;
final int sectionsScanned = i + 1;
- if (sectionsScanned % ARCHIVE_SCAN_SECTION_LOG_INTERVAL == 0 &&
sectionsScanned < SECTIONS_PER_CONTAINER) {
+ if (sectionsScanned % ARCHIVE_SCAN_SECTION_LOG_INTERVAL == 0 &&
sectionsScanned < sectionsPerContainer) {
LOG.info("Scanned {} Sections for [{}] found {} Archived
Files", sectionsScanned, containerName, archivedFilesFound);
}
}
- LOG.info("Finished scanning {} Sections for [{}] found {} Archived
Files", SECTIONS_PER_CONTAINER, containerName, archivedFilesFound);
+ LOG.info("Finished scanning {} Sections for [{}] found {} Archived
Files", sectionsPerContainer, containerName, archivedFilesFound);
}
// Visible for testing
@@ -551,7 +558,7 @@ public class FileSystemRepository implements
ContentRepository {
public Set<ResourceClaim> getActiveResourceClaims(final File containerDir,
final String containerName) {
final Set<ResourceClaim> activeResourceClaims = new HashSet<>();
- for (int i = 0; i < SECTIONS_PER_CONTAINER; i++) {
+ for (int i = 0; i < sectionsPerContainer; i++) {
final String sectionName = String.valueOf(i);
final File sectionDir = new File(containerDir, sectionName);
if (!sectionDir.exists()) {
@@ -681,7 +688,7 @@ public class FileSystemRepository implements
ContentRepository {
containerState.waitForArchiveExpiration();
}
- final long modulatedSectionIndex = currentIndex %
SECTIONS_PER_CONTAINER;
+ final long modulatedSectionIndex = currentIndex %
sectionsPerContainer;
final String section =
String.valueOf(modulatedSectionIndex).intern();
final String claimId = System.currentTimeMillis() + "-" +
currentIndex;
@@ -1065,6 +1072,10 @@ public class FileSystemRepository implements
ContentRepository {
resourceClaimManager.purge();
}
+ long getClaimTruncationRuns() {
+ return claimTruncationRuns.get();
+ }
+
private class TruncateClaims implements Runnable {
@Override
@@ -1094,11 +1105,13 @@ public class FileSystemRepository implements
ContentRepository {
final List<ContentClaim> toTruncate = new ArrayList<>();
resourceClaimManager.drainTruncatableClaims(toTruncate,
10_000);
if (toTruncate.isEmpty()) {
- return;
+ break;
}
truncateClaims(toTruncate, truncationActivationCache);
}
+
+ claimTruncationRuns.incrementAndGet();
}
private void truncateClaims(final List<ContentClaim> toTruncate, final
Map<String, Boolean> truncationActivationCache) {
@@ -1460,7 +1473,7 @@ public class FileSystemRepository implements
ContentRepository {
final StopWatch stopWatch = new StopWatch(true);
final AtomicLong expiredFilesDeleted = new AtomicLong(0L);
final AtomicLong expiredBytesDeleted = new AtomicLong(0L);
- for (int i = 0; i < SECTIONS_PER_CONTAINER; i++) {
+ for (int i = 0; i < sectionsPerContainer; i++) {
final Path sectionContainer = container.resolve(String.valueOf(i));
final Path archive = sectionContainer.resolve("archive");
if (!Files.exists(archive)) {
diff --git
a/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/test/java/org/apache/nifi/controller/repository/TestFileSystemRepository.java
b/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/test/java/org/apache/nifi/controller/repository/TestFileSystemRepository.java
index ba1a1ba4b60..490384a8472 100644
---
a/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/test/java/org/apache/nifi/controller/repository/TestFileSystemRepository.java
+++
b/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/test/java/org/apache/nifi/controller/repository/TestFileSystemRepository.java
@@ -71,14 +71,10 @@ import static org.junit.jupiter.api.Assertions.assertTrue;
import static org.junit.jupiter.api.Assertions.fail;
public class TestFileSystemRepository {
-
+ public static final int SECTIONS_PER_CONTAINER = 12;
public static final File helloWorldFile = new
File("src/test/resources/hello.txt");
private static final Logger logger =
LoggerFactory.getLogger(TestFileSystemRepository.class);
- // The tests configure CONTENT_ARCHIVE_CLEANUP_FREQUENCY to "1 sec", which
is the minimum allowed interval for
- // the background TruncateClaims task. Waiting twice that interval ensures
at least one cycle has executed.
- private static final long BACKGROUND_TASK_WAIT_MILLIS = 2_000L;
-
@TempDir
private Path tempDir;
@@ -100,7 +96,7 @@ public class TestFileSystemRepository {
);
nifiProperties =
NiFiProperties.createBasicNiFiProperties(originalNifiPropertiesFile.toString(),
additionalProperties);
maxClaimLength =
DataUnit.parseDataSize(nifiProperties.getMaxAppendableClaimSize(),
DataUnit.B).longValue();
- repository = new FileSystemRepository(nifiProperties);
+ repository = new FileSystemRepository(nifiProperties,
SECTIONS_PER_CONTAINER);
claimManager = new StandardResourceClaimManager();
repository.initialize(new
StandardContentRepositoryContext(claimManager, EventReporter.NO_OP));
repository.purge();
@@ -160,7 +156,7 @@ public class TestFileSystemRepository {
// the freshly archived claim before this test observes it, leaving
the archive count oscillating at 0.
shutdown();
- final FileSystemRepository localRepository = new
FileSystemRepository(nifiProperties) {
+ final FileSystemRepository localRepository = new
FileSystemRepository(nifiProperties, SECTIONS_PER_CONTAINER) {
@Override
public long getContainerUsableSpace(final String containerName) {
return Long.MAX_VALUE;
@@ -215,8 +211,8 @@ public class TestFileSystemRepository {
public void testArchivedClaimRemovedDueToAge() throws IOException,
InterruptedException {
// Recreate Repository with specific properties
final Map<String, String> propertyOverrides = new HashMap<>();
-
propertyOverrides.put(NiFiProperties.CONTENT_ARCHIVE_MAX_RETENTION_PERIOD, "2
sec");
-
propertyOverrides.put(NiFiProperties.CONTENT_ARCHIVE_CLEANUP_FREQUENCY, "1
sec");
+
propertyOverrides.put(NiFiProperties.CONTENT_ARCHIVE_MAX_RETENTION_PERIOD, "20
millis");
+
propertyOverrides.put(NiFiProperties.CONTENT_ARCHIVE_CLEANUP_FREQUENCY, "10
millis");
propertyOverrides.put(NiFiProperties.CONTENT_ARCHIVE_MAX_USAGE_PERCENTAGE,
"99%");
recreateRepositoryWithPropertyOverrides(propertyOverrides);
@@ -272,7 +268,7 @@ public class TestFileSystemRepository {
repository.shutdown();
nifiProperties =
NiFiProperties.createBasicNiFiProperties(TestFileSystemRepository.class.getResource("/conf/nifi.properties").getFile(),
propertyOverrides);
- repository = new FileSystemRepository(nifiProperties);
+ repository = new FileSystemRepository(nifiProperties,
SECTIONS_PER_CONTAINER);
claimManager = new StandardResourceClaimManager();
repository.initialize(new
StandardContentRepositoryContext(claimManager, EventReporter.NO_OP));
repository.purge();
@@ -319,7 +315,7 @@ public class TestFileSystemRepository {
}
}
- repository = new FileSystemRepository(nifiProperties);
+ repository = new FileSystemRepository(nifiProperties,
SECTIONS_PER_CONTAINER);
for (final String containerName : containerPaths.keySet()) {
assertEquals(3, repository.getArchiveCount(containerName));
@@ -378,7 +374,7 @@ public class TestFileSystemRepository {
Files.createDirectories(bogus);
bogus.toFile().setReadable(false);
- repository = new FileSystemRepository(nifiProperties);
+ repository = new FileSystemRepository(nifiProperties,
SECTIONS_PER_CONTAINER);
repository.initialize(new StandardContentRepositoryContext(new
StandardResourceClaimManager(), EventReporter.NO_OP));
} finally {
bogus.toFile().setReadable(true);
@@ -464,7 +460,7 @@ public class TestFileSystemRepository {
repository.shutdown();
Thread.sleep(1000L);
- repository = new FileSystemRepository(nifiProperties);
+ repository = new FileSystemRepository(nifiProperties,
SECTIONS_PER_CONTAINER);
repository.initialize(new StandardContentRepositoryContext(new
StandardResourceClaimManager(), EventReporter.NO_OP));
repository.purge();
@@ -760,7 +756,7 @@ public class TestFileSystemRepository {
// We are creating our own 'local' repository in this test so shut
down the one created in the setup() method
shutdown();
- repository = new FileSystemRepository(nifiProperties) {
+ repository = new FileSystemRepository(nifiProperties,
SECTIONS_PER_CONTAINER) {
@Override
protected boolean archive(Path curPath) {
archivedPaths.add(curPath);
@@ -811,7 +807,7 @@ public class TestFileSystemRepository {
// We are creating our own 'local' repository in this test so shut
down the one created in the setup() method
shutdown();
- repository = new FileSystemRepository(nifiProperties) {
+ repository = new FileSystemRepository(nifiProperties,
SECTIONS_PER_CONTAINER) {
@Override
protected boolean archive(Path curPath) {
if (getOpenStreamCount() > 0) {
@@ -889,7 +885,7 @@ public class TestFileSystemRepository {
// We are creating our own 'local' repository in this test so shut
down the one created in the setup() method
shutdown();
- repository = new FileSystemRepository(nifiProperties) {
+ repository = new FileSystemRepository(nifiProperties,
SECTIONS_PER_CONTAINER) {
@Override
protected boolean archive(Path curPath) {
if (getOpenStreamCount() > 0) {
@@ -1012,7 +1008,7 @@ public class TestFileSystemRepository {
// We need to create our own repository that overrides
getContainerUsableSpace to simulate disk pressure
shutdown();
- final FileSystemRepository localRepository = new
FileSystemRepository(nifiProperties) {
+ final FileSystemRepository localRepository = new
FileSystemRepository(nifiProperties, SECTIONS_PER_CONTAINER) {
@Override
public long getContainerUsableSpace(final String containerName) {
return 0; // Extreme disk pressure
@@ -1081,7 +1077,7 @@ public class TestFileSystemRepository {
public void testTruncateNotActiveWhenDiskNotPressured() throws
IOException, InterruptedException {
shutdown();
- final FileSystemRepository localRepository = new
FileSystemRepository(nifiProperties) {
+ final FileSystemRepository localRepository = new
FileSystemRepository(nifiProperties, SECTIONS_PER_CONTAINER) {
@Override
public long getContainerUsableSpace(final String containerName) {
return Long.MAX_VALUE;
@@ -1116,7 +1112,11 @@ public class TestFileSystemRepository {
localClaimManager.decrementClaimantCount(largeClaim.getResourceClaim());
localClaimManager.markTruncatable(largeClaim);
- Thread.sleep(BACKGROUND_TASK_WAIT_MILLIS);
+ final long initialRunCount =
localRepository.getClaimTruncationRuns();
+ while (localRepository.getClaimTruncationRuns() ==
initialRunCount) {
+ Thread.sleep(50L);
+ }
+
assertEquals(originalSize, Files.size(filePath));
} finally {
localRepository.shutdown();
@@ -1129,7 +1129,7 @@ public class TestFileSystemRepository {
shutdown();
final AtomicLong usableSpace = new AtomicLong(Long.MAX_VALUE);
- final FileSystemRepository localRepository = new
FileSystemRepository(nifiProperties) {
+ final FileSystemRepository localRepository = new
FileSystemRepository(nifiProperties, SECTIONS_PER_CONTAINER) {
@Override
public long getContainerUsableSpace(final String containerName) {
return usableSpace.get();
@@ -1167,7 +1167,11 @@ public class TestFileSystemRepository {
localClaimManager.markTruncatable(largeClaim);
// Wait for at least one run of the background task with no disk
pressure. File should not be truncated.
- Thread.sleep(BACKGROUND_TASK_WAIT_MILLIS);
+ final long initialRunCount =
localRepository.getClaimTruncationRuns();
+ while (localRepository.getClaimTruncationRuns() ==
initialRunCount) {
+ Thread.sleep(50L);
+ }
+
assertEquals(originalSize, Files.size(filePath));
// Now turn on disk pressure
@@ -1204,7 +1208,7 @@ public class TestFileSystemRepository {
public void
testTruncateClaimsSkipsClaimWithPositiveTruncationReferenceCount() throws
IOException, InterruptedException {
shutdown();
- final FileSystemRepository localRepository = new
FileSystemRepository(nifiProperties) {
+ final FileSystemRepository localRepository = new
FileSystemRepository(nifiProperties, SECTIONS_PER_CONTAINER) {
@Override
public long getContainerUsableSpace(final String containerName) {
return 0;
@@ -1260,8 +1264,11 @@ public class TestFileSystemRepository {
localClaimManager.markTruncatable(largeClaim);
localClaimManager.incrementTruncationReferenceCount(largeClaim);
+ final long initialRunCount = repository.getClaimTruncationRuns();
// Wait long enough for the TruncateClaims background task to have
executed at least once
- Thread.sleep(BACKGROUND_TASK_WAIT_MILLIS);
+ while (localRepository.getClaimTruncationRuns() ==
initialRunCount) {
+ Thread.sleep(50L);
+ }
assertEquals(originalSize, Files.size(filePath),
"File should not be truncated because TruncateClaims
should skip claims with positive truncation reference count");
