This is an automated email from the ASF dual-hosted git repository.
joewitt 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 169dff927a9 NIFI-15958 Added progress logging for repository scanning
This closes #11671
169dff927a9 is described below
commit 169dff927a9addcb0482788b6706bf959a363ca4
Author: exceptionfactory <[email protected]>
AuthorDate: Fri Sep 11 11:09:10 2026 -0500
NIFI-15958 Added progress logging for repository scanning
This closes #11671
---
.../nifi/provenance/store/WriteAheadStorePartition.java | 10 ++++++++++
.../nifi/controller/repository/FileSystemRepository.java | 14 ++++++++++++--
2 files changed, 22 insertions(+), 2 deletions(-)
diff --git
a/nifi-framework-bundle/nifi-framework-extensions/nifi-provenance-repository-bundle/nifi-persistent-provenance-repository/src/main/java/org/apache/nifi/provenance/store/WriteAheadStorePartition.java
b/nifi-framework-bundle/nifi-framework-extensions/nifi-provenance-repository-bundle/nifi-persistent-provenance-repository/src/main/java/org/apache/nifi/provenance/store/WriteAheadStorePartition.java
index 688ba9da577..5aeff416a9c 100644
---
a/nifi-framework-bundle/nifi-framework-extensions/nifi-provenance-repository-bundle/nifi-persistent-provenance-repository/src/main/java/org/apache/nifi/provenance/store/WriteAheadStorePartition.java
+++
b/nifi-framework-bundle/nifi-framework-extensions/nifi-provenance-repository-bundle/nifi-persistent-provenance-repository/src/main/java/org/apache/nifi/provenance/store/WriteAheadStorePartition.java
@@ -59,6 +59,7 @@ import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.Future;
import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicInteger;
import java.util.concurrent.atomic.AtomicLong;
import java.util.concurrent.atomic.AtomicReference;
import java.util.stream.Collectors;
@@ -66,6 +67,7 @@ import java.util.stream.Stream;
public class WriteAheadStorePartition implements EventStorePartition {
private static final Logger logger =
LoggerFactory.getLogger(WriteAheadStorePartition.class);
+ private static final int REINDEX_FILE_PROGRESS_LOG_INTERVAL = 25;
private final RepositoryConfiguration config;
private final File partitionDirectory;
@@ -611,6 +613,8 @@ public class WriteAheadStorePartition implements
EventStorePartition {
final ExecutorService executor =
Executors.newFixedThreadPool(Math.min(4, eventFilesToReindex.size()), new
NamedThreadFactory("Re-Index Provenance Events", true));
final List<Future<?>> futures = new
ArrayList<>(eventFilesToReindex.size());
final AtomicLong reindexedCount = new AtomicLong(0L);
+ final AtomicInteger filesReindexed = new AtomicInteger(0);
+ final int totalFilesToReindex = eventFilesToReindex.size();
// Re-Index the last bunch of events.
// We don't use an Event Iterator here because it's possible that one
of the event files could be corrupt (for example, if NiFi does while
@@ -659,6 +663,12 @@ public class WriteAheadStorePartition implements
EventStorePartition {
logger.warn("Failed to find event with ID {} in Event File
{}", minEventIdToReindex, eventFile, eof);
} catch (final Exception e) {
logger.error("Failed to index Provenance Events found in
{}", eventFile, e);
+ } finally {
+ final int filesCompleted =
filesReindexed.incrementAndGet();
+ if (filesCompleted == 1 || filesCompleted ==
totalFilesToReindex || filesCompleted % REINDEX_FILE_PROGRESS_LOG_INTERVAL ==
0) {
+ logger.info("Re-indexed {} of {} Files for Partition
[{}] including {} Events",
+ filesCompleted, totalFilesToReindex,
partitionName, reindexedCount.get());
+ }
}
};
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 ddcb048da27..7a9fc1eda6f 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
@@ -89,6 +89,7 @@ public class FileSystemRepository implements
ContentRepository {
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";
+ private static final int ARCHIVE_SCAN_SECTION_LOG_INTERVAL = 256;
// 100 MB cap for the configurable
NiFiProperties.MAX_APPENDABLE_CLAIM_SIZE property to prevent
// unnecessarily large resource claim files
public static final String APPENDABLE_CLAIM_LENGTH_CAP = "100 MB";
@@ -326,7 +327,7 @@ public class FileSystemRepository implements
ContentRepository {
// If the path didn't exist to begin with, there's no archive
directory, so don't bother scanning.
if (pathExists) {
- futures.add(executor.submit(() ->
scanArchiveDirectories(realPath.toFile(), containerState)));
+ futures.add(executor.submit(() ->
scanArchiveDirectories(containerName, realPath.toFile(), containerState)));
}
}
@@ -349,7 +350,8 @@ public class FileSystemRepository implements
ContentRepository {
containers.putAll(realPathMap);
}
- private void scanArchiveDirectories(final File containerDir, final
ContainerState containerState) {
+ private void scanArchiveDirectories(final String containerName, final File
containerDir, final ContainerState containerState) {
+ long archivedFilesFound = 0L;
for (int i = 0; i < SECTIONS_PER_CONTAINER; i++) {
final File sectionDir = new File(containerDir, String.valueOf(i));
final File archiveDir = new File(sectionDir, ARCHIVE_DIR_NAME);
@@ -363,7 +365,15 @@ public class FileSystemRepository implements
ContentRepository {
}
containerState.incrementArchiveCount(filenames.length);
+ archivedFilesFound += filenames.length;
+
+ final int sectionsScanned = i + 1;
+ if (sectionsScanned % ARCHIVE_SCAN_SECTION_LOG_INTERVAL == 0 &&
sectionsScanned < SECTIONS_PER_CONTAINER) {
+ 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);
}
// Visible for testing
