Repository: nifi Updated Branches: refs/heads/master dc9b4cb51 -> c11954722
NIFI-5043: TailFile in Multifile mode should not open new readers in onTrigger This closes #2606. Signed-off-by: Mark Payne <[email protected]> Project: http://git-wip-us.apache.org/repos/asf/nifi/repo Commit: http://git-wip-us.apache.org/repos/asf/nifi/commit/c1195472 Tree: http://git-wip-us.apache.org/repos/asf/nifi/tree/c1195472 Diff: http://git-wip-us.apache.org/repos/asf/nifi/diff/c1195472 Branch: refs/heads/master Commit: c119547222514cf09dd457f0f21b54b307f5f56d Parents: dc9b4cb Author: Marco Gaido <[email protected]> Authored: Thu Apr 5 13:44:55 2018 +0200 Committer: Mark Payne <[email protected]> Committed: Thu Apr 5 13:53:01 2018 -0400 ---------------------------------------------------------------------- .../nifi/processors/standard/TailFile.java | 25 +++++++++++++------- 1 file changed, 16 insertions(+), 9 deletions(-) ---------------------------------------------------------------------- http://git-wip-us.apache.org/repos/asf/nifi/blob/c1195472/nifi-nar-bundles/nifi-standard-bundle/nifi-standard-processors/src/main/java/org/apache/nifi/processors/standard/TailFile.java ---------------------------------------------------------------------- diff --git a/nifi-nar-bundles/nifi-standard-bundle/nifi-standard-processors/src/main/java/org/apache/nifi/processors/standard/TailFile.java b/nifi-nar-bundles/nifi-standard-bundle/nifi-standard-processors/src/main/java/org/apache/nifi/processors/standard/TailFile.java index fc61c96..13226fc 100644 --- a/nifi-nar-bundles/nifi-standard-bundle/nifi-standard-processors/src/main/java/org/apache/nifi/processors/standard/TailFile.java +++ b/nifi-nar-bundles/nifi-standard-bundle/nifi-standard-processors/src/main/java/org/apache/nifi/processors/standard/TailFile.java @@ -290,11 +290,7 @@ public class TailFile extends AbstractProcessor { this.requireStateLookup = true; } - @OnScheduled - public void recoverState(final ProcessContext context) throws IOException { - // set isMultiChanging - isMultiChanging.set(context.getProperty(MODE).getValue().equals(MODE_MULTIFILE.getValue())); - + private List<String> lookup(final ProcessContext context) { // set last lookup to now lastLookup.set(new Date().getTime()); @@ -306,13 +302,21 @@ public class TailFile extends AbstractProcessor { if(context.getProperty(MODE).getValue().equals(MODE_MULTIFILE.getValue())) { filesToTail.addAll(getFilesToTail(context.getProperty(BASE_DIRECTORY).evaluateAttributeExpressions().getValue(), - context.getProperty(FILENAME).evaluateAttributeExpressions().getValue(), - context.getProperty(RECURSIVE).asBoolean(), - maxAge)); + context.getProperty(FILENAME).evaluateAttributeExpressions().getValue(), + context.getProperty(RECURSIVE).asBoolean(), + maxAge)); } else { filesToTail.add(context.getProperty(FILENAME).evaluateAttributeExpressions().getValue()); } + return filesToTail; + } + + @OnScheduled + public void recoverState(final ProcessContext context) throws IOException { + // set isMultiChanging + isMultiChanging.set(context.getProperty(MODE).getValue().equals(MODE_MULTIFILE.getValue())); + List<String> filesToTail = lookup(context); final Scope scope = getStateScope(context); final StateMap stateMap = context.getStateManager().getState(scope); @@ -577,7 +581,10 @@ public class TailFile extends AbstractProcessor { long timeSinceLastLookup = new Date().getTime() - lastLookup.get(); if(timeSinceLastLookup > context.getProperty(LOOKUP_FREQUENCY).asTimePeriod(TimeUnit.MILLISECONDS)) { try { - recoverState(context); + final List<String> filesToTail = lookup(context); + final Scope scope = getStateScope(context); + final StateMap stateMap = context.getStateManager().getState(scope); + initStates(filesToTail, stateMap.toMap(), false); } catch (IOException e) { getLogger().error("Exception raised while attempting to recover state about where the tailing last left off", e); context.yield();
