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/1bed849d Tree: http://git-wip-us.apache.org/repos/asf/nifi/tree/1bed849d Diff: http://git-wip-us.apache.org/repos/asf/nifi/diff/1bed849d Branch: refs/heads/HDF-3.1-maint Commit: 1bed849dcb83ac24de0e0629ef4e3db2fe29d2c4 Parents: 8f988c3 Author: Marco Gaido <[email protected]> Authored: Thu Apr 5 13:44:55 2018 +0200 Committer: Matt Gilman <[email protected]> Committed: Fri Apr 13 09:34:24 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/1bed849d/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 2234265..01893e9 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 @@ -282,11 +282,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()); @@ -298,13 +294,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); @@ -569,7 +573,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();
