You are viewing a plain text version of this content. The canonical link for it is here.
Posted to issues@nifi.apache.org by mans2singh <gi...@git.apache.org> on 2017/07/16 02:52:39 UTC
[GitHub] nifi pull request #2012: NIFI-4188 - Nifi RethinkDB Get processor
GitHub user mans2singh opened a pull request:
https://github.com/apache/nifi/pull/2012
NIFI-4188 - Nifi RethinkDB Get processor
Thank you for submitting a contribution to Apache NiFi.
In order to streamline the review of the contribution we ask you
to ensure the following steps have been taken:
### For all changes:
- [x] Is there a JIRA ticket associated with this PR? Is it referenced
in the commit message?
- [x] Does your PR title start with NIFI-XXXX where XXXX is the JIRA number you are trying to resolve? Pay particular attention to the hyphen "-" character.
- [x] Has your PR been rebased against the latest commit within the target branch (typically master)?
- [x] Is your initial contribution a single, squashed commit?
### For code changes:
- [x] Have you ensured that the full suite of tests is executed via mvn -Pcontrib-check clean install at the root nifi folder?
- [x] Have you written or updated unit tests to verify your changes?
- [x] If adding new dependencies to the code, are these dependencies licensed in a way that is compatible for inclusion under [ASF 2.0](http://www.apache.org/legal/resolved.html#category-a)?
- [ ] If applicable, have you updated the LICENSE file, including the main LICENSE file under nifi-assembly?
- [x] If applicable, have you updated the NOTICE file, including the main NOTICE file found under nifi-assembly?
- [x] If adding new Properties, have you added .displayName in addition to .name (programmatic access) for each of the new properties?
### For documentation related changes:
- [x] Have you ensured that format looks appropriate for the output in which it is rendered?
### Note:
Please ensure that once the PR is submitted, you check travis-ci for build issues and submit an update to your PR as soon as possible.
You can merge this pull request into a Git repository by running:
$ git pull https://github.com/mans2singh/nifi NIFI-4188
Alternatively you can review and apply these changes as the patch at:
https://github.com/apache/nifi/pull/2012.patch
To close this pull request, make a commit to your master/trunk branch
with (at least) the following in the commit message:
This closes #2012
----
commit 0776617449934d8d961bc7d505defbdf00862220
Author: mans2singh <ma...@yahoo.com>
Date: 2017-07-16T02:18:09Z
NIFI-4188 Nifi RethinkDB Get processor
modified: nifi-nar-bundles/nifi-rethinkdb-bundle/nifi-rethinkdb-nar/src/main/resources/META-INF/NOTICE
modified: nifi-nar-bundles/nifi-rethinkdb-bundle/nifi-rethinkdb-processors/pom.xml
modified: nifi-nar-bundles/nifi-rethinkdb-bundle/nifi-rethinkdb-processors/src/main/java/org/apache/nifi/processors/rethinkdb/AbstractRethinkDBProcessor.java
new file: nifi-nar-bundles/nifi-rethinkdb-bundle/nifi-rethinkdb-processors/src/main/java/org/apache/nifi/processors/rethinkdb/GetRethinkDB.java
modified: nifi-nar-bundles/nifi-rethinkdb-bundle/nifi-rethinkdb-processors/src/main/java/org/apache/nifi/processors/rethinkdb/PutRethinkDB.java
modified: nifi-nar-bundles/nifi-rethinkdb-bundle/nifi-rethinkdb-processors/src/main/resources/META-INF/services/org.apache.nifi.processor.Processor
new file: nifi-nar-bundles/nifi-rethinkdb-bundle/nifi-rethinkdb-processors/src/test/java/org/apache/nifi/processors/rethinkdb/ITGetRethinkDBTest.java
new file: nifi-nar-bundles/nifi-rethinkdb-bundle/nifi-rethinkdb-processors/src/test/java/org/apache/nifi/processors/rethinkdb/TestGetRethinkDB.java
----
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] nifi issue #2012: NIFI-4188 - Nifi RethinkDB Get processor
Posted by jvwing <gi...@git.apache.org>.
Github user jvwing commented on the issue:
https://github.com/apache/nifi/pull/2012
My question about the not_found relationship is not if we should have it or not -- I think it is a great addition. My question is about the logging for that code path. You log an error on line 150 when a flowfile is routed to not_found, and I'm asking if you believe not_found is "bad enough" to deserve an error. It seems possible that it is a perfectly normal occurrence that keys are not found in the table, and the flowfile would be routed to not_found without any worries. As you point out, the user can choose to treat not_found separately from failure, or not, as they choose. But it is logged as an error regardless of their choice.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] nifi pull request #2012: NIFI-4188 - Nifi RethinkDB Get processor
Posted by jvwing <gi...@git.apache.org>.
Github user jvwing commented on a diff in the pull request:
https://github.com/apache/nifi/pull/2012#discussion_r128149561
--- Diff: nifi-nar-bundles/nifi-rethinkdb-bundle/nifi-rethinkdb-processors/src/main/java/org/apache/nifi/processors/rethinkdb/GetRethinkDB.java ---
@@ -137,9 +135,9 @@ public void onTrigger(final ProcessContext context, final ProcessSession session
String id = context.getProperty(RETHINKDB_DOCUMENT_ID).evaluateAttributeExpressions(flowFile).getValue();
String readMode = context.getProperty(READ_MODE).evaluateAttributeExpressions(flowFile).getValue();
- if ( StringUtils.isEmpty(id) ) {
- getLogger().error("Empty id '" + id + "'");
- flowFile = session.putAttribute(flowFile, RETHINKDB_ERROR_MESSAGE, "Empty id '" + id + "'");
+ if ( StringUtils.isBlank(id) ) {
--- End diff --
I do not recommend this change to using `isBlank()`, because it will return true if the id is whitespace only, and I believe whitespace is in fact a valid RethinkDB key.
I'm sorry I was not clear about my concern. In the earlier code using `isEmpty()`, the if block would be entered if id was null or an empty string. However, property values are not null, they would be an empty string at least. So the block appeared to only be entered in practice if id was an empty string. Given that we knew id was always an empty string, it did not seem necessary to interpolate the value. It is an empty string, and the result is always the same.
This appeared to be confirmed by your testBlankId() unit test that exercised this code and checked its value against `Blank id ''`. It appears to be further confirmed by the new testNullId() unit test that checks against the exact same value.
How about just logging it as `Document Identifier cannot be empty` or similar?
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] nifi pull request #2012: NIFI-4188 - Nifi RethinkDB Get processor
Posted by jvwing <gi...@git.apache.org>.
Github user jvwing commented on a diff in the pull request:
https://github.com/apache/nifi/pull/2012#discussion_r127879552
--- Diff: nifi-nar-bundles/nifi-rethinkdb-bundle/nifi-rethinkdb-processors/src/main/java/org/apache/nifi/processors/rethinkdb/GetRethinkDB.java ---
@@ -0,0 +1,197 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.nifi.processors.rethinkdb;
+
+import org.apache.commons.lang3.StringUtils;
+import org.apache.nifi.annotation.behavior.EventDriven;
+import org.apache.nifi.annotation.behavior.InputRequirement;
+import org.apache.nifi.annotation.behavior.WritesAttribute;
+import org.apache.nifi.annotation.behavior.WritesAttributes;
+import org.apache.nifi.annotation.documentation.CapabilityDescription;
+import org.apache.nifi.annotation.documentation.SeeAlso;
+import org.apache.nifi.annotation.documentation.Tags;
+import org.apache.nifi.annotation.lifecycle.OnScheduled;
+import org.apache.nifi.annotation.lifecycle.OnStopped;
+import org.apache.nifi.components.AllowableValue;
+import org.apache.nifi.components.PropertyDescriptor;
+import org.apache.nifi.expression.AttributeExpression.ResultType;
+import org.apache.nifi.flowfile.FlowFile;
+import org.apache.nifi.processor.ProcessContext;
+import org.apache.nifi.processor.ProcessSession;
+import org.apache.nifi.processor.Relationship;
+import org.apache.nifi.processor.exception.ProcessException;
+import org.apache.nifi.processor.util.StandardValidators;
+
+import com.google.gson.Gson;
+
+import java.io.ByteArrayInputStream;
+import java.nio.charset.Charset;
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+
+@InputRequirement(InputRequirement.Requirement.INPUT_REQUIRED)
+@EventDriven
+@Tags({"rethinkdb", "get", "read"})
+@CapabilityDescription("Processor to get a JSON document from RethinkDB (https://www.rethinkdb.com/) using the document id. The FlowFile will contain the retrieved document")
+@WritesAttributes({
+ @WritesAttribute(attribute = GetRethinkDB.RETHINKDB_ERROR_MESSAGE, description = "RethinkDB error message"),
+ })
+@SeeAlso({PutRethinkDB.class})
+public class GetRethinkDB extends AbstractRethinkDBProcessor {
+
+ public static AllowableValue READ_MODE_SINGLE = new AllowableValue("single", "Single", "Read values from memory from primary replica (Default)");
+ public static AllowableValue READ_MODE_MAJORITY = new AllowableValue("majority", "Majority", "Read values commited to disk on majority of replicas");
--- End diff --
Spell checker says `committed` instead of `commited`.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] nifi issue #2012: NIFI-4188 - Nifi RethinkDB Get processor
Posted by jvwing <gi...@git.apache.org>.
Github user jvwing commented on the issue:
https://github.com/apache/nifi/pull/2012
Reviewing...
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] nifi pull request #2012: NIFI-4188 - Nifi RethinkDB Get processor
Posted by jvwing <gi...@git.apache.org>.
Github user jvwing commented on a diff in the pull request:
https://github.com/apache/nifi/pull/2012#discussion_r127879683
--- Diff: nifi-nar-bundles/nifi-rethinkdb-bundle/nifi-rethinkdb-processors/src/main/java/org/apache/nifi/processors/rethinkdb/GetRethinkDB.java ---
@@ -0,0 +1,197 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.nifi.processors.rethinkdb;
+
+import org.apache.commons.lang3.StringUtils;
+import org.apache.nifi.annotation.behavior.EventDriven;
+import org.apache.nifi.annotation.behavior.InputRequirement;
+import org.apache.nifi.annotation.behavior.WritesAttribute;
+import org.apache.nifi.annotation.behavior.WritesAttributes;
+import org.apache.nifi.annotation.documentation.CapabilityDescription;
+import org.apache.nifi.annotation.documentation.SeeAlso;
+import org.apache.nifi.annotation.documentation.Tags;
+import org.apache.nifi.annotation.lifecycle.OnScheduled;
+import org.apache.nifi.annotation.lifecycle.OnStopped;
+import org.apache.nifi.components.AllowableValue;
+import org.apache.nifi.components.PropertyDescriptor;
+import org.apache.nifi.expression.AttributeExpression.ResultType;
+import org.apache.nifi.flowfile.FlowFile;
+import org.apache.nifi.processor.ProcessContext;
+import org.apache.nifi.processor.ProcessSession;
+import org.apache.nifi.processor.Relationship;
+import org.apache.nifi.processor.exception.ProcessException;
+import org.apache.nifi.processor.util.StandardValidators;
+
+import com.google.gson.Gson;
+
+import java.io.ByteArrayInputStream;
+import java.nio.charset.Charset;
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+
+@InputRequirement(InputRequirement.Requirement.INPUT_REQUIRED)
+@EventDriven
+@Tags({"rethinkdb", "get", "read"})
+@CapabilityDescription("Processor to get a JSON document from RethinkDB (https://www.rethinkdb.com/) using the document id. The FlowFile will contain the retrieved document")
+@WritesAttributes({
+ @WritesAttribute(attribute = GetRethinkDB.RETHINKDB_ERROR_MESSAGE, description = "RethinkDB error message"),
+ })
+@SeeAlso({PutRethinkDB.class})
+public class GetRethinkDB extends AbstractRethinkDBProcessor {
+
+ public static AllowableValue READ_MODE_SINGLE = new AllowableValue("single", "Single", "Read values from memory from primary replica (Default)");
+ public static AllowableValue READ_MODE_MAJORITY = new AllowableValue("majority", "Majority", "Read values commited to disk on majority of replicas");
+ public static AllowableValue READ_MODE_OUTDATED = new AllowableValue("outdated", "Outdated", "Read values from memory from an arbitrary replica ");
+
+ protected static final PropertyDescriptor READ_MODE = new PropertyDescriptor.Builder()
+ .name("rethinkdb-read-mode")
+ .displayName("Read Mode")
+ .description("Read mode used for consistency")
+ .required(true)
+ .defaultValue(READ_MODE_SINGLE.getValue())
+ .allowableValues(READ_MODE_SINGLE, READ_MODE_MAJORITY, READ_MODE_OUTDATED)
+ .expressionLanguageSupported(true)
+ .build();
+
+ public static final PropertyDescriptor RETHINKDB_DOCUMENT_ID = new PropertyDescriptor.Builder()
+ .displayName("Document Identifier")
+ .name("rethinkdb-document-identifier")
+ .description("A FlowFile attribute, or attribute expression used " +
+ "for determining RethinkDB key for the Flow File content")
+ .required(true)
+ .addValidator(StandardValidators.createAttributeExpressionLanguageValidator(ResultType.STRING, true))
+ .expressionLanguageSupported(true)
+ .build();
+
+ protected String READ_MODE_KEY = "read_mode";
+
+ private static final Set<Relationship> relationships;
+ private static final List<PropertyDescriptor> propertyDescriptors;
+
+ public static final String RETHINKDB_ERROR_MESSAGE = "rethinkdb.error.message";
--- End diff --
Should RETHINKDB_ERROR_MESSAGE be shared in AbstractRethinkDBProcessor now that it is used by both processors?
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] nifi issue #2012: NIFI-4188 - Nifi RethinkDB Get processor
Posted by mans2singh <gi...@git.apache.org>.
Github user mans2singh commented on the issue:
https://github.com/apache/nifi/pull/2012
Hello @jvwing - I've updated the code based on your review comments (updated notice, refactored constants to base class).
Regarding the name Get vs Fetch - I am trying to use the terminology used by RethinkDB and their apis. If you think that Fetch is more expressive and I can change it. Also, I've added tag 'fetch' so that the processor will turn up in searches.
I was interpolating id in case of empty id is because sometimes it might be null. So, I've updated the code to check blank rather than just empty and have also added unit tests for those scenarios.
Regarding 'not_found' relation - I wanted to give the users option to handle this scenario separately than other errors. From what I understand, the user may connect both 'failure' and 'not_found' to same component without losing any flexibility.
Please let me know if you have any other recommendations/thoughts.
Thanks again.
Mans
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] nifi issue #2012: NIFI-4188 - Nifi RethinkDB Get processor
Posted by jvwing <gi...@git.apache.org>.
Github user jvwing commented on the issue:
https://github.com/apache/nifi/pull/2012
I'm OK with the name "GetRethinkDB". Thanks for the tagging.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] nifi issue #2012: NIFI-4188 - Nifi RethinkDB Get processor
Posted by mans2singh <gi...@git.apache.org>.
Github user mans2singh commented on the issue:
https://github.com/apache/nifi/pull/2012
Thanks @jvwing for your advice and guidance.
Mans
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] nifi pull request #2012: NIFI-4188 - Nifi RethinkDB Get processor
Posted by jvwing <gi...@git.apache.org>.
Github user jvwing commented on a diff in the pull request:
https://github.com/apache/nifi/pull/2012#discussion_r127880998
--- Diff: nifi-nar-bundles/nifi-rethinkdb-bundle/nifi-rethinkdb-processors/src/main/java/org/apache/nifi/processors/rethinkdb/GetRethinkDB.java ---
@@ -0,0 +1,197 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.nifi.processors.rethinkdb;
+
+import org.apache.commons.lang3.StringUtils;
+import org.apache.nifi.annotation.behavior.EventDriven;
+import org.apache.nifi.annotation.behavior.InputRequirement;
+import org.apache.nifi.annotation.behavior.WritesAttribute;
+import org.apache.nifi.annotation.behavior.WritesAttributes;
+import org.apache.nifi.annotation.documentation.CapabilityDescription;
+import org.apache.nifi.annotation.documentation.SeeAlso;
+import org.apache.nifi.annotation.documentation.Tags;
+import org.apache.nifi.annotation.lifecycle.OnScheduled;
+import org.apache.nifi.annotation.lifecycle.OnStopped;
+import org.apache.nifi.components.AllowableValue;
+import org.apache.nifi.components.PropertyDescriptor;
+import org.apache.nifi.expression.AttributeExpression.ResultType;
+import org.apache.nifi.flowfile.FlowFile;
+import org.apache.nifi.processor.ProcessContext;
+import org.apache.nifi.processor.ProcessSession;
+import org.apache.nifi.processor.Relationship;
+import org.apache.nifi.processor.exception.ProcessException;
+import org.apache.nifi.processor.util.StandardValidators;
+
+import com.google.gson.Gson;
+
+import java.io.ByteArrayInputStream;
+import java.nio.charset.Charset;
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+
+@InputRequirement(InputRequirement.Requirement.INPUT_REQUIRED)
+@EventDriven
+@Tags({"rethinkdb", "get", "read"})
+@CapabilityDescription("Processor to get a JSON document from RethinkDB (https://www.rethinkdb.com/) using the document id. The FlowFile will contain the retrieved document")
+@WritesAttributes({
+ @WritesAttribute(attribute = GetRethinkDB.RETHINKDB_ERROR_MESSAGE, description = "RethinkDB error message"),
+ })
+@SeeAlso({PutRethinkDB.class})
+public class GetRethinkDB extends AbstractRethinkDBProcessor {
+
+ public static AllowableValue READ_MODE_SINGLE = new AllowableValue("single", "Single", "Read values from memory from primary replica (Default)");
+ public static AllowableValue READ_MODE_MAJORITY = new AllowableValue("majority", "Majority", "Read values commited to disk on majority of replicas");
+ public static AllowableValue READ_MODE_OUTDATED = new AllowableValue("outdated", "Outdated", "Read values from memory from an arbitrary replica ");
+
+ protected static final PropertyDescriptor READ_MODE = new PropertyDescriptor.Builder()
+ .name("rethinkdb-read-mode")
+ .displayName("Read Mode")
+ .description("Read mode used for consistency")
+ .required(true)
+ .defaultValue(READ_MODE_SINGLE.getValue())
+ .allowableValues(READ_MODE_SINGLE, READ_MODE_MAJORITY, READ_MODE_OUTDATED)
+ .expressionLanguageSupported(true)
+ .build();
+
+ public static final PropertyDescriptor RETHINKDB_DOCUMENT_ID = new PropertyDescriptor.Builder()
+ .displayName("Document Identifier")
+ .name("rethinkdb-document-identifier")
+ .description("A FlowFile attribute, or attribute expression used " +
+ "for determining RethinkDB key for the Flow File content")
+ .required(true)
+ .addValidator(StandardValidators.createAttributeExpressionLanguageValidator(ResultType.STRING, true))
+ .expressionLanguageSupported(true)
+ .build();
+
+ protected String READ_MODE_KEY = "read_mode";
+
+ private static final Set<Relationship> relationships;
+ private static final List<PropertyDescriptor> propertyDescriptors;
+
+ public static final String RETHINKDB_ERROR_MESSAGE = "rethinkdb.error.message";
+
+ protected Gson gson = new Gson();
+
+ static {
+ final Set<Relationship> tempRelationships = new HashSet<>();
+ tempRelationships.add(REL_SUCCESS);
+ tempRelationships.add(REL_FAILURE);
+ tempRelationships.add(REL_NOT_FOUND);
+ relationships = Collections.unmodifiableSet(tempRelationships);
+
+ final List<PropertyDescriptor> tempDescriptors = new ArrayList<>();
+ tempDescriptors.add(DB_NAME);
+ tempDescriptors.add(DB_HOST);
+ tempDescriptors.add(DB_PORT);
+ tempDescriptors.add(USERNAME);
+ tempDescriptors.add(PASSWORD);
+ tempDescriptors.add(TABLE_NAME);
+ tempDescriptors.add(CHARSET);
+ tempDescriptors.add(RETHINKDB_DOCUMENT_ID);
+ tempDescriptors.add(READ_MODE);
+ tempDescriptors.add(MAX_DOCUMENTS_SIZE);
+ propertyDescriptors = Collections.unmodifiableList(tempDescriptors);
+ }
+
+ @Override
+ public Set<Relationship> getRelationships() {
+ return relationships;
+ }
+
+ @Override
+ public final List<PropertyDescriptor> getSupportedPropertyDescriptors() {
+ return propertyDescriptors;
+ }
+
+ @OnScheduled
+ public void onScheduled(final ProcessContext context) {
+ super.onScheduled(context);
+ }
+
+ @Override
+ public void onTrigger(final ProcessContext context, final ProcessSession session) throws ProcessException {
+ FlowFile flowFile = session.get();
+ if (flowFile == null) {
+ return;
+ }
+
+ Charset charset = Charset.forName(context.getProperty(CHARSET).evaluateAttributeExpressions(flowFile).getValue());
+ String id = context.getProperty(RETHINKDB_DOCUMENT_ID).evaluateAttributeExpressions(flowFile).getValue();
+ String readMode = context.getProperty(READ_MODE).evaluateAttributeExpressions(flowFile).getValue();
+
+ if ( StringUtils.isEmpty(id) ) {
+ getLogger().error("Empty id '" + id + "'");
+ flowFile = session.putAttribute(flowFile, RETHINKDB_ERROR_MESSAGE, "Empty id '" + id + "'");
+ session.transfer(flowFile, REL_FAILURE);
+ return;
+ }
+
+ try {
+ long startTimeMillis = System.currentTimeMillis();
+ Map<String,Object> document = getDocument(id, readMode);
+
+ if ( document == null ) {
+ getLogger().error("Document not found for id " + id);
+ flowFile = session.putAttribute(flowFile, RETHINKDB_ERROR_MESSAGE, "Document with " + id + " not found");
--- End diff --
Did you mean something like `"Document with id '" + id + "' not found"` similar to the log message above?
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] nifi issue #2012: NIFI-4188 - Nifi RethinkDB Get processor
Posted by mans2singh <gi...@git.apache.org>.
Github user mans2singh commented on the issue:
https://github.com/apache/nifi/pull/2012
Hi @jvwing -
Thanks for your review.
I've updated the code based on your review (changed logging level for not_found to debug and id check to empty + updates to error message and tests).
Please let me know if there is any other recommendation.
Mans
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] nifi issue #2012: NIFI-4188 - Nifi RethinkDB Get processor
Posted by jvwing <gi...@git.apache.org>.
Github user jvwing commented on the issue:
https://github.com/apache/nifi/pull/2012
Thanks for those updates, @mans2singh, everything looks good. I'll merge shortly.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] nifi pull request #2012: NIFI-4188 - Nifi RethinkDB Get processor
Posted by asfgit <gi...@git.apache.org>.
Github user asfgit closed the pull request at:
https://github.com/apache/nifi/pull/2012
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] nifi pull request #2012: NIFI-4188 - Nifi RethinkDB Get processor
Posted by jvwing <gi...@git.apache.org>.
Github user jvwing commented on a diff in the pull request:
https://github.com/apache/nifi/pull/2012#discussion_r127880492
--- Diff: nifi-nar-bundles/nifi-rethinkdb-bundle/nifi-rethinkdb-processors/src/main/java/org/apache/nifi/processors/rethinkdb/GetRethinkDB.java ---
@@ -0,0 +1,197 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.nifi.processors.rethinkdb;
+
+import org.apache.commons.lang3.StringUtils;
+import org.apache.nifi.annotation.behavior.EventDriven;
+import org.apache.nifi.annotation.behavior.InputRequirement;
+import org.apache.nifi.annotation.behavior.WritesAttribute;
+import org.apache.nifi.annotation.behavior.WritesAttributes;
+import org.apache.nifi.annotation.documentation.CapabilityDescription;
+import org.apache.nifi.annotation.documentation.SeeAlso;
+import org.apache.nifi.annotation.documentation.Tags;
+import org.apache.nifi.annotation.lifecycle.OnScheduled;
+import org.apache.nifi.annotation.lifecycle.OnStopped;
+import org.apache.nifi.components.AllowableValue;
+import org.apache.nifi.components.PropertyDescriptor;
+import org.apache.nifi.expression.AttributeExpression.ResultType;
+import org.apache.nifi.flowfile.FlowFile;
+import org.apache.nifi.processor.ProcessContext;
+import org.apache.nifi.processor.ProcessSession;
+import org.apache.nifi.processor.Relationship;
+import org.apache.nifi.processor.exception.ProcessException;
+import org.apache.nifi.processor.util.StandardValidators;
+
+import com.google.gson.Gson;
+
+import java.io.ByteArrayInputStream;
+import java.nio.charset.Charset;
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+
+@InputRequirement(InputRequirement.Requirement.INPUT_REQUIRED)
+@EventDriven
+@Tags({"rethinkdb", "get", "read"})
+@CapabilityDescription("Processor to get a JSON document from RethinkDB (https://www.rethinkdb.com/) using the document id. The FlowFile will contain the retrieved document")
+@WritesAttributes({
+ @WritesAttribute(attribute = GetRethinkDB.RETHINKDB_ERROR_MESSAGE, description = "RethinkDB error message"),
+ })
+@SeeAlso({PutRethinkDB.class})
+public class GetRethinkDB extends AbstractRethinkDBProcessor {
+
+ public static AllowableValue READ_MODE_SINGLE = new AllowableValue("single", "Single", "Read values from memory from primary replica (Default)");
+ public static AllowableValue READ_MODE_MAJORITY = new AllowableValue("majority", "Majority", "Read values commited to disk on majority of replicas");
+ public static AllowableValue READ_MODE_OUTDATED = new AllowableValue("outdated", "Outdated", "Read values from memory from an arbitrary replica ");
+
+ protected static final PropertyDescriptor READ_MODE = new PropertyDescriptor.Builder()
+ .name("rethinkdb-read-mode")
+ .displayName("Read Mode")
+ .description("Read mode used for consistency")
+ .required(true)
+ .defaultValue(READ_MODE_SINGLE.getValue())
+ .allowableValues(READ_MODE_SINGLE, READ_MODE_MAJORITY, READ_MODE_OUTDATED)
+ .expressionLanguageSupported(true)
+ .build();
+
+ public static final PropertyDescriptor RETHINKDB_DOCUMENT_ID = new PropertyDescriptor.Builder()
+ .displayName("Document Identifier")
+ .name("rethinkdb-document-identifier")
+ .description("A FlowFile attribute, or attribute expression used " +
+ "for determining RethinkDB key for the Flow File content")
+ .required(true)
+ .addValidator(StandardValidators.createAttributeExpressionLanguageValidator(ResultType.STRING, true))
+ .expressionLanguageSupported(true)
+ .build();
+
+ protected String READ_MODE_KEY = "read_mode";
+
+ private static final Set<Relationship> relationships;
+ private static final List<PropertyDescriptor> propertyDescriptors;
+
+ public static final String RETHINKDB_ERROR_MESSAGE = "rethinkdb.error.message";
+
+ protected Gson gson = new Gson();
+
+ static {
+ final Set<Relationship> tempRelationships = new HashSet<>();
+ tempRelationships.add(REL_SUCCESS);
+ tempRelationships.add(REL_FAILURE);
+ tempRelationships.add(REL_NOT_FOUND);
+ relationships = Collections.unmodifiableSet(tempRelationships);
+
+ final List<PropertyDescriptor> tempDescriptors = new ArrayList<>();
+ tempDescriptors.add(DB_NAME);
+ tempDescriptors.add(DB_HOST);
+ tempDescriptors.add(DB_PORT);
+ tempDescriptors.add(USERNAME);
+ tempDescriptors.add(PASSWORD);
+ tempDescriptors.add(TABLE_NAME);
+ tempDescriptors.add(CHARSET);
+ tempDescriptors.add(RETHINKDB_DOCUMENT_ID);
+ tempDescriptors.add(READ_MODE);
+ tempDescriptors.add(MAX_DOCUMENTS_SIZE);
+ propertyDescriptors = Collections.unmodifiableList(tempDescriptors);
+ }
+
+ @Override
+ public Set<Relationship> getRelationships() {
+ return relationships;
+ }
+
+ @Override
+ public final List<PropertyDescriptor> getSupportedPropertyDescriptors() {
+ return propertyDescriptors;
+ }
+
+ @OnScheduled
+ public void onScheduled(final ProcessContext context) {
+ super.onScheduled(context);
+ }
+
+ @Override
+ public void onTrigger(final ProcessContext context, final ProcessSession session) throws ProcessException {
+ FlowFile flowFile = session.get();
+ if (flowFile == null) {
+ return;
+ }
+
+ Charset charset = Charset.forName(context.getProperty(CHARSET).evaluateAttributeExpressions(flowFile).getValue());
+ String id = context.getProperty(RETHINKDB_DOCUMENT_ID).evaluateAttributeExpressions(flowFile).getValue();
+ String readMode = context.getProperty(READ_MODE).evaluateAttributeExpressions(flowFile).getValue();
+
+ if ( StringUtils.isEmpty(id) ) {
+ getLogger().error("Empty id '" + id + "'");
+ flowFile = session.putAttribute(flowFile, RETHINKDB_ERROR_MESSAGE, "Empty id '" + id + "'");
--- End diff --
Do we need to interpolate `id` here after we know it is an empty string? Your unit test testBlankId() suggests that the result will just be single quotes.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] nifi pull request #2012: NIFI-4188 - Nifi RethinkDB Get processor
Posted by jvwing <gi...@git.apache.org>.
Github user jvwing commented on a diff in the pull request:
https://github.com/apache/nifi/pull/2012#discussion_r127882745
--- Diff: nifi-nar-bundles/nifi-rethinkdb-bundle/nifi-rethinkdb-processors/src/main/java/org/apache/nifi/processors/rethinkdb/GetRethinkDB.java ---
@@ -0,0 +1,197 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.nifi.processors.rethinkdb;
+
+import org.apache.commons.lang3.StringUtils;
+import org.apache.nifi.annotation.behavior.EventDriven;
+import org.apache.nifi.annotation.behavior.InputRequirement;
+import org.apache.nifi.annotation.behavior.WritesAttribute;
+import org.apache.nifi.annotation.behavior.WritesAttributes;
+import org.apache.nifi.annotation.documentation.CapabilityDescription;
+import org.apache.nifi.annotation.documentation.SeeAlso;
+import org.apache.nifi.annotation.documentation.Tags;
+import org.apache.nifi.annotation.lifecycle.OnScheduled;
+import org.apache.nifi.annotation.lifecycle.OnStopped;
+import org.apache.nifi.components.AllowableValue;
+import org.apache.nifi.components.PropertyDescriptor;
+import org.apache.nifi.expression.AttributeExpression.ResultType;
+import org.apache.nifi.flowfile.FlowFile;
+import org.apache.nifi.processor.ProcessContext;
+import org.apache.nifi.processor.ProcessSession;
+import org.apache.nifi.processor.Relationship;
+import org.apache.nifi.processor.exception.ProcessException;
+import org.apache.nifi.processor.util.StandardValidators;
+
+import com.google.gson.Gson;
+
+import java.io.ByteArrayInputStream;
+import java.nio.charset.Charset;
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+
+@InputRequirement(InputRequirement.Requirement.INPUT_REQUIRED)
+@EventDriven
+@Tags({"rethinkdb", "get", "read"})
+@CapabilityDescription("Processor to get a JSON document from RethinkDB (https://www.rethinkdb.com/) using the document id. The FlowFile will contain the retrieved document")
+@WritesAttributes({
+ @WritesAttribute(attribute = GetRethinkDB.RETHINKDB_ERROR_MESSAGE, description = "RethinkDB error message"),
+ })
+@SeeAlso({PutRethinkDB.class})
+public class GetRethinkDB extends AbstractRethinkDBProcessor {
+
+ public static AllowableValue READ_MODE_SINGLE = new AllowableValue("single", "Single", "Read values from memory from primary replica (Default)");
+ public static AllowableValue READ_MODE_MAJORITY = new AllowableValue("majority", "Majority", "Read values commited to disk on majority of replicas");
+ public static AllowableValue READ_MODE_OUTDATED = new AllowableValue("outdated", "Outdated", "Read values from memory from an arbitrary replica ");
+
+ protected static final PropertyDescriptor READ_MODE = new PropertyDescriptor.Builder()
+ .name("rethinkdb-read-mode")
+ .displayName("Read Mode")
+ .description("Read mode used for consistency")
+ .required(true)
+ .defaultValue(READ_MODE_SINGLE.getValue())
+ .allowableValues(READ_MODE_SINGLE, READ_MODE_MAJORITY, READ_MODE_OUTDATED)
+ .expressionLanguageSupported(true)
+ .build();
+
+ public static final PropertyDescriptor RETHINKDB_DOCUMENT_ID = new PropertyDescriptor.Builder()
+ .displayName("Document Identifier")
+ .name("rethinkdb-document-identifier")
+ .description("A FlowFile attribute, or attribute expression used " +
+ "for determining RethinkDB key for the Flow File content")
+ .required(true)
+ .addValidator(StandardValidators.createAttributeExpressionLanguageValidator(ResultType.STRING, true))
+ .expressionLanguageSupported(true)
+ .build();
+
+ protected String READ_MODE_KEY = "read_mode";
+
+ private static final Set<Relationship> relationships;
+ private static final List<PropertyDescriptor> propertyDescriptors;
+
+ public static final String RETHINKDB_ERROR_MESSAGE = "rethinkdb.error.message";
+
+ protected Gson gson = new Gson();
+
+ static {
+ final Set<Relationship> tempRelationships = new HashSet<>();
+ tempRelationships.add(REL_SUCCESS);
+ tempRelationships.add(REL_FAILURE);
+ tempRelationships.add(REL_NOT_FOUND);
+ relationships = Collections.unmodifiableSet(tempRelationships);
+
+ final List<PropertyDescriptor> tempDescriptors = new ArrayList<>();
+ tempDescriptors.add(DB_NAME);
+ tempDescriptors.add(DB_HOST);
+ tempDescriptors.add(DB_PORT);
+ tempDescriptors.add(USERNAME);
+ tempDescriptors.add(PASSWORD);
+ tempDescriptors.add(TABLE_NAME);
+ tempDescriptors.add(CHARSET);
+ tempDescriptors.add(RETHINKDB_DOCUMENT_ID);
+ tempDescriptors.add(READ_MODE);
+ tempDescriptors.add(MAX_DOCUMENTS_SIZE);
+ propertyDescriptors = Collections.unmodifiableList(tempDescriptors);
+ }
+
+ @Override
+ public Set<Relationship> getRelationships() {
+ return relationships;
+ }
+
+ @Override
+ public final List<PropertyDescriptor> getSupportedPropertyDescriptors() {
+ return propertyDescriptors;
+ }
+
+ @OnScheduled
+ public void onScheduled(final ProcessContext context) {
+ super.onScheduled(context);
+ }
+
+ @Override
+ public void onTrigger(final ProcessContext context, final ProcessSession session) throws ProcessException {
+ FlowFile flowFile = session.get();
+ if (flowFile == null) {
+ return;
+ }
+
+ Charset charset = Charset.forName(context.getProperty(CHARSET).evaluateAttributeExpressions(flowFile).getValue());
+ String id = context.getProperty(RETHINKDB_DOCUMENT_ID).evaluateAttributeExpressions(flowFile).getValue();
+ String readMode = context.getProperty(READ_MODE).evaluateAttributeExpressions(flowFile).getValue();
+
+ if ( StringUtils.isEmpty(id) ) {
+ getLogger().error("Empty id '" + id + "'");
+ flowFile = session.putAttribute(flowFile, RETHINKDB_ERROR_MESSAGE, "Empty id '" + id + "'");
+ session.transfer(flowFile, REL_FAILURE);
+ return;
+ }
+
+ try {
+ long startTimeMillis = System.currentTimeMillis();
+ Map<String,Object> document = getDocument(id, readMode);
+
+ if ( document == null ) {
+ getLogger().error("Document not found for id " + id);
--- End diff --
Is not found an error? It seems possible that this could happen at high volume. Given that it has it's own relationship, and there is a separate relationship for failure, didn't this "succeed" in not finding the key? I don't have a strong opinion, I'll accept it as-is if you believe in it.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---