From 091b79aa4fa94b29a6ea2c1084fdb344f964e1ad Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?S=C3=B6nke=20Liebau?= Date: Fri, 21 Aug 2026 12:19:56 +0000 Subject: [PATCH] feat(nifi): Backport NIFI-15958 to log periodic progress during archive scan and provenance re-index Applies apache/nifi#11269 to NiFi 2.6.0, 2.7.2 and 2.9.0. --- CHANGELOG.md | 2 + ...eriodic-progress-while-waiting-for-c.patch | 152 ++++++++++++++++++ ...eriodic-progress-while-waiting-for-c.patch | 152 ++++++++++++++++++ ...eriodic-progress-while-waiting-for-c.patch | 152 ++++++++++++++++++ 4 files changed, 458 insertions(+) create mode 100644 nifi/stackable/patches/2.6.0/0010-NIFI-15958-Log-periodic-progress-while-waiting-for-c.patch create mode 100644 nifi/stackable/patches/2.7.2/0010-NIFI-15958-Log-periodic-progress-while-waiting-for-c.patch create mode 100644 nifi/stackable/patches/2.9.0/0010-NIFI-15958-Log-periodic-progress-while-waiting-for-c.patch diff --git a/CHANGELOG.md b/CHANGELOG.md index 76c516713..524bd8926 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -7,6 +7,7 @@ All notable changes to this project will be documented in this file. ### Added - airflow, superset, druid, nifi: Add SBOMs for the frontend (npm) dependencies ([#1600]). +- nifi: Backport NIFI-15958 to log periodic progress while waiting for the content archive scan and provenance re-index, for `2.6.0`, `2.7.2`, and `2.9.0` ([#1610]). ### Changed @@ -25,6 +26,7 @@ All notable changes to this project will be documented in this file. [#1595]: https://github.com/stackabletech/docker-images/pull/1595 [#1596]: https://github.com/stackabletech/docker-images/pull/1596 [#1600]: https://github.com/stackabletech/docker-images/pull/1600 +[#1610]: https://github.com/stackabletech/docker-images/issues/1610 ## [26.7.0] - 2026-07-21 diff --git a/nifi/stackable/patches/2.6.0/0010-NIFI-15958-Log-periodic-progress-while-waiting-for-c.patch b/nifi/stackable/patches/2.6.0/0010-NIFI-15958-Log-periodic-progress-while-waiting-for-c.patch new file mode 100644 index 000000000..239a7c824 --- /dev/null +++ b/nifi/stackable/patches/2.6.0/0010-NIFI-15958-Log-periodic-progress-while-waiting-for-c.patch @@ -0,0 +1,152 @@ +From a8d7887240e96c281a30cf051db85d588f7d6e80 Mon Sep 17 00:00:00 2001 +From: Lars Francke +Date: Wed, 20 May 2026 16:33:13 +0200 +Subject: NIFI-15958 Log periodic progress while waiting for content archive + scan and provenance re-index + +Previously these calls blocked indefinitely on a Future.get() call. Now they wake up once a minute to report progress. + +Related: NIFI-4737 +--- + .../PartitionedWriteAheadEventStore.java | 41 ++++++++++++++---- + .../repository/FileSystemRepository.java | 42 +++++++++++++++---- + 2 files changed, 68 insertions(+), 15 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/PartitionedWriteAheadEventStore.java b/nifi-framework-bundle/nifi-framework-extensions/nifi-provenance-repository-bundle/nifi-persistent-provenance-repository/src/main/java/org/apache/nifi/provenance/store/PartitionedWriteAheadEventStore.java +index 749ca1d04e..f242a01976 100644 +--- a/nifi-framework-bundle/nifi-framework-extensions/nifi-provenance-repository-bundle/nifi-persistent-provenance-repository/src/main/java/org/apache/nifi/provenance/store/PartitionedWriteAheadEventStore.java ++++ b/nifi-framework-bundle/nifi-framework-extensions/nifi-provenance-repository-bundle/nifi-persistent-provenance-repository/src/main/java/org/apache/nifi/provenance/store/PartitionedWriteAheadEventStore.java +@@ -24,6 +24,8 @@ import org.apache.nifi.provenance.index.EventIndex; + import org.apache.nifi.provenance.serialization.EventFileCompressor; + import org.apache.nifi.provenance.store.iterator.AggregateEventIterator; + import org.apache.nifi.provenance.store.iterator.EventIterator; ++import org.slf4j.Logger; ++import org.slf4j.LoggerFactory; + + import java.io.File; + import java.io.IOException; +@@ -37,9 +39,13 @@ import java.util.concurrent.ExecutorService; + import java.util.concurrent.Executors; + import java.util.concurrent.Future; + import java.util.concurrent.LinkedBlockingQueue; ++import java.util.concurrent.TimeUnit; ++import java.util.concurrent.TimeoutException; + import java.util.concurrent.atomic.AtomicLong; + + public class PartitionedWriteAheadEventStore extends PartitionedEventStore { ++ private static final Logger logger = LoggerFactory.getLogger(PartitionedWriteAheadEventStore.class); ++ private static final long REINDEX_PROGRESS_LOG_INTERVAL_MINUTES = 1; + private final BlockingQueue filesToCompress; + private final List partitions; + private final RepositoryConfiguration repoConfig; +@@ -125,14 +131,35 @@ public class PartitionedWriteAheadEventStore extends PartitionedEventStore { + } + + executor.shutdown(); ++ ++ final long startNanos = System.nanoTime(); ++ final long warnIntervalNanos = TimeUnit.MINUTES.toNanos(REINDEX_PROGRESS_LOG_INTERVAL_MINUTES); ++ long nextWarnAtNanos = warnIntervalNanos; ++ + for (final Future future : futures) { +- try { +- future.get(); +- } catch (InterruptedException e) { +- Thread.currentThread().interrupt(); +- throw new RuntimeException("Failed to re-index events because Thread was interrupted", e); +- } catch (ExecutionException e) { +- throw new RuntimeException("Failed to re-index events", e); ++ while (true) { ++ try { ++ future.get(REINDEX_PROGRESS_LOG_INTERVAL_MINUTES, TimeUnit.MINUTES); ++ break; ++ } catch (final TimeoutException e) { ++ final long elapsedNanos = System.nanoTime() - startNanos; ++ if (elapsedNanos >= nextWarnAtNanos) { ++ long pending = 0; ++ for (final Future f : futures) { ++ if (!f.isDone()) { ++ pending++; ++ } ++ } ++ logger.info("Provenance re-indexing still in progress after {} seconds; {} of {} partitions still pending", ++ TimeUnit.NANOSECONDS.toSeconds(elapsedNanos), pending, numPartitions); ++ nextWarnAtNanos = elapsedNanos + warnIntervalNanos; ++ } ++ } catch (final InterruptedException e) { ++ Thread.currentThread().interrupt(); ++ throw new RuntimeException("Failed to re-index events because Thread was interrupted", e); ++ } catch (final ExecutionException e) { ++ throw new RuntimeException("Failed to re-index events", e); ++ } + } + } + } +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 c2d21a5565..b3c0a8dcf6 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 +@@ -72,6 +72,7 @@ import java.util.concurrent.Future; + import java.util.concurrent.LinkedBlockingQueue; + import java.util.concurrent.ScheduledExecutorService; + import java.util.concurrent.TimeUnit; ++import java.util.concurrent.TimeoutException; + import java.util.concurrent.atomic.AtomicLong; + import java.util.concurrent.locks.Condition; + import java.util.concurrent.locks.Lock; +@@ -87,6 +88,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 long ARCHIVE_SCAN_PROGRESS_LOG_INTERVAL_MINUTES = 1L; + // 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"; +@@ -321,15 +323,39 @@ public class FileSystemRepository implements ContentRepository { + + executor.shutdown(); + +- // Wait for all futures to complete ++ // Wait for all futures to complete, logging a periodic warning if the archive scan is taking a long time. ++ // On busy installs the archive directories can hold large numbers of files. ++ final long startNanos = System.nanoTime(); ++ final long warnIntervalNanos = TimeUnit.MINUTES.toNanos(ARCHIVE_SCAN_PROGRESS_LOG_INTERVAL_MINUTES); ++ long nextWarnAtNanos = warnIntervalNanos; ++ + for (final Future future : futures) { +- try { +- future.get(); +- } catch (final ExecutionException | InterruptedException e) { +- if (e.getCause() instanceof IOException) { +- throw (IOException) e.getCause(); +- } else { +- throw new RuntimeException(e); ++ while (true) { ++ try { ++ future.get(ARCHIVE_SCAN_PROGRESS_LOG_INTERVAL_MINUTES, TimeUnit.MINUTES); ++ break; ++ } catch (final TimeoutException e) { ++ final long elapsedNanos = System.nanoTime() - startNanos; ++ if (elapsedNanos >= nextWarnAtNanos) { ++ long pending = 0; ++ for (final Future f : futures) { ++ if (!f.isDone()) { ++ pending++; ++ } ++ } ++ LOG.info("Content repository archive directory scan still in progress after {} seconds; {} of {} containers still pending", ++ TimeUnit.NANOSECONDS.toSeconds(elapsedNanos), pending, futures.size()); ++ nextWarnAtNanos = elapsedNanos + warnIntervalNanos; ++ } ++ } catch (final InterruptedException e) { ++ Thread.currentThread().interrupt(); ++ throw new RuntimeException("Failed to scan archive directories because Thread was interrupted", e); ++ } catch (final ExecutionException e) { ++ if (e.getCause() instanceof final IOException ioException) { ++ throw ioException; ++ } else { ++ throw new RuntimeException("Failed to scan archive directories", e); ++ } + } + } + } diff --git a/nifi/stackable/patches/2.7.2/0010-NIFI-15958-Log-periodic-progress-while-waiting-for-c.patch b/nifi/stackable/patches/2.7.2/0010-NIFI-15958-Log-periodic-progress-while-waiting-for-c.patch new file mode 100644 index 000000000..d11a01b34 --- /dev/null +++ b/nifi/stackable/patches/2.7.2/0010-NIFI-15958-Log-periodic-progress-while-waiting-for-c.patch @@ -0,0 +1,152 @@ +From 087c05c3ad1a43cc3114324e6ea0d14c3f6e600c Mon Sep 17 00:00:00 2001 +From: Lars Francke +Date: Wed, 20 May 2026 16:33:13 +0200 +Subject: NIFI-15958 Log periodic progress while waiting for content archive + scan and provenance re-index + +Previously these calls blocked indefinitely on a Future.get() call. Now they wake up once a minute to report progress. + +Related: NIFI-4737 +--- + .../PartitionedWriteAheadEventStore.java | 41 ++++++++++++++---- + .../repository/FileSystemRepository.java | 42 +++++++++++++++---- + 2 files changed, 68 insertions(+), 15 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/PartitionedWriteAheadEventStore.java b/nifi-framework-bundle/nifi-framework-extensions/nifi-provenance-repository-bundle/nifi-persistent-provenance-repository/src/main/java/org/apache/nifi/provenance/store/PartitionedWriteAheadEventStore.java +index 749ca1d04e..f242a01976 100644 +--- a/nifi-framework-bundle/nifi-framework-extensions/nifi-provenance-repository-bundle/nifi-persistent-provenance-repository/src/main/java/org/apache/nifi/provenance/store/PartitionedWriteAheadEventStore.java ++++ b/nifi-framework-bundle/nifi-framework-extensions/nifi-provenance-repository-bundle/nifi-persistent-provenance-repository/src/main/java/org/apache/nifi/provenance/store/PartitionedWriteAheadEventStore.java +@@ -24,6 +24,8 @@ import org.apache.nifi.provenance.index.EventIndex; + import org.apache.nifi.provenance.serialization.EventFileCompressor; + import org.apache.nifi.provenance.store.iterator.AggregateEventIterator; + import org.apache.nifi.provenance.store.iterator.EventIterator; ++import org.slf4j.Logger; ++import org.slf4j.LoggerFactory; + + import java.io.File; + import java.io.IOException; +@@ -37,9 +39,13 @@ import java.util.concurrent.ExecutorService; + import java.util.concurrent.Executors; + import java.util.concurrent.Future; + import java.util.concurrent.LinkedBlockingQueue; ++import java.util.concurrent.TimeUnit; ++import java.util.concurrent.TimeoutException; + import java.util.concurrent.atomic.AtomicLong; + + public class PartitionedWriteAheadEventStore extends PartitionedEventStore { ++ private static final Logger logger = LoggerFactory.getLogger(PartitionedWriteAheadEventStore.class); ++ private static final long REINDEX_PROGRESS_LOG_INTERVAL_MINUTES = 1; + private final BlockingQueue filesToCompress; + private final List partitions; + private final RepositoryConfiguration repoConfig; +@@ -125,14 +131,35 @@ public class PartitionedWriteAheadEventStore extends PartitionedEventStore { + } + + executor.shutdown(); ++ ++ final long startNanos = System.nanoTime(); ++ final long warnIntervalNanos = TimeUnit.MINUTES.toNanos(REINDEX_PROGRESS_LOG_INTERVAL_MINUTES); ++ long nextWarnAtNanos = warnIntervalNanos; ++ + for (final Future future : futures) { +- try { +- future.get(); +- } catch (InterruptedException e) { +- Thread.currentThread().interrupt(); +- throw new RuntimeException("Failed to re-index events because Thread was interrupted", e); +- } catch (ExecutionException e) { +- throw new RuntimeException("Failed to re-index events", e); ++ while (true) { ++ try { ++ future.get(REINDEX_PROGRESS_LOG_INTERVAL_MINUTES, TimeUnit.MINUTES); ++ break; ++ } catch (final TimeoutException e) { ++ final long elapsedNanos = System.nanoTime() - startNanos; ++ if (elapsedNanos >= nextWarnAtNanos) { ++ long pending = 0; ++ for (final Future f : futures) { ++ if (!f.isDone()) { ++ pending++; ++ } ++ } ++ logger.info("Provenance re-indexing still in progress after {} seconds; {} of {} partitions still pending", ++ TimeUnit.NANOSECONDS.toSeconds(elapsedNanos), pending, numPartitions); ++ nextWarnAtNanos = elapsedNanos + warnIntervalNanos; ++ } ++ } catch (final InterruptedException e) { ++ Thread.currentThread().interrupt(); ++ throw new RuntimeException("Failed to re-index events because Thread was interrupted", e); ++ } catch (final ExecutionException e) { ++ throw new RuntimeException("Failed to re-index events", e); ++ } + } + } + } +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 c2d21a5565..b3c0a8dcf6 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 +@@ -72,6 +72,7 @@ import java.util.concurrent.Future; + import java.util.concurrent.LinkedBlockingQueue; + import java.util.concurrent.ScheduledExecutorService; + import java.util.concurrent.TimeUnit; ++import java.util.concurrent.TimeoutException; + import java.util.concurrent.atomic.AtomicLong; + import java.util.concurrent.locks.Condition; + import java.util.concurrent.locks.Lock; +@@ -87,6 +88,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 long ARCHIVE_SCAN_PROGRESS_LOG_INTERVAL_MINUTES = 1L; + // 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"; +@@ -321,15 +323,39 @@ public class FileSystemRepository implements ContentRepository { + + executor.shutdown(); + +- // Wait for all futures to complete ++ // Wait for all futures to complete, logging a periodic warning if the archive scan is taking a long time. ++ // On busy installs the archive directories can hold large numbers of files. ++ final long startNanos = System.nanoTime(); ++ final long warnIntervalNanos = TimeUnit.MINUTES.toNanos(ARCHIVE_SCAN_PROGRESS_LOG_INTERVAL_MINUTES); ++ long nextWarnAtNanos = warnIntervalNanos; ++ + for (final Future future : futures) { +- try { +- future.get(); +- } catch (final ExecutionException | InterruptedException e) { +- if (e.getCause() instanceof IOException) { +- throw (IOException) e.getCause(); +- } else { +- throw new RuntimeException(e); ++ while (true) { ++ try { ++ future.get(ARCHIVE_SCAN_PROGRESS_LOG_INTERVAL_MINUTES, TimeUnit.MINUTES); ++ break; ++ } catch (final TimeoutException e) { ++ final long elapsedNanos = System.nanoTime() - startNanos; ++ if (elapsedNanos >= nextWarnAtNanos) { ++ long pending = 0; ++ for (final Future f : futures) { ++ if (!f.isDone()) { ++ pending++; ++ } ++ } ++ LOG.info("Content repository archive directory scan still in progress after {} seconds; {} of {} containers still pending", ++ TimeUnit.NANOSECONDS.toSeconds(elapsedNanos), pending, futures.size()); ++ nextWarnAtNanos = elapsedNanos + warnIntervalNanos; ++ } ++ } catch (final InterruptedException e) { ++ Thread.currentThread().interrupt(); ++ throw new RuntimeException("Failed to scan archive directories because Thread was interrupted", e); ++ } catch (final ExecutionException e) { ++ if (e.getCause() instanceof final IOException ioException) { ++ throw ioException; ++ } else { ++ throw new RuntimeException("Failed to scan archive directories", e); ++ } + } + } + } diff --git a/nifi/stackable/patches/2.9.0/0010-NIFI-15958-Log-periodic-progress-while-waiting-for-c.patch b/nifi/stackable/patches/2.9.0/0010-NIFI-15958-Log-periodic-progress-while-waiting-for-c.patch new file mode 100644 index 000000000..7bc8cd52b --- /dev/null +++ b/nifi/stackable/patches/2.9.0/0010-NIFI-15958-Log-periodic-progress-while-waiting-for-c.patch @@ -0,0 +1,152 @@ +From ab12fadcf4b9b20e448098d6f6bbaf2e8ab6af87 Mon Sep 17 00:00:00 2001 +From: Lars Francke +Date: Wed, 20 May 2026 16:33:13 +0200 +Subject: NIFI-15958 Log periodic progress while waiting for content archive + scan and provenance re-index + +Previously these calls blocked indefinitely on a Future.get() call. Now they wake up once a minute to report progress. + +Related: NIFI-4737 +--- + .../PartitionedWriteAheadEventStore.java | 41 ++++++++++++++---- + .../repository/FileSystemRepository.java | 42 +++++++++++++++---- + 2 files changed, 68 insertions(+), 15 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/PartitionedWriteAheadEventStore.java b/nifi-framework-bundle/nifi-framework-extensions/nifi-provenance-repository-bundle/nifi-persistent-provenance-repository/src/main/java/org/apache/nifi/provenance/store/PartitionedWriteAheadEventStore.java +index 749ca1d04e..f242a01976 100644 +--- a/nifi-framework-bundle/nifi-framework-extensions/nifi-provenance-repository-bundle/nifi-persistent-provenance-repository/src/main/java/org/apache/nifi/provenance/store/PartitionedWriteAheadEventStore.java ++++ b/nifi-framework-bundle/nifi-framework-extensions/nifi-provenance-repository-bundle/nifi-persistent-provenance-repository/src/main/java/org/apache/nifi/provenance/store/PartitionedWriteAheadEventStore.java +@@ -24,6 +24,8 @@ import org.apache.nifi.provenance.index.EventIndex; + import org.apache.nifi.provenance.serialization.EventFileCompressor; + import org.apache.nifi.provenance.store.iterator.AggregateEventIterator; + import org.apache.nifi.provenance.store.iterator.EventIterator; ++import org.slf4j.Logger; ++import org.slf4j.LoggerFactory; + + import java.io.File; + import java.io.IOException; +@@ -37,9 +39,13 @@ import java.util.concurrent.ExecutorService; + import java.util.concurrent.Executors; + import java.util.concurrent.Future; + import java.util.concurrent.LinkedBlockingQueue; ++import java.util.concurrent.TimeUnit; ++import java.util.concurrent.TimeoutException; + import java.util.concurrent.atomic.AtomicLong; + + public class PartitionedWriteAheadEventStore extends PartitionedEventStore { ++ private static final Logger logger = LoggerFactory.getLogger(PartitionedWriteAheadEventStore.class); ++ private static final long REINDEX_PROGRESS_LOG_INTERVAL_MINUTES = 1; + private final BlockingQueue filesToCompress; + private final List partitions; + private final RepositoryConfiguration repoConfig; +@@ -125,14 +131,35 @@ public class PartitionedWriteAheadEventStore extends PartitionedEventStore { + } + + executor.shutdown(); ++ ++ final long startNanos = System.nanoTime(); ++ final long warnIntervalNanos = TimeUnit.MINUTES.toNanos(REINDEX_PROGRESS_LOG_INTERVAL_MINUTES); ++ long nextWarnAtNanos = warnIntervalNanos; ++ + for (final Future future : futures) { +- try { +- future.get(); +- } catch (InterruptedException e) { +- Thread.currentThread().interrupt(); +- throw new RuntimeException("Failed to re-index events because Thread was interrupted", e); +- } catch (ExecutionException e) { +- throw new RuntimeException("Failed to re-index events", e); ++ while (true) { ++ try { ++ future.get(REINDEX_PROGRESS_LOG_INTERVAL_MINUTES, TimeUnit.MINUTES); ++ break; ++ } catch (final TimeoutException e) { ++ final long elapsedNanos = System.nanoTime() - startNanos; ++ if (elapsedNanos >= nextWarnAtNanos) { ++ long pending = 0; ++ for (final Future f : futures) { ++ if (!f.isDone()) { ++ pending++; ++ } ++ } ++ logger.info("Provenance re-indexing still in progress after {} seconds; {} of {} partitions still pending", ++ TimeUnit.NANOSECONDS.toSeconds(elapsedNanos), pending, numPartitions); ++ nextWarnAtNanos = elapsedNanos + warnIntervalNanos; ++ } ++ } catch (final InterruptedException e) { ++ Thread.currentThread().interrupt(); ++ throw new RuntimeException("Failed to re-index events because Thread was interrupted", e); ++ } catch (final ExecutionException e) { ++ throw new RuntimeException("Failed to re-index events", e); ++ } + } + } + } +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 15ec9e78cd..9204a496b3 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 +@@ -72,6 +72,7 @@ import java.util.concurrent.Future; + import java.util.concurrent.LinkedBlockingQueue; + import java.util.concurrent.ScheduledExecutorService; + import java.util.concurrent.TimeUnit; ++import java.util.concurrent.TimeoutException; + import java.util.concurrent.atomic.AtomicLong; + import java.util.concurrent.locks.Condition; + import java.util.concurrent.locks.Lock; +@@ -87,6 +88,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 long ARCHIVE_SCAN_PROGRESS_LOG_INTERVAL_MINUTES = 1L; + // 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"; +@@ -319,15 +321,39 @@ public class FileSystemRepository implements ContentRepository { + + executor.shutdown(); + +- // Wait for all futures to complete ++ // Wait for all futures to complete, logging a periodic warning if the archive scan is taking a long time. ++ // On busy installs the archive directories can hold large numbers of files. ++ final long startNanos = System.nanoTime(); ++ final long warnIntervalNanos = TimeUnit.MINUTES.toNanos(ARCHIVE_SCAN_PROGRESS_LOG_INTERVAL_MINUTES); ++ long nextWarnAtNanos = warnIntervalNanos; ++ + for (final Future future : futures) { +- try { +- future.get(); +- } catch (final ExecutionException | InterruptedException e) { +- if (e.getCause() instanceof IOException) { +- throw (IOException) e.getCause(); +- } else { +- throw new RuntimeException(e); ++ while (true) { ++ try { ++ future.get(ARCHIVE_SCAN_PROGRESS_LOG_INTERVAL_MINUTES, TimeUnit.MINUTES); ++ break; ++ } catch (final TimeoutException e) { ++ final long elapsedNanos = System.nanoTime() - startNanos; ++ if (elapsedNanos >= nextWarnAtNanos) { ++ long pending = 0; ++ for (final Future f : futures) { ++ if (!f.isDone()) { ++ pending++; ++ } ++ } ++ LOG.info("Content repository archive directory scan still in progress after {} seconds; {} of {} containers still pending", ++ TimeUnit.NANOSECONDS.toSeconds(elapsedNanos), pending, futures.size()); ++ nextWarnAtNanos = elapsedNanos + warnIntervalNanos; ++ } ++ } catch (final InterruptedException e) { ++ Thread.currentThread().interrupt(); ++ throw new RuntimeException("Failed to scan archive directories because Thread was interrupted", e); ++ } catch (final ExecutionException e) { ++ if (e.getCause() instanceof final IOException ioException) { ++ throw ioException; ++ } else { ++ throw new RuntimeException("Failed to scan archive directories", e); ++ } + } + } + }