Repository: nifi Updated Branches: refs/heads/master 9dfef8f43 -> 0d45f21f4
NIFI-1200 fixed CPU saturation in FileSystemRepository added minimal interval added WARN message if set interval is lower NIFI-1200 fixed unused imports Project: http://git-wip-us.apache.org/repos/asf/nifi/repo Commit: http://git-wip-us.apache.org/repos/asf/nifi/commit/0395b903 Tree: http://git-wip-us.apache.org/repos/asf/nifi/tree/0395b903 Diff: http://git-wip-us.apache.org/repos/asf/nifi/diff/0395b903 Branch: refs/heads/master Commit: 0395b903152caef57644662dc4a00bcd490e26d7 Parents: 68975dc Author: Oleg Zhurakousky <[email protected]> Authored: Tue Jan 26 16:21:31 2016 -0500 Committer: Oleg Zhurakousky <[email protected]> Committed: Tue Jan 26 16:37:19 2016 -0500 ---------------------------------------------------------------------- .../repository/FileSystemRepository.java | 39 ++++++++++++++------ .../repository/TestFileSystemRepository.java | 35 +++++++++++++++++- 2 files changed, 62 insertions(+), 12 deletions(-) ---------------------------------------------------------------------- http://git-wip-us.apache.org/repos/asf/nifi/blob/0395b903/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/controller/repository/FileSystemRepository.java ---------------------------------------------------------------------- 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 c72a19c..9fec793 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 @@ -84,6 +84,7 @@ import org.slf4j.LoggerFactory; public class FileSystemRepository implements ContentRepository { public static final int SECTIONS_PER_CONTAINER = 1024; + public static final long MIN_CLEANUP_INTERVAL_MILLIS = 1000; public static final String ARCHIVE_DIR_NAME = "archive"; public static final Pattern MAX_ARCHIVE_SIZE_PATTERN = Pattern.compile("\\d{1,2}%"); private static final Logger LOG = LoggerFactory.getLogger(FileSystemRepository.class); @@ -226,17 +227,8 @@ public class FileSystemRepository implements ContentRepository { executor.scheduleWithFixedDelay(new ArchiveOrDestroyDestructableClaims(), 1, 1, TimeUnit.SECONDS); } - final String archiveCleanupFrequency = properties.getProperty(NiFiProperties.CONTENT_ARCHIVE_CLEANUP_FREQUENCY); - final long cleanupMillis; - if (archiveCleanupFrequency == null) { - cleanupMillis = 1000L; - } else { - try { - cleanupMillis = FormatUtils.getTimeDuration(archiveCleanupFrequency.trim(), TimeUnit.MILLISECONDS); - } catch (final Exception e) { - throw new RuntimeException("Invalid value set for property " + NiFiProperties.CONTENT_ARCHIVE_CLEANUP_FREQUENCY); - } - } + final long cleanupMillis = this.determineCleanupInterval(properties); + for (final Map.Entry<String, Path> containerEntry : containers.entrySet()) { final String containerName = containerEntry.getKey(); final Path containerPath = containerEntry.getValue(); @@ -1704,4 +1696,29 @@ public class FileSystemRepository implements ContentRepository { } } + /** + * Will determine the scheduling interval to be used by archive cleanup task + * (in milliseconds). This method will enforce the minimum allowed value of + * 1 second (1000 milliseconds). If attempt is made to set lower value a + * warning will be logged and the method will return minimum value of 1000 + */ + private long determineCleanupInterval(NiFiProperties properties) { + long cleanupInterval = MIN_CLEANUP_INTERVAL_MILLIS; + String archiveCleanupFrequency = properties.getProperty(NiFiProperties.CONTENT_ARCHIVE_CLEANUP_FREQUENCY); + if (archiveCleanupFrequency != null) { + try { + cleanupInterval = FormatUtils.getTimeDuration(archiveCleanupFrequency.trim(), TimeUnit.MILLISECONDS); + } catch (Exception e) { + throw new RuntimeException( + "Invalid value set for property " + NiFiProperties.CONTENT_ARCHIVE_CLEANUP_FREQUENCY); + } + if (cleanupInterval < MIN_CLEANUP_INTERVAL_MILLIS) { + LOG.warn("The value of " + NiFiProperties.CONTENT_ARCHIVE_CLEANUP_FREQUENCY + " property is set to '" + + archiveCleanupFrequency + "' which is " + + "below the allowed minimum of 1 second (1000 milliseconds). Minimum value of 1 sec will be used as scheduling interval for archive cleanup task."); + cleanupInterval = MIN_CLEANUP_INTERVAL_MILLIS; + } + } + return cleanupInterval; + } } http://git-wip-us.apache.org/repos/asf/nifi/blob/0395b903/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/test/java/org/apache/nifi/controller/repository/TestFileSystemRepository.java ---------------------------------------------------------------------- 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 5da67a0..c40d0e3 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 @@ -45,6 +45,12 @@ import org.apache.nifi.util.NiFiProperties; import org.junit.After; import org.junit.Before; import org.junit.Test; +import org.slf4j.LoggerFactory; + +import ch.qos.logback.classic.Level; +import ch.qos.logback.classic.Logger; +import ch.qos.logback.classic.spi.ILoggingEvent; +import ch.qos.logback.core.read.ListAppender; public class TestFileSystemRepository { @@ -61,7 +67,6 @@ public class TestFileSystemRepository { if (rootFile.exists()) { DiskUtils.deleteRecursively(rootFile); } - repository = new FileSystemRepository(); repository.initialize(new StandardResourceClaimManager()); repository.purge(); @@ -73,6 +78,34 @@ public class TestFileSystemRepository { } @Test + public void testMinimalArchiveCleanupIntervalHonoredAndLogged() throws Exception { + Logger root = (Logger) LoggerFactory.getLogger(Logger.ROOT_LOGGER_NAME); + ListAppender<ILoggingEvent> testAppender = new ListAppender<>(); + testAppender.setName("Test"); + testAppender.start(); + root.addAppender(testAppender); + NiFiProperties.getInstance().setProperty(NiFiProperties.CONTENT_ARCHIVE_CLEANUP_FREQUENCY, "1 millis"); + repository = new FileSystemRepository(); + repository.initialize(new StandardResourceClaimManager()); + repository.purge(); + + + boolean messageFound = false; + String message = "The value of nifi.content.repository.archive.cleanup.frequency property " + + "is set to '1 millis' which is below the allowed minimum of 1 second (1000 milliseconds). " + + "Minimum value of 1 sec will be used as scheduling interval for archive cleanup task."; + for (ILoggingEvent event : testAppender.list) { + String actualMessage = event.getFormattedMessage(); + if (actualMessage.equals(message)) { + assertEquals(event.getLevel(), Level.WARN); + messageFound = true; + break; + } + } + assertTrue(messageFound); + } + + @Test public void testBogusFile() throws IOException { repository.shutdown(); System.setProperty(NiFiProperties.PROPERTIES_FILE_PATH, "src/test/resources/nifi.properties");
