Repository: nifi Updated Branches: refs/heads/master 2c8c9374a -> ccedc71c8
NIFI-5077 ExtractGrok support for `keep empty captures` Support for the new option to keep empty captures. I did not add to the GrokReader because I am not sure of the effect on the schema extraction. Signed-off-by: Pierre Villard <[email protected]> This closes #2698. Project: http://git-wip-us.apache.org/repos/asf/nifi/repo Commit: http://git-wip-us.apache.org/repos/asf/nifi/commit/ccedc71c Tree: http://git-wip-us.apache.org/repos/asf/nifi/tree/ccedc71c Diff: http://git-wip-us.apache.org/repos/asf/nifi/diff/ccedc71c Branch: refs/heads/master Commit: ccedc71c8b89e1792d79197f399b37b1ac541e3f Parents: 2c8c937 Author: Otto Fowler <[email protected]> Authored: Mon May 14 09:35:23 2018 -0400 Committer: Pierre Villard <[email protected]> Committed: Tue May 15 19:24:10 2018 +0200 ---------------------------------------------------------------------- .../nifi/processors/standard/ExtractGrok.java | 18 +++++++++++++++ .../processors/standard/TestExtractGrok.java | 24 ++++++++++++++++++++ 2 files changed, 42 insertions(+) ---------------------------------------------------------------------- http://git-wip-us.apache.org/repos/asf/nifi/blob/ccedc71c/nifi-nar-bundles/nifi-standard-bundle/nifi-standard-processors/src/main/java/org/apache/nifi/processors/standard/ExtractGrok.java ---------------------------------------------------------------------- diff --git a/nifi-nar-bundles/nifi-standard-bundle/nifi-standard-processors/src/main/java/org/apache/nifi/processors/standard/ExtractGrok.java b/nifi-nar-bundles/nifi-standard-bundle/nifi-standard-processors/src/main/java/org/apache/nifi/processors/standard/ExtractGrok.java index c5675ea..5896b93 100644 --- a/nifi-nar-bundles/nifi-standard-bundle/nifi-standard-processors/src/main/java/org/apache/nifi/processors/standard/ExtractGrok.java +++ b/nifi-nar-bundles/nifi-standard-bundle/nifi-standard-processors/src/main/java/org/apache/nifi/processors/standard/ExtractGrok.java @@ -67,6 +67,7 @@ import java.util.Set; import java.util.concurrent.BlockingQueue; import java.util.concurrent.LinkedBlockingQueue; import java.util.concurrent.TimeUnit; +import java.util.concurrent.atomic.AtomicBoolean; @EventDriven @SupportsBatching @@ -102,6 +103,15 @@ public class ExtractGrok extends AbstractProcessor { .addValidator(StandardValidators.FILE_EXISTS_VALIDATOR) .build(); + public static final PropertyDescriptor KEEP_EMPTY_CAPTURES = new PropertyDescriptor.Builder() + .name("Keep Empty Captures") + .description("If true, then empty capture values will be included in the returned capture map.") + .required(true) + .defaultValue("true") + .allowableValues("true","false") + .addValidator(StandardValidators.BOOLEAN_VALIDATOR) + .build(); + public static final PropertyDescriptor DESTINATION = new PropertyDescriptor.Builder() .name("Destination") .description("Control if Grok output value is written as a new flowfile attributes, in this case " + @@ -156,6 +166,8 @@ public class ExtractGrok extends AbstractProcessor { private volatile Grok grok; private final BlockingQueue<byte[]> bufferQueue = new LinkedBlockingQueue<>(); + private final AtomicBoolean keepEmptyCaputures = new AtomicBoolean(true); + static { final Set<Relationship> _relationships = new HashSet<>(); _relationships.add(REL_MATCH); @@ -169,6 +181,7 @@ public class ExtractGrok extends AbstractProcessor { _descriptors.add(CHARACTER_SET); _descriptors.add(MAX_BUFFER_SIZE); _descriptors.add(NAMED_CAPTURES_ONLY); + _descriptors.add(KEEP_EMPTY_CAPTURES); descriptors = Collections.unmodifiableList(_descriptors); } @@ -234,6 +247,9 @@ public class ExtractGrok extends AbstractProcessor { @OnScheduled public void onScheduled(final ProcessContext context) throws GrokException, IOException { + + keepEmptyCaputures.set(context.getProperty(KEEP_EMPTY_CAPTURES).asBoolean()); + for (int i = 0; i < context.getMaxConcurrentTasks(); i++) { final int maxBufferSize = context.getProperty(MAX_BUFFER_SIZE).asDataSize(DataUnit.B).intValue(); final byte[] buffer = new byte[maxBufferSize]; @@ -254,6 +270,7 @@ public class ExtractGrok extends AbstractProcessor { } } grok = grokCompiler.compile(context.getProperty(GROK_EXPRESSION).getValue(), context.getProperty(NAMED_CAPTURES_ONLY).asBoolean()); + } @Override @@ -286,6 +303,7 @@ public class ExtractGrok extends AbstractProcessor { } final Match gm = grok.match(contentString); + gm.setKeepEmptyCaptures(keepEmptyCaputures.get()); final Map<String,Object> captureMap = gm.capture(); if (captureMap.isEmpty()) { http://git-wip-us.apache.org/repos/asf/nifi/blob/ccedc71c/nifi-nar-bundles/nifi-standard-bundle/nifi-standard-processors/src/test/java/org/apache/nifi/processors/standard/TestExtractGrok.java ---------------------------------------------------------------------- diff --git a/nifi-nar-bundles/nifi-standard-bundle/nifi-standard-processors/src/test/java/org/apache/nifi/processors/standard/TestExtractGrok.java b/nifi-nar-bundles/nifi-standard-bundle/nifi-standard-processors/src/test/java/org/apache/nifi/processors/standard/TestExtractGrok.java index ed70e5f..bc62cb4 100644 --- a/nifi-nar-bundles/nifi-standard-bundle/nifi-standard-processors/src/test/java/org/apache/nifi/processors/standard/TestExtractGrok.java +++ b/nifi-nar-bundles/nifi-standard-bundle/nifi-standard-processors/src/test/java/org/apache/nifi/processors/standard/TestExtractGrok.java @@ -66,6 +66,30 @@ public class TestExtractGrok { } @Test + public void testExtractGrokKeepEmptyCaptures() throws Exception { + String expression = "%{NUMBER}|%{NUMBER}"; + testRunner.setProperty(ExtractGrok.GROK_EXPRESSION,expression); + testRunner.enqueue("-42"); + testRunner.run(); + testRunner.assertAllFlowFilesTransferred(ExtractGrok.REL_MATCH); + final MockFlowFile matched = testRunner.getFlowFilesForRelationship(ExtractGrok.REL_MATCH).get(0); + matched.assertAttributeEquals("grok.NUMBER","[-42, null]"); + } + + @Test + public void testExtractGrokDoNotKeepEmptyCaptures() throws Exception { + String expression = "%{NUMBER}|%{NUMBER}"; + testRunner.setProperty(ExtractGrok.GROK_EXPRESSION,expression); + testRunner.setProperty(ExtractGrok.KEEP_EMPTY_CAPTURES,"false"); + testRunner.enqueue("-42"); + testRunner.run(); + testRunner.assertAllFlowFilesTransferred(ExtractGrok.REL_MATCH); + final MockFlowFile matched = testRunner.getFlowFilesForRelationship(ExtractGrok.REL_MATCH).get(0); + matched.assertAttributeEquals("grok.NUMBER","-42"); + } + + + @Test public void testExtractGrokWithUnMatchedContent() throws IOException { testRunner.setProperty(ExtractGrok.GROK_EXPRESSION, "%{URI}"); testRunner.setProperty(ExtractGrok.GROK_PATTERN_FILE, "src/test/resources/TestExtractGrok/patterns");
