This is an automated email from the ASF dual-hosted git repository. joewitt pushed a commit to branch support/nifi-1.9.x in repository https://gitbox.apache.org/repos/asf/nifi.git
commit ffb268fb7a6e974d88cc57dc82066fb4f43c7f55 Author: Mike Thomsen <[email protected]> AuthorDate: Thu Jan 31 19:13:46 2019 -0500 NIFI-5987 Fixed issue where an invalid query pulled from an attribute would cause GetMongo to not route to failure. Signed-off-by: zenfenan <[email protected]> This closes #3285 --- .../mongodb/AbstractMongoQueryProcessor.java | 7 +-- .../apache/nifi/processors/mongodb/GetMongo.java | 12 +++-- .../apache/nifi/processors/mongodb/GetMongoIT.java | 57 ++++++++++++++++++++++ 3 files changed, 67 insertions(+), 9 deletions(-) diff --git a/nifi-nar-bundles/nifi-mongodb-bundle/nifi-mongodb-processors/src/main/java/org/apache/nifi/processors/mongodb/AbstractMongoQueryProcessor.java b/nifi-nar-bundles/nifi-mongodb-bundle/nifi-mongodb-processors/src/main/java/org/apache/nifi/processors/mongodb/AbstractMongoQueryProcessor.java index 6660551..7c66b1b 100644 --- a/nifi-nar-bundles/nifi-mongodb-bundle/nifi-mongodb-processors/src/main/java/org/apache/nifi/processors/mongodb/AbstractMongoQueryProcessor.java +++ b/nifi-nar-bundles/nifi-mongodb-bundle/nifi-mongodb-processors/src/main/java/org/apache/nifi/processors/mongodb/AbstractMongoQueryProcessor.java @@ -121,12 +121,7 @@ public abstract class AbstractMongoQueryProcessor extends AbstractMongoProcessor query = Document.parse(new String(out.toByteArray())); } catch (Exception ex) { getLogger().error("Error reading FlowFile : ", ex); - if (input != null) { //Likely culprit is a bad query - session.transfer(input, REL_FAILURE); - session.commit(); - } else { - throw new ProcessException(ex); - } + throw new ProcessException(ex); } } diff --git a/nifi-nar-bundles/nifi-mongodb-bundle/nifi-mongodb-processors/src/main/java/org/apache/nifi/processors/mongodb/GetMongo.java b/nifi-nar-bundles/nifi-mongodb-bundle/nifi-mongodb-processors/src/main/java/org/apache/nifi/processors/mongodb/GetMongo.java index ff65f86..286c47e 100644 --- a/nifi-nar-bundles/nifi-mongodb-bundle/nifi-mongodb-processors/src/main/java/org/apache/nifi/processors/mongodb/GetMongo.java +++ b/nifi-nar-bundles/nifi-mongodb-bundle/nifi-mongodb-processors/src/main/java/org/apache/nifi/processors/mongodb/GetMongo.java @@ -149,10 +149,16 @@ public class GetMongo extends AbstractMongoQueryProcessor { } } - final Document query = getQuery(context, session, input ); + final Document query; + try { + query = getQuery(context, session, input); + } catch (Exception ex) { + getLogger().error("Error parsing query.", ex); + if (input != null) { + session.transfer(input, REL_FAILURE); + } - if (query == null) { - return; + return; //We need to stop immediately. } final String jsonTypeSetting = context.getProperty(JSON_TYPE).getValue(); diff --git a/nifi-nar-bundles/nifi-mongodb-bundle/nifi-mongodb-processors/src/test/java/org/apache/nifi/processors/mongodb/GetMongoIT.java b/nifi-nar-bundles/nifi-mongodb-bundle/nifi-mongodb-processors/src/test/java/org/apache/nifi/processors/mongodb/GetMongoIT.java index 76139d5..286a70d 100644 --- a/nifi-nar-bundles/nifi-mongodb-bundle/nifi-mongodb-processors/src/test/java/org/apache/nifi/processors/mongodb/GetMongoIT.java +++ b/nifi-nar-bundles/nifi-mongodb-bundle/nifi-mongodb-processors/src/test/java/org/apache/nifi/processors/mongodb/GetMongoIT.java @@ -581,6 +581,7 @@ public class GetMongoIT { Assert.assertTrue(format.matcher((String) result.get("date_field")).matches()); } + @Test public void testClientService() throws Exception { MongoDBClientService clientService = new MongoDBControllerService(); runner.addControllerService("clientService", clientService); @@ -594,4 +595,60 @@ public class GetMongoIT { runner.run(); runner.assertTransferCount(GetMongo.REL_SUCCESS, 3); } + + @Test + public void testInvalidQueryGoesToFailure() { + //Test variable registry mode + runner.setVariable("badattribute", "<<?>>"); + runner.setProperty(GetMongo.QUERY, "${badattribute}"); + runner.run(); + runner.assertTransferCount(GetMongo.REL_FAILURE, 0); + runner.assertTransferCount(GetMongo.REL_SUCCESS, 0); + runner.assertTransferCount(GetMongo.REL_ORIGINAL, 0); + + runner.clearTransferState(); + + //Test that it doesn't blow up with variable registry values holding a proper value + runner.setVariable("badattribute", "{}"); + runner.run(); + runner.assertTransferCount(GetMongo.REL_FAILURE, 0); + runner.assertTransferCount(GetMongo.REL_SUCCESS, 3); + runner.assertTransferCount(GetMongo.REL_ORIGINAL, 0); + + runner.clearTransferState(); + + //Test a bad flowfile attribute + runner.setIncomingConnection(true); + runner.setProperty(GetMongo.QUERY, "${badfromff}"); + runner.enqueue("<<?>>", new HashMap<String, String>(){{ + put("badfromff", "{\"prop\":}"); + }}); + runner.run(); + runner.assertTransferCount(GetMongo.REL_FAILURE, 1); + runner.assertTransferCount(GetMongo.REL_SUCCESS, 0); + runner.assertTransferCount(GetMongo.REL_ORIGINAL, 0); + + runner.clearTransferState(); + + //Test for regression on a good query from a flowfile attribute + runner.setIncomingConnection(true); + runner.setProperty(GetMongo.QUERY, "${badfromff}"); + runner.enqueue("<<?>>", new HashMap<String, String>(){{ + put("badfromff", "{}"); + }}); + runner.run(); + runner.assertTransferCount(GetMongo.REL_FAILURE, 0); + runner.assertTransferCount(GetMongo.REL_SUCCESS, 3); + runner.assertTransferCount(GetMongo.REL_ORIGINAL, 1); + + runner.clearTransferState(); + runner.removeProperty(GetMongo.QUERY); + + //Test for regression against the body w/out any EL involved. + runner.enqueue("<<?>>"); + runner.run(); + runner.assertTransferCount(GetMongo.REL_FAILURE, 1); + runner.assertTransferCount(GetMongo.REL_SUCCESS, 0); + runner.assertTransferCount(GetMongo.REL_ORIGINAL, 0); + } }
