Apache
/
nifi
mirror of https://github.com/apache/nifi.git
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

NIFI-2975: Only start expiring FlowFiles after FlowFile Repository has been restored 

This closes #1274.

Signed-off-by: Bryan Bende <bbende@apache.org>
This commit is contained in:
Mark Payne 2016-11-28 16:15:05 -05:00 committed by Bryan Bende
parent 92f17a995b
commit 9d6ca6e1e6
No known key found for this signature in database
GPG Key ID: A0DDA9ED50711C39
1 changed files with 5 additions and 1 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

6
nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/controller/FlowController.java
View File

@ -484,7 +484,6 @@ public class FlowController implements EventAccess, ControllerServiceProvider, R
// PRIMARY_NODE_ONLY is deprecated, but still exists to handle processors that are still defined with it (they haven't been re-configured with executeNode = PRIMARY).
processScheduler.setSchedulingAgent(SchedulingStrategy.PRIMARY_NODE_ONLY, timerDrivenAgent);
processScheduler.setSchedulingAgent(SchedulingStrategy.CRON_DRIVEN, quartzSchedulingAgent);
processScheduler.scheduleFrameworkTask(new ExpireFlowFiles(this, contextFactory), "Expire FlowFiles", 30L, 30L, TimeUnit.SECONDS);
startConnectablesAfterInitialization = new ArrayList<>();
startRemoteGroupPortsAfterInitialization = new ArrayList<>();
@ -699,6 +698,11 @@ public class FlowController implements EventAccess, ControllerServiceProvider, R
flowFileRepository.loadFlowFiles(this, maxIdFromSwapFiles + 1);
// Begin expiring FlowFiles that are old
final ProcessContextFactory contextFactory = new ProcessContextFactory(contentRepository, flowFileRepository,
flowFileEventRepository, counterRepositoryRef.get(), provenanceRepository);
processScheduler.scheduleFrameworkTask(new ExpireFlowFiles(this, contextFactory), "Expire FlowFiles", 30L, 30L, TimeUnit.SECONDS);
// now that we've loaded the FlowFiles, this has restored our ContentClaims' states, so we can tell the
// ContentRepository to purge superfluous files
contentRepository.cleanup();