You are viewing a plain text version of this content. The canonical link for it is here.
Posted to issues@nifi.apache.org by GitBox <gi...@apache.org> on 2020/11/30 16:06:58 UTC
[GitHub] [nifi] ChrisSamo632 opened a new pull request #4693: NIFI-8001 Dynamic Properties as Request Parameters for Elasticsearch β¦
ChrisSamo632 opened a new pull request #4693:
URL: https://github.com/apache/nifi/pull/4693
β¦Client Service processors; NIFI-8003 UpdateByQueryElasticsearchProcessor; addressed various warnings and inefficiencies found in existing processor code
#### Description of PR
NIFI-8001 Dynamic Properties can be used to specify request (query string) parameters through ElasticsearchClientService processors (e.g. PutElasticsearchRecord)
NIFI-8003 UpdateByQueryElasticsearch (based on existing DeleteByQueryElasticsearch) processor
Addressed several warnings/inefficiencies within the existing processor/service implementations and fixed previously broken ES Integration tests (maven-failsafe-plugin 3-M5 doesn't currently work for these tests due to a classpath loading issue expected to be fixed in the upcoming -M6, so reverted to -M3 for the time being; also some of the integration tests didn't work when run together as one test would change the data that a subsequent test expected to be in its original state).
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 `main`)?
- [x] Is your initial contribution a single, squashed commit? _Additional commits in response to PR reviewer feedback should be made on this branch and pushed to allow change tracking. Do not `squash` or use `--force` when pushing to allow for clean monitoring of changes._
### 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?
- [ ] Have you verified that the full build is successful on JDK 8?
- [x] Have you verified that the full build is successful on JDK 11?
- [ ] 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`?
- [ ] 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 GitHub Actions CI for build issues and submit an update to your PR as soon as possible.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [nifi] ChrisSamo632 commented on a change in pull request #4693: NIFI-8001 Elasticsearch Dynamic Parameters
Posted by GitBox <gi...@apache.org>.
ChrisSamo632 commented on a change in pull request #4693:
URL: https://github.com/apache/nifi/pull/4693#discussion_r724354340
##########
File path: nifi-nar-bundles/nifi-elasticsearch-bundle/nifi-elasticsearch-client-service/src/main/java/org/apache/nifi/elasticsearch/ElasticSearchClientServiceImpl.java
##########
@@ -182,8 +181,8 @@ private Response runQuery(final String endpoint, final String query, final Strin
final HttpEntity queryEntity = new NStringEntity(query, ContentType.APPLICATION_JSON);
try {
- return client.performRequest("POST", sb.toString(), requestParameters != null ? requestParameters : Collections.emptyMap(), queryEntity);
- } catch (final Exception e) {
+ return performRequest("POST", sb.toString(), requestParameters, queryEntity);
Review comment:
Not sure what you want to be `final` here? I've applied `final` to everywhere I can see it may be usable within the class (there were a couple of things I'd missed previously, but this line isn't assigning any variables, so there's nothing to make `final`)?
##########
File path: nifi-nar-bundles/nifi-elasticsearch-bundle/nifi-elasticsearch-client-service/src/main/java/org/apache/nifi/elasticsearch/ElasticSearchClientServiceImpl.java
##########
@@ -359,9 +358,21 @@ public DeleteOperationResponse deleteByQuery(final String query, final String in
return new DeleteOperationResponse(watch.getDuration(TimeUnit.MILLISECONDS));
}
+
+ public UpdateOperationResponse updateByQuery(final String query, final String index, final String type, final Map<String, String> requestParameters) {
+ long start = System.currentTimeMillis();
+ Response response = runQuery("_update_by_query", query, index, type, requestParameters);
+ long end = System.currentTimeMillis();
Review comment:
Think this was an existing thing, I generally don't like aligning things in thiw way either, so happy to change π€·
##########
File path: nifi-nar-bundles/nifi-elasticsearch-bundle/nifi-elasticsearch-restapi-processors/src/main/java/org/apache/nifi/processors/elasticsearch/PutElasticsearchRecord.java
##########
@@ -414,8 +434,8 @@ private void removeBadRecordFlowFiles(List<FlowFile> bad, ProcessSession session
bad.clear();
}
- private FlowFile indexDocuments(BulkOperation bundle, ProcessSession session, FlowFile input) throws Exception {
- IndexOperationResponse response = clientService.bulk(bundle.getOperationList());
+ private FlowFile indexDocuments(BulkOperation bundle, ProcessContext context, ProcessSession session, FlowFile input) throws Exception {
+ IndexOperationResponse response = clientService.bulk(bundle.getOperationList(), getUrlQueryParameters(context, input));
Review comment:
`final`s, `final`s everywhere π
##########
File path: nifi-nar-bundles/nifi-elasticsearch-bundle/nifi-elasticsearch-restapi-processors/src/main/java/org/apache/nifi/processors/elasticsearch/SearchElasticsearch.java
##########
@@ -65,6 +66,11 @@
"Search After/Point in Time queries must include a valid \"sort\" field. The processor will retrieve multiple pages of results " +
"until either no more results are available or the Pagination Keep Alive expiration is reached, after which the query will " +
"restart with the first page of results being retrieved.")
+@DynamicProperty(
+ name = "A URL query parameter",
+ value = "The value to set it to",
+ expressionLanguageScope = ExpressionLanguageScope.FLOWFILE_ATTRIBUTES,
+ description = "Adds the specified property name/value as a query parameter in the Elasticsearch URL used for processing")
Review comment:
I'm confused, which other comments are you referring to (I can't make this `final` π)?
##########
File path: nifi-nar-bundles/nifi-elasticsearch-bundle/nifi-elasticsearch-restapi-processors/src/main/java/org/apache/nifi/processors/elasticsearch/UpdateByQueryElasticsearch.java
##########
@@ -0,0 +1,64 @@
+/*
+ * 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.elasticsearch;
+
+import org.apache.nifi.annotation.behavior.DynamicProperty;
+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.Tags;
+import org.apache.nifi.elasticsearch.ElasticSearchClientService;
+import org.apache.nifi.elasticsearch.OperationResponse;
+import org.apache.nifi.expression.ExpressionLanguageScope;
+
+import java.util.Map;
+
+@WritesAttributes({
+ @WritesAttribute(attribute = "elasticsearch.update.took", description = "The amount of time that it took to complete the update operation in ms."),
+ @WritesAttribute(attribute = "elasticsearch.update.error", description = "The error message provided by Elasticsearch if there is an error running the update.")
+})
+@InputRequirement(InputRequirement.Requirement.INPUT_ALLOWED)
+@Tags({ "elastic", "elasticsearch", "update", "query"})
+@CapabilityDescription("Update documents in an Elasticsearch index using a query. The query can be loaded from a flowfile body " +
+ "or from the Query parameter.")
+@DynamicProperty(
+ name = "A URL query parameter",
+ value = "The value to set it to",
+ expressionLanguageScope = ExpressionLanguageScope.FLOWFILE_ATTRIBUTES,
Review comment:
Which other comments?
##########
File path: nifi-nar-bundles/nifi-elasticsearch-bundle/nifi-elasticsearch-restapi-processors/src/test/groovy/org/apache/nifi/processors/elasticsearch/AbstractByQueryElasticsearchTest.groovy
##########
@@ -0,0 +1,218 @@
+/*
+ * 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.elasticsearch
+
+import org.apache.nifi.util.MockFlowFile
+import org.apache.nifi.util.TestRunner
+import org.junit.Assert
+import org.junit.Test
+
+abstract class AbstractByQueryElasticsearchTest {
Review comment:
While it should be fairly straight forward to convert them, I think that should be part of a larger initiative if/when decided upon. Also, we couldn't convert just this one class, we'd need to convert all of the tests (pretty much) in this package.
Also, currently, all the tests in the package are Groovy and not Java, I think moving just a sub-set of the tests would lead to more confusion rather than solving anything.
Maybe we should raise a follow-on Jira ticket to suggest the move and see what people think?
##########
File path: nifi-nar-bundles/nifi-elasticsearch-bundle/nifi-elasticsearch-restapi-processors/src/main/java/org/apache/nifi/processors/elasticsearch/SearchElasticsearch.java
##########
@@ -65,6 +66,11 @@
"Search After/Point in Time queries must include a valid \"sort\" field. The processor will retrieve multiple pages of results " +
"until either no more results are available or the Pagination Keep Alive expiration is reached, after which the query will " +
"restart with the first page of results being retrieved.")
+@DynamicProperty(
+ name = "A URL query parameter",
+ value = "The value to set it to",
+ expressionLanguageScope = ExpressionLanguageScope.FLOWFILE_ATTRIBUTES,
+ description = "Adds the specified property name/value as a query parameter in the Elasticsearch URL used for processing")
Review comment:
Ah wait, I'd missed a load of comments (GitHub hides a lot by default where there are many comments... I'll look through the rest shortly)
##########
File path: nifi-nar-bundles/nifi-elasticsearch-bundle/nifi-elasticsearch-restapi-processors/src/main/java/org/apache/nifi/processors/elasticsearch/SearchElasticsearch.java
##########
@@ -65,6 +66,11 @@
"Search After/Point in Time queries must include a valid \"sort\" field. The processor will retrieve multiple pages of results " +
"until either no more results are available or the Pagination Keep Alive expiration is reached, after which the query will " +
"restart with the first page of results being retrieved.")
+@DynamicProperty(
+ name = "A URL query parameter",
+ value = "The value to set it to",
+ expressionLanguageScope = ExpressionLanguageScope.FLOWFILE_ATTRIBUTES,
+ description = "Adds the specified property name/value as a query parameter in the Elasticsearch URL used for processing")
Review comment:
Found the relevant comments, addressed
##########
File path: nifi-nar-bundles/nifi-elasticsearch-bundle/nifi-elasticsearch-restapi-processors/src/test/groovy/org/apache/nifi/processors/elasticsearch/AbstractByQueryElasticsearchTest.groovy
##########
@@ -0,0 +1,218 @@
+/*
+ * 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.elasticsearch
+
+import org.apache.nifi.util.MockFlowFile
+import org.apache.nifi.util.TestRunner
+import org.junit.Assert
+import org.junit.Test
+
+abstract class AbstractByQueryElasticsearchTest {
Review comment:
[NIFI-9290](https://issues.apache.org/jira/browse/NIFI-9290) raised to cover this consideration
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@nifi.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [nifi] ChrisSamo632 commented on a change in pull request #4693: NIFI-8001 Elasticsearch Dynamic Parameters
Posted by GitBox <gi...@apache.org>.
ChrisSamo632 commented on a change in pull request #4693:
URL: https://github.com/apache/nifi/pull/4693#discussion_r724364848
##########
File path: nifi-nar-bundles/nifi-elasticsearch-bundle/nifi-elasticsearch-client-service/src/main/java/org/apache/nifi/elasticsearch/ElasticSearchClientServiceImpl.java
##########
@@ -359,9 +358,21 @@ public DeleteOperationResponse deleteByQuery(final String query, final String in
return new DeleteOperationResponse(watch.getDuration(TimeUnit.MILLISECONDS));
}
+
+ public UpdateOperationResponse updateByQuery(final String query, final String index, final String type, final Map<String, String> requestParameters) {
+ long start = System.currentTimeMillis();
+ Response response = runQuery("_update_by_query", query, index, type, requestParameters);
+ long end = System.currentTimeMillis();
Review comment:
Think this was an existing thing, I generally don't like aligning things in thiw way either, so happy to change π€·
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@nifi.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [nifi] ChrisSamo632 commented on pull request #4693: NIFI-8001 Elasticsearch Dynamic Parameters
Posted by GitBox <gi...@apache.org>.
ChrisSamo632 commented on pull request #4693:
URL: https://github.com/apache/nifi/pull/4693#issuecomment-869953462
Impacts https://github.com/apache/nifi/pull/5193 (NIFI-8002), which uses the same request parameters approach to control paginated queries. This PR should be rebased if the other is merged first (or vice versa).
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@nifi.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [nifi] ChrisSamo632 commented on a change in pull request #4693: NIFI-8001 Elasticsearch Dynamic Parameters
Posted by GitBox <gi...@apache.org>.
ChrisSamo632 commented on a change in pull request #4693:
URL: https://github.com/apache/nifi/pull/4693#discussion_r724795973
##########
File path: nifi-nar-bundles/nifi-elasticsearch-bundle/nifi-elasticsearch-restapi-processors/src/main/java/org/apache/nifi/processors/elasticsearch/PutElasticsearchRecord.java
##########
@@ -426,22 +429,21 @@ public void onTrigger(ProcessContext context, ProcessSession session) {
session.transfer(input, REL_SUCCESS);
}
- private void removeBadRecordFlowFiles(List<FlowFile> bad, ProcessSession session) {
- for (FlowFile badFlowFile : bad) {
+ private void removeBadRecordFlowFiles(final List<FlowFile> bad, final ProcessSession session) {
+ for (final FlowFile badFlowFile : bad) {
session.remove(badFlowFile);
}
bad.clear();
}
- private FlowFile indexDocuments(BulkOperation bundle, ProcessContext context, ProcessSession session, FlowFile input) throws Exception {
- IndexOperationResponse response = clientService.bulk(bundle.getOperationList(), getUrlQueryParameters(context, input));
+ private FlowFile indexDocuments(final BulkOperation bundle, final ProcessContext context, final ProcessSession session, final FlowFile input) throws Exception {
+ final IndexOperationResponse response = clientService.bulk(bundle.getOperationList(), getUrlQueryParameters(context, input));
if (response.hasErrors()) {
if(logErrors || getLogger().isDebugEnabled()) {
- List<Map<String, Object>> errors = response.getItems();
- ObjectMapper mapper = new ObjectMapper();
- mapper.enable(SerializationFeature.INDENT_OUTPUT);
- String output = String.format("An error was encountered while processing bulk operations. Server response below:%n%n%s", mapper.writeValueAsString(errors));
+ final List<Map<String, Object>> errors = response.getItems();
+ objectMapper.enable(SerializationFeature.INDENT_OUTPUT);
Review comment:
Moved it in to `onScheduled` instead - the ObjectMapper only needs to be created if errors are likely to be logged, so this will save unnecessary instantiations and setup
Noticed that a similar change could be made in the `AbstractElasticsearchHttpProcessor` (part of `nifi-elasticsearch-processors`) as a new `ObjectMapper` was being created for parsing every response in each of the processors that inherrit that class
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@nifi.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [nifi] ChrisSamo632 removed a comment on pull request #4693: NIFI-8001 Elasticsearch Dynamic Parameters
Posted by GitBox <gi...@apache.org>.
ChrisSamo632 removed a comment on pull request #4693:
URL: https://github.com/apache/nifi/pull/4693#issuecomment-869953462
Impacts https://github.com/apache/nifi/pull/5193 (NIFI-8002), which uses the same request parameters approach to control paginated queries. This PR should be rebased if the other is merged first (or vice versa).
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@nifi.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [nifi] gresockj commented on a change in pull request #4693: NIFI-8001 Elasticsearch Dynamic Parameters
Posted by GitBox <gi...@apache.org>.
gresockj commented on a change in pull request #4693:
URL: https://github.com/apache/nifi/pull/4693#discussion_r724404411
##########
File path: nifi-nar-bundles/nifi-elasticsearch-bundle/nifi-elasticsearch-restapi-processors/src/test/groovy/org/apache/nifi/processors/elasticsearch/AbstractByQueryElasticsearchTest.groovy
##########
@@ -0,0 +1,218 @@
+/*
+ * 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.elasticsearch
+
+import org.apache.nifi.util.MockFlowFile
+import org.apache.nifi.util.TestRunner
+import org.junit.Assert
+import org.junit.Test
+
+abstract class AbstractByQueryElasticsearchTest {
Review comment:
Fair enough, sounds good.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@nifi.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [nifi] ChrisSamo632 commented on pull request #4693: NIFI-8001 Elasticsearch Dynamic Parameters
Posted by GitBox <gi...@apache.org>.
ChrisSamo632 commented on pull request #4693:
URL: https://github.com/apache/nifi/pull/4693#issuecomment-869953462
Impacts https://github.com/apache/nifi/pull/5193 (NIFI-8002), which uses the same request parameters approach to control paginated queries. This PR should be rebased if the other is merged first (or vice versa).
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@nifi.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [nifi] ChrisSamo632 commented on a change in pull request #4693: NIFI-8001 Elasticsearch Dynamic Parameters
Posted by GitBox <gi...@apache.org>.
ChrisSamo632 commented on a change in pull request #4693:
URL: https://github.com/apache/nifi/pull/4693#discussion_r724386054
##########
File path: nifi-nar-bundles/nifi-elasticsearch-bundle/nifi-elasticsearch-restapi-processors/src/main/java/org/apache/nifi/processors/elasticsearch/PutElasticsearchRecord.java
##########
@@ -414,8 +434,8 @@ private void removeBadRecordFlowFiles(List<FlowFile> bad, ProcessSession session
bad.clear();
}
- private FlowFile indexDocuments(BulkOperation bundle, ProcessSession session, FlowFile input) throws Exception {
- IndexOperationResponse response = clientService.bulk(bundle.getOperationList());
+ private FlowFile indexDocuments(BulkOperation bundle, ProcessContext context, ProcessSession session, FlowFile input) throws Exception {
+ IndexOperationResponse response = clientService.bulk(bundle.getOperationList(), getUrlQueryParameters(context, input));
Review comment:
`final`s, `final`s everywhere π
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@nifi.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [nifi] ChrisSamo632 commented on a change in pull request #4693: NIFI-8001 Elasticsearch Dynamic Parameters
Posted by GitBox <gi...@apache.org>.
ChrisSamo632 commented on a change in pull request #4693:
URL: https://github.com/apache/nifi/pull/4693#discussion_r724354340
##########
File path: nifi-nar-bundles/nifi-elasticsearch-bundle/nifi-elasticsearch-client-service/src/main/java/org/apache/nifi/elasticsearch/ElasticSearchClientServiceImpl.java
##########
@@ -182,8 +181,8 @@ private Response runQuery(final String endpoint, final String query, final Strin
final HttpEntity queryEntity = new NStringEntity(query, ContentType.APPLICATION_JSON);
try {
- return client.performRequest("POST", sb.toString(), requestParameters != null ? requestParameters : Collections.emptyMap(), queryEntity);
- } catch (final Exception e) {
+ return performRequest("POST", sb.toString(), requestParameters, queryEntity);
Review comment:
Not sure what you want to be `final` here? I've applied `final` to everywhere I can see it may be usable within the class (there were a couple of things I'd missed previously, but this line isn't assigning any variables, so there's nothing to make `final`)?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@nifi.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [nifi] ChrisSamo632 commented on a change in pull request #4693: NIFI-8001 Elasticsearch Dynamic Parameters
Posted by GitBox <gi...@apache.org>.
ChrisSamo632 commented on a change in pull request #4693:
URL: https://github.com/apache/nifi/pull/4693#discussion_r724390382
##########
File path: nifi-nar-bundles/nifi-elasticsearch-bundle/nifi-elasticsearch-restapi-processors/src/test/groovy/org/apache/nifi/processors/elasticsearch/AbstractByQueryElasticsearchTest.groovy
##########
@@ -0,0 +1,218 @@
+/*
+ * 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.elasticsearch
+
+import org.apache.nifi.util.MockFlowFile
+import org.apache.nifi.util.TestRunner
+import org.junit.Assert
+import org.junit.Test
+
+abstract class AbstractByQueryElasticsearchTest {
Review comment:
While it should be fairly straight forward to convert them, I think that should be part of a larger initiative if/when decided upon. Also, we couldn't convert just this one class, we'd need to convert all of the tests (pretty much) in this package.
Also, currently, all the tests in the package are Groovy and not Java, I think moving just a sub-set of the tests would lead to more confusion rather than solving anything.
Maybe we should raise a follow-on Jira ticket to suggest the move and see what people think?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@nifi.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [nifi] gresockj commented on a change in pull request #4693: NIFI-8001 Elasticsearch Dynamic Parameters
Posted by GitBox <gi...@apache.org>.
gresockj commented on a change in pull request #4693:
URL: https://github.com/apache/nifi/pull/4693#discussion_r724403203
##########
File path: nifi-nar-bundles/nifi-elasticsearch-bundle/nifi-elasticsearch-client-service/src/main/java/org/apache/nifi/elasticsearch/ElasticSearchClientServiceImpl.java
##########
@@ -182,8 +181,8 @@ private Response runQuery(final String endpoint, final String query, final Strin
final HttpEntity queryEntity = new NStringEntity(query, ContentType.APPLICATION_JSON);
try {
- return client.performRequest("POST", sb.toString(), requestParameters != null ? requestParameters : Collections.emptyMap(), queryEntity);
- } catch (final Exception e) {
+ return performRequest("POST", sb.toString(), requestParameters, queryEntity);
Review comment:
Sorry, looks like I clicked the wrong line. I meant the Exception, which used to be final, but no longer is with the PR
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@nifi.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [nifi] ChrisSamo632 commented on a change in pull request #4693: NIFI-8001 Elasticsearch Dynamic Parameters
Posted by GitBox <gi...@apache.org>.
ChrisSamo632 commented on a change in pull request #4693:
URL: https://github.com/apache/nifi/pull/4693#discussion_r724448626
##########
File path: nifi-nar-bundles/nifi-elasticsearch-bundle/nifi-elasticsearch-restapi-processors/src/test/groovy/org/apache/nifi/processors/elasticsearch/AbstractByQueryElasticsearchTest.groovy
##########
@@ -0,0 +1,218 @@
+/*
+ * 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.elasticsearch
+
+import org.apache.nifi.util.MockFlowFile
+import org.apache.nifi.util.TestRunner
+import org.junit.Assert
+import org.junit.Test
+
+abstract class AbstractByQueryElasticsearchTest {
Review comment:
[NIFI-9290](https://issues.apache.org/jira/browse/NIFI-9290) raised to cover this consideration
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@nifi.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [nifi] gresockj commented on a change in pull request #4693: NIFI-8001 Elasticsearch Dynamic Parameters
Posted by GitBox <gi...@apache.org>.
gresockj commented on a change in pull request #4693:
URL: https://github.com/apache/nifi/pull/4693#discussion_r724054441
##########
File path: nifi-nar-bundles/nifi-elasticsearch-bundle/nifi-elasticsearch-client-service/src/main/java/org/apache/nifi/elasticsearch/ElasticSearchClientServiceImpl.java
##########
@@ -182,8 +181,8 @@ private Response runQuery(final String endpoint, final String query, final Strin
final HttpEntity queryEntity = new NStringEntity(query, ContentType.APPLICATION_JSON);
try {
- return client.performRequest("POST", sb.toString(), requestParameters != null ? requestParameters : Collections.emptyMap(), queryEntity);
- } catch (final Exception e) {
+ return performRequest("POST", sb.toString(), requestParameters, queryEntity);
Review comment:
Consider `final`
##########
File path: nifi-nar-bundles/nifi-elasticsearch-bundle/nifi-elasticsearch-client-service/src/main/java/org/apache/nifi/elasticsearch/ElasticSearchLookupService.java
##########
@@ -104,26 +103,25 @@ public void onEnabled(final ConfigurationContext context) {
type = context.getProperty(TYPE).evaluateAttributeExpressions().getValue();
mapper = new ObjectMapper();
- List<PropertyDescriptor> dynamic = context.getProperties().entrySet().stream()
- .filter( e -> e.getKey().isDynamic())
- .map(e -> e.getKey())
+ List<PropertyDescriptor> dynamic = context.getProperties().keySet().stream()
+ .filter(PropertyDescriptor::isDynamic)
.collect(Collectors.toList());
- Map<String, RecordPath> _temp = new HashMap<>();
+ Map<String, RecordPath> temp = new HashMap<>();
for (PropertyDescriptor desc : dynamic) {
String value = context.getProperty(desc).getValue();
Review comment:
We could add some `final` modifiers here while we're at it.
##########
File path: nifi-nar-bundles/nifi-elasticsearch-bundle/nifi-elasticsearch-client-service/src/main/java/org/apache/nifi/elasticsearch/ElasticSearchLookupService.java
##########
@@ -176,7 +174,7 @@ public PropertyDescriptor getSupportedDynamicPropertyDescriptor(String name) {
}
}
- private void validateCoordinates(Map coordinates) throws LookupFailureException {
+ private void validateCoordinates(Map<String, Object> coordinates) throws LookupFailureException {
List<String> reasons = new ArrayList<>();
Review comment:
Might as well make this `final`
##########
File path: nifi-nar-bundles/nifi-elasticsearch-bundle/nifi-elasticsearch-restapi-processors/src/main/java/org/apache/nifi/processors/elasticsearch/AbstractByQueryElasticsearch.java
##########
@@ -0,0 +1,150 @@
+/*
+ * 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.elasticsearch;
+
+import org.apache.nifi.annotation.lifecycle.OnScheduled;
+import org.apache.nifi.components.PropertyDescriptor;
+import org.apache.nifi.elasticsearch.ElasticSearchClientService;
+import org.apache.nifi.elasticsearch.OperationResponse;
+import org.apache.nifi.expression.ExpressionLanguageScope;
+import org.apache.nifi.flowfile.FlowFile;
+import org.apache.nifi.processor.AbstractProcessor;
+import org.apache.nifi.processor.ProcessContext;
+import org.apache.nifi.processor.ProcessSession;
+import org.apache.nifi.processor.Relationship;
+import org.apache.nifi.processor.util.StandardValidators;
+import org.apache.nifi.util.StringUtils;
+
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+
+public abstract class AbstractByQueryElasticsearch extends AbstractProcessor implements ElasticsearchRestProcessor {
+ public static final Relationship REL_FAILURE = new Relationship.Builder().name("failure")
+ .description("If the \"by query\" operation fails, and a flowfile was read, it will be sent to this relationship.")
+ .build();
+
+ public static final Relationship REL_SUCCESS = new Relationship.Builder().name("success")
+ .description("If the \"by query\" operation succeeds, and a flowfile was read, it will be sent to this relationship.")
+ .build();
+
+ private static final Set<Relationship> relationships;
+ private static final List<PropertyDescriptor> propertyDescriptors;
+
+ private volatile ElasticSearchClientService clientService;
+
+ static {
+ final Set<Relationship> rels = new HashSet<>();
+ rels.add(REL_SUCCESS);
+ rels.add(REL_FAILURE);
+ relationships = Collections.unmodifiableSet(rels);
+
+ final List<PropertyDescriptor> descriptors = new ArrayList<>();
+ descriptors.add(QUERY);
+ descriptors.add(QUERY_ATTRIBUTE);
+ descriptors.add(INDEX);
+ descriptors.add(TYPE);
+ descriptors.add(CLIENT_SERVICE);
+
+ propertyDescriptors = Collections.unmodifiableList(descriptors);
+ }
+
+ abstract String tookAttribute();
Review comment:
I think we should prefer a verb naming for these methods: `getTookAttribute` and `getErrorAttribute`
##########
File path: nifi-nar-bundles/nifi-elasticsearch-bundle/nifi-elasticsearch-restapi-processors/src/main/java/org/apache/nifi/processors/elasticsearch/DeleteByQueryElasticsearch.java
##########
@@ -17,129 +17,48 @@
package org.apache.nifi.processors.elasticsearch;
+import org.apache.nifi.annotation.behavior.DynamicProperty;
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.Tags;
-import org.apache.nifi.annotation.lifecycle.OnScheduled;
-import org.apache.nifi.components.PropertyDescriptor;
-import org.apache.nifi.elasticsearch.DeleteOperationResponse;
import org.apache.nifi.elasticsearch.ElasticSearchClientService;
-import org.apache.nifi.flowfile.FlowFile;
-import org.apache.nifi.processor.AbstractProcessor;
-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.util.StringUtils;
+import org.apache.nifi.elasticsearch.OperationResponse;
+import org.apache.nifi.expression.ExpressionLanguageScope;
-import java.util.ArrayList;
-import java.util.Collections;
-import java.util.HashMap;
-import java.util.HashSet;
-import java.util.List;
import java.util.Map;
-import java.util.Set;
+@WritesAttributes({
+ @WritesAttribute(attribute = "elasticsearch.delete.took", description = "The amount of time that it took to complete the delete operation in ms."),
+ @WritesAttribute(attribute = "elasticsearch.delete.error", description = "The error message provided by Elasticsearch if there is an error running the delete.")
+})
@InputRequirement(InputRequirement.Requirement.INPUT_ALLOWED)
+@Tags({ "elastic", "elasticsearch", "delete", "query"})
@CapabilityDescription("Delete from an Elasticsearch index using a query. The query can be loaded from a flowfile body " +
"or from the Query parameter.")
-@Tags({ "elastic", "elasticsearch", "delete", "query"})
-@WritesAttributes({
- @WritesAttribute(attribute = "elasticsearch.delete.took", description = "The amount of time that it took to complete the delete operation in ms."),
- @WritesAttribute(attribute = "elasticsearch.delete.error", description = "The error message provided by Elasticsearch if there is an error running the delete.")
-})
-public class DeleteByQueryElasticsearch extends AbstractProcessor implements ElasticsearchRestProcessor {
- public static final Relationship REL_FAILURE = new Relationship.Builder().name("failure")
- .description("If the delete by query fails, and a flowfile was read, it will be sent to this relationship.").build();
-
- public static final Relationship REL_SUCCESS = new Relationship.Builder().name("success")
- .description("If the delete by query succeeds, and a flowfile was read, it will be sent to this relationship.")
- .build();
-
-
+@DynamicProperty(
+ name = "A URL query parameter",
+ value = "The value to set it to",
+ expressionLanguageScope = ExpressionLanguageScope.FLOWFILE_ATTRIBUTES,
+ description = "Adds the specified property name/value as a query parameter in the Elasticsearch URL used for processing")
Review comment:
I think we should mention that these parameters will override any matching parameters in the request body, which might be provided by other non-dynamic properties.
##########
File path: nifi-nar-bundles/nifi-elasticsearch-bundle/nifi-elasticsearch-restapi-processors/src/main/java/org/apache/nifi/processors/elasticsearch/PutElasticsearchRecord.java
##########
@@ -254,14 +271,16 @@
return DESCRIPTORS;
}
- private RecordReaderFactory readerFactory;
- private RecordPathCache recordPathCache;
- private ElasticSearchClientService clientService;
- private RecordSetWriterFactory writerFactory;
- private boolean logErrors;
- private volatile String dateFormat;
- private volatile String timeFormat;
- private volatile String timestampFormat;
+ @Override
+ protected PropertyDescriptor getSupportedDynamicPropertyDescriptor(String propertyDescriptorName) {
Review comment:
`final`
##########
File path: nifi-nar-bundles/nifi-elasticsearch-bundle/nifi-elasticsearch-restapi-processors/src/main/java/org/apache/nifi/processors/elasticsearch/UpdateByQueryElasticsearch.java
##########
@@ -0,0 +1,64 @@
+/*
+ * 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.elasticsearch;
+
+import org.apache.nifi.annotation.behavior.DynamicProperty;
+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.Tags;
+import org.apache.nifi.elasticsearch.ElasticSearchClientService;
+import org.apache.nifi.elasticsearch.OperationResponse;
+import org.apache.nifi.expression.ExpressionLanguageScope;
+
+import java.util.Map;
+
+@WritesAttributes({
+ @WritesAttribute(attribute = "elasticsearch.update.took", description = "The amount of time that it took to complete the update operation in ms."),
+ @WritesAttribute(attribute = "elasticsearch.update.error", description = "The error message provided by Elasticsearch if there is an error running the update.")
+})
+@InputRequirement(InputRequirement.Requirement.INPUT_ALLOWED)
+@Tags({ "elastic", "elasticsearch", "update", "query"})
+@CapabilityDescription("Update documents in an Elasticsearch index using a query. The query can be loaded from a flowfile body " +
+ "or from the Query parameter.")
+@DynamicProperty(
+ name = "A URL query parameter",
+ value = "The value to set it to",
+ expressionLanguageScope = ExpressionLanguageScope.FLOWFILE_ATTRIBUTES,
Review comment:
Same comments as above
##########
File path: nifi-nar-bundles/nifi-elasticsearch-bundle/nifi-elasticsearch-client-service/src/main/java/org/apache/nifi/elasticsearch/ElasticSearchLookupService.java
##########
@@ -104,26 +103,25 @@ public void onEnabled(final ConfigurationContext context) {
type = context.getProperty(TYPE).evaluateAttributeExpressions().getValue();
mapper = new ObjectMapper();
- List<PropertyDescriptor> dynamic = context.getProperties().entrySet().stream()
- .filter( e -> e.getKey().isDynamic())
- .map(e -> e.getKey())
+ List<PropertyDescriptor> dynamic = context.getProperties().keySet().stream()
+ .filter(PropertyDescriptor::isDynamic)
.collect(Collectors.toList());
- Map<String, RecordPath> _temp = new HashMap<>();
+ Map<String, RecordPath> temp = new HashMap<>();
Review comment:
Let's rename this something more descriptive like `tempRecordPathMappings`
##########
File path: nifi-nar-bundles/nifi-elasticsearch-bundle/nifi-elasticsearch-client-service/src/main/java/org/apache/nifi/elasticsearch/ElasticSearchLookupService.java
##########
@@ -104,26 +103,25 @@ public void onEnabled(final ConfigurationContext context) {
type = context.getProperty(TYPE).evaluateAttributeExpressions().getValue();
mapper = new ObjectMapper();
- List<PropertyDescriptor> dynamic = context.getProperties().entrySet().stream()
- .filter( e -> e.getKey().isDynamic())
- .map(e -> e.getKey())
+ List<PropertyDescriptor> dynamic = context.getProperties().keySet().stream()
Review comment:
Let's take the opportunity to rename this `dynamicDescriptors` or something similar.
##########
File path: nifi-nar-bundles/nifi-elasticsearch-bundle/nifi-elasticsearch-client-service/src/main/java/org/apache/nifi/elasticsearch/ElasticSearchClientServiceImpl.java
##########
@@ -359,9 +358,21 @@ public DeleteOperationResponse deleteByQuery(final String query, final String in
return new DeleteOperationResponse(watch.getDuration(TimeUnit.MILLISECONDS));
}
+
+ public UpdateOperationResponse updateByQuery(final String query, final String index, final String type, final Map<String, String> requestParameters) {
+ long start = System.currentTimeMillis();
+ Response response = runQuery("_update_by_query", query, index, type, requestParameters);
+ long end = System.currentTimeMillis();
Review comment:
Though it aligns the `System.currentTimeMillis`, I think we should prefer removing the extra spaces here.
##########
File path: nifi-nar-bundles/nifi-elasticsearch-bundle/nifi-elasticsearch-client-service/src/main/java/org/apache/nifi/elasticsearch/ElasticSearchLookupService.java
##########
@@ -83,15 +83,14 @@
private String type;
private ObjectMapper mapper;
- private final List<PropertyDescriptor> DESCRIPTORS;
+ private final List<PropertyDescriptor> descriptors;
public ElasticSearchLookupService() {
- List<PropertyDescriptor> _desc = new ArrayList<>();
- _desc.addAll(super.getSupportedPropertyDescriptors());
- _desc.add(CLIENT_SERVICE);
- _desc.add(INDEX);
- _desc.add(TYPE);
- DESCRIPTORS = Collections.unmodifiableList(_desc);
+ List<PropertyDescriptor> desc = new ArrayList<>(super.getSupportedPropertyDescriptors());
+ desc.add(CLIENT_SERVICE);
+ desc.add(INDEX);
+ desc.add(TYPE);
+ descriptors = Collections.unmodifiableList(desc);
}
private volatile ConcurrentHashMap<String, RecordPath> mappings;
Review comment:
Let's move this up with the other variables, and rename it something more descriptive like `recordPathMappings`
##########
File path: nifi-nar-bundles/nifi-elasticsearch-bundle/nifi-elasticsearch-restapi-processors/src/main/java/org/apache/nifi/processors/elasticsearch/AbstractByQueryElasticsearch.java
##########
@@ -0,0 +1,150 @@
+/*
+ * 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.elasticsearch;
+
+import org.apache.nifi.annotation.lifecycle.OnScheduled;
+import org.apache.nifi.components.PropertyDescriptor;
+import org.apache.nifi.elasticsearch.ElasticSearchClientService;
+import org.apache.nifi.elasticsearch.OperationResponse;
+import org.apache.nifi.expression.ExpressionLanguageScope;
+import org.apache.nifi.flowfile.FlowFile;
+import org.apache.nifi.processor.AbstractProcessor;
+import org.apache.nifi.processor.ProcessContext;
+import org.apache.nifi.processor.ProcessSession;
+import org.apache.nifi.processor.Relationship;
+import org.apache.nifi.processor.util.StandardValidators;
+import org.apache.nifi.util.StringUtils;
+
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+
+public abstract class AbstractByQueryElasticsearch extends AbstractProcessor implements ElasticsearchRestProcessor {
+ public static final Relationship REL_FAILURE = new Relationship.Builder().name("failure")
+ .description("If the \"by query\" operation fails, and a flowfile was read, it will be sent to this relationship.")
+ .build();
+
+ public static final Relationship REL_SUCCESS = new Relationship.Builder().name("success")
+ .description("If the \"by query\" operation succeeds, and a flowfile was read, it will be sent to this relationship.")
+ .build();
+
+ private static final Set<Relationship> relationships;
+ private static final List<PropertyDescriptor> propertyDescriptors;
+
+ private volatile ElasticSearchClientService clientService;
+
+ static {
+ final Set<Relationship> rels = new HashSet<>();
+ rels.add(REL_SUCCESS);
+ rels.add(REL_FAILURE);
+ relationships = Collections.unmodifiableSet(rels);
+
+ final List<PropertyDescriptor> descriptors = new ArrayList<>();
+ descriptors.add(QUERY);
+ descriptors.add(QUERY_ATTRIBUTE);
+ descriptors.add(INDEX);
+ descriptors.add(TYPE);
+ descriptors.add(CLIENT_SERVICE);
+
+ propertyDescriptors = Collections.unmodifiableList(descriptors);
+ }
+
+ abstract String tookAttribute();
+
+ abstract String errorAttribute();
+
+ abstract OperationResponse performOperation(final ElasticSearchClientService clientService, final String query,
+ final String index, final String type,
+ final Map<String, String> requestParameters);
+
+ @Override
+ public Set<Relationship> getRelationships() {
+ return relationships;
+ }
+
+ @Override
+ public final List<PropertyDescriptor> getSupportedPropertyDescriptors() {
+ return propertyDescriptors;
+ }
+
+ @Override
+ protected PropertyDescriptor getSupportedDynamicPropertyDescriptor(String propertyDescriptorName) {
Review comment:
Could be `final`
##########
File path: nifi-nar-bundles/nifi-elasticsearch-bundle/nifi-elasticsearch-client-service/src/main/java/org/apache/nifi/elasticsearch/ElasticSearchLookupService.java
##########
@@ -290,22 +288,19 @@ private Record getByQuery(final Map<String, Object> query, Map<String, String> c
}
private Record applyMappings(Record record, Map<String, Object> source) {
- Record _rec = new MapRecord(record.getSchema(), new HashMap<>());
+ Record rec = new MapRecord(record.getSchema(), new HashMap<>());
- mappings.entrySet().forEach(entry -> {
+ mappings.forEach((key, path) -> {
try {
- Object o = JsonPath.read(source, entry.getKey());
- RecordPath path = entry.getValue();
- Optional<FieldValue> first = path.evaluate(_rec).getSelectedFields().findFirst();
- if (first.isPresent()) {
- first.get().updateValue(o);
- }
+ Object o = JsonPath.read(source, key);
Review comment:
Could use `final` here, etc.
##########
File path: nifi-nar-bundles/nifi-elasticsearch-bundle/nifi-elasticsearch-restapi-processors/src/main/java/org/apache/nifi/processors/elasticsearch/DeleteByQueryElasticsearch.java
##########
@@ -17,129 +17,48 @@
package org.apache.nifi.processors.elasticsearch;
+import org.apache.nifi.annotation.behavior.DynamicProperty;
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.Tags;
-import org.apache.nifi.annotation.lifecycle.OnScheduled;
-import org.apache.nifi.components.PropertyDescriptor;
-import org.apache.nifi.elasticsearch.DeleteOperationResponse;
import org.apache.nifi.elasticsearch.ElasticSearchClientService;
-import org.apache.nifi.flowfile.FlowFile;
-import org.apache.nifi.processor.AbstractProcessor;
-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.util.StringUtils;
+import org.apache.nifi.elasticsearch.OperationResponse;
+import org.apache.nifi.expression.ExpressionLanguageScope;
-import java.util.ArrayList;
-import java.util.Collections;
-import java.util.HashMap;
-import java.util.HashSet;
-import java.util.List;
import java.util.Map;
-import java.util.Set;
+@WritesAttributes({
+ @WritesAttribute(attribute = "elasticsearch.delete.took", description = "The amount of time that it took to complete the delete operation in ms."),
+ @WritesAttribute(attribute = "elasticsearch.delete.error", description = "The error message provided by Elasticsearch if there is an error running the delete.")
+})
@InputRequirement(InputRequirement.Requirement.INPUT_ALLOWED)
+@Tags({ "elastic", "elasticsearch", "delete", "query"})
@CapabilityDescription("Delete from an Elasticsearch index using a query. The query can be loaded from a flowfile body " +
"or from the Query parameter.")
-@Tags({ "elastic", "elasticsearch", "delete", "query"})
-@WritesAttributes({
- @WritesAttribute(attribute = "elasticsearch.delete.took", description = "The amount of time that it took to complete the delete operation in ms."),
- @WritesAttribute(attribute = "elasticsearch.delete.error", description = "The error message provided by Elasticsearch if there is an error running the delete.")
-})
-public class DeleteByQueryElasticsearch extends AbstractProcessor implements ElasticsearchRestProcessor {
- public static final Relationship REL_FAILURE = new Relationship.Builder().name("failure")
- .description("If the delete by query fails, and a flowfile was read, it will be sent to this relationship.").build();
-
- public static final Relationship REL_SUCCESS = new Relationship.Builder().name("success")
- .description("If the delete by query succeeds, and a flowfile was read, it will be sent to this relationship.")
- .build();
-
-
+@DynamicProperty(
+ name = "A URL query parameter",
Review comment:
Rephrasing suggestion:
```suggestion
name = "The name of a URL query parameter to add",
```
##########
File path: nifi-nar-bundles/nifi-elasticsearch-bundle/nifi-elasticsearch-restapi-processors/src/main/java/org/apache/nifi/processors/elasticsearch/AbstractPaginatedJsonQueryElasticsearch.java
##########
@@ -141,14 +141,21 @@ SearchResponse doQuery(final PaginatedJsonQueryParameters paginatedJsonQueryPara
if (!newQuery && PAGINATION_SCROLL.getValue().equals(paginationType)) {
response = clientService.get().scroll(queryJson);
} else {
+ Map<String, String> requestParameters = getUrlQueryParameters(context, input);
Review comment:
`final`
##########
File path: nifi-nar-bundles/nifi-elasticsearch-bundle/nifi-elasticsearch-client-service/src/main/java/org/apache/nifi/elasticsearch/ElasticSearchClientServiceImpl.java
##########
@@ -359,9 +358,21 @@ public DeleteOperationResponse deleteByQuery(final String query, final String in
return new DeleteOperationResponse(watch.getDuration(TimeUnit.MILLISECONDS));
}
+
+ public UpdateOperationResponse updateByQuery(final String query, final String index, final String type, final Map<String, String> requestParameters) {
+ long start = System.currentTimeMillis();
Review comment:
This variables could be `final`
##########
File path: nifi-nar-bundles/nifi-elasticsearch-bundle/nifi-elasticsearch-restapi-processors/src/main/java/org/apache/nifi/processors/elasticsearch/AbstractByQueryElasticsearch.java
##########
@@ -0,0 +1,150 @@
+/*
+ * 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.elasticsearch;
+
+import org.apache.nifi.annotation.lifecycle.OnScheduled;
+import org.apache.nifi.components.PropertyDescriptor;
+import org.apache.nifi.elasticsearch.ElasticSearchClientService;
+import org.apache.nifi.elasticsearch.OperationResponse;
+import org.apache.nifi.expression.ExpressionLanguageScope;
+import org.apache.nifi.flowfile.FlowFile;
+import org.apache.nifi.processor.AbstractProcessor;
+import org.apache.nifi.processor.ProcessContext;
+import org.apache.nifi.processor.ProcessSession;
+import org.apache.nifi.processor.Relationship;
+import org.apache.nifi.processor.util.StandardValidators;
+import org.apache.nifi.util.StringUtils;
+
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+
+public abstract class AbstractByQueryElasticsearch extends AbstractProcessor implements ElasticsearchRestProcessor {
+ public static final Relationship REL_FAILURE = new Relationship.Builder().name("failure")
+ .description("If the \"by query\" operation fails, and a flowfile was read, it will be sent to this relationship.")
+ .build();
+
+ public static final Relationship REL_SUCCESS = new Relationship.Builder().name("success")
+ .description("If the \"by query\" operation succeeds, and a flowfile was read, it will be sent to this relationship.")
+ .build();
+
+ private static final Set<Relationship> relationships;
+ private static final List<PropertyDescriptor> propertyDescriptors;
+
+ private volatile ElasticSearchClientService clientService;
+
+ static {
+ final Set<Relationship> rels = new HashSet<>();
+ rels.add(REL_SUCCESS);
+ rels.add(REL_FAILURE);
+ relationships = Collections.unmodifiableSet(rels);
+
+ final List<PropertyDescriptor> descriptors = new ArrayList<>();
+ descriptors.add(QUERY);
+ descriptors.add(QUERY_ATTRIBUTE);
+ descriptors.add(INDEX);
+ descriptors.add(TYPE);
+ descriptors.add(CLIENT_SERVICE);
+
+ propertyDescriptors = Collections.unmodifiableList(descriptors);
+ }
+
+ abstract String tookAttribute();
+
+ abstract String errorAttribute();
+
+ abstract OperationResponse performOperation(final ElasticSearchClientService clientService, final String query,
+ final String index, final String type,
+ final Map<String, String> requestParameters);
+
+ @Override
+ public Set<Relationship> getRelationships() {
+ return relationships;
+ }
+
+ @Override
+ public final List<PropertyDescriptor> getSupportedPropertyDescriptors() {
+ return propertyDescriptors;
+ }
+
+ @Override
+ protected PropertyDescriptor getSupportedDynamicPropertyDescriptor(String propertyDescriptorName) {
+ return new PropertyDescriptor.Builder()
+ .name(propertyDescriptorName)
+ .required(false)
+ .addValidator(StandardValidators.NON_EMPTY_VALIDATOR)
+ .expressionLanguageSupported(ExpressionLanguageScope.FLOWFILE_ATTRIBUTES)
+ .dynamic(true)
+ .build();
+ }
+
+ @OnScheduled
+ public void onScheduled(final ProcessContext context) {
+ clientService = context.getProperty(CLIENT_SERVICE).asControllerService(ElasticSearchClientService.class);
+ }
+
+ @Override
+ public void onTrigger(final ProcessContext context, final ProcessSession session) {
+ FlowFile input = null;
+ if (context.hasIncomingConnection()) {
+ input = session.get();
+
+ if (input == null && context.hasNonLoopConnection()) {
+ return;
+ }
+ }
+
+ try {
+ final String query = getQuery(input, context, session);
+ final String index = context.getProperty(INDEX).evaluateAttributeExpressions(input).getValue();
+ final String type = context.getProperty(TYPE).isSet()
+ ? context.getProperty(TYPE).evaluateAttributeExpressions(input).getValue()
+ : null;
+ final String queryAttr = context.getProperty(QUERY_ATTRIBUTE).isSet()
+ ? context.getProperty(QUERY_ATTRIBUTE).evaluateAttributeExpressions(input).getValue()
+ : null;
+
+ final OperationResponse or = performOperation(this.clientService, query, index, type, getUrlQueryParameters(context, input));
+
+ if (input == null) {
+ input = session.create();
+ }
+
+ final Map<String, String> attrs = new HashMap<>();
+ attrs.put(tookAttribute(), String.valueOf(or.getTook()));
+ if (!StringUtils.isBlank(queryAttr)) {
+ attrs.put(queryAttr, query);
+ }
+
+ input = session.putAllAttributes(input, attrs);
+
+ session.transfer(input, REL_SUCCESS);
+ } catch (Exception e) {
Review comment:
`final`
##########
File path: nifi-nar-bundles/nifi-elasticsearch-bundle/nifi-elasticsearch-restapi-processors/src/main/java/org/apache/nifi/processors/elasticsearch/DeleteByQueryElasticsearch.java
##########
@@ -17,129 +17,48 @@
package org.apache.nifi.processors.elasticsearch;
+import org.apache.nifi.annotation.behavior.DynamicProperty;
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.Tags;
-import org.apache.nifi.annotation.lifecycle.OnScheduled;
-import org.apache.nifi.components.PropertyDescriptor;
-import org.apache.nifi.elasticsearch.DeleteOperationResponse;
import org.apache.nifi.elasticsearch.ElasticSearchClientService;
-import org.apache.nifi.flowfile.FlowFile;
-import org.apache.nifi.processor.AbstractProcessor;
-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.util.StringUtils;
+import org.apache.nifi.elasticsearch.OperationResponse;
+import org.apache.nifi.expression.ExpressionLanguageScope;
-import java.util.ArrayList;
-import java.util.Collections;
-import java.util.HashMap;
-import java.util.HashSet;
-import java.util.List;
import java.util.Map;
-import java.util.Set;
+@WritesAttributes({
+ @WritesAttribute(attribute = "elasticsearch.delete.took", description = "The amount of time that it took to complete the delete operation in ms."),
+ @WritesAttribute(attribute = "elasticsearch.delete.error", description = "The error message provided by Elasticsearch if there is an error running the delete.")
+})
@InputRequirement(InputRequirement.Requirement.INPUT_ALLOWED)
+@Tags({ "elastic", "elasticsearch", "delete", "query"})
@CapabilityDescription("Delete from an Elasticsearch index using a query. The query can be loaded from a flowfile body " +
"or from the Query parameter.")
-@Tags({ "elastic", "elasticsearch", "delete", "query"})
-@WritesAttributes({
- @WritesAttribute(attribute = "elasticsearch.delete.took", description = "The amount of time that it took to complete the delete operation in ms."),
- @WritesAttribute(attribute = "elasticsearch.delete.error", description = "The error message provided by Elasticsearch if there is an error running the delete.")
-})
-public class DeleteByQueryElasticsearch extends AbstractProcessor implements ElasticsearchRestProcessor {
- public static final Relationship REL_FAILURE = new Relationship.Builder().name("failure")
- .description("If the delete by query fails, and a flowfile was read, it will be sent to this relationship.").build();
-
- public static final Relationship REL_SUCCESS = new Relationship.Builder().name("success")
- .description("If the delete by query succeeds, and a flowfile was read, it will be sent to this relationship.")
- .build();
-
-
+@DynamicProperty(
+ name = "A URL query parameter",
+ value = "The value to set it to",
Review comment:
Rephrasing suggestion:
```suggestion
value = "The value of the URL query parameter",
```
##########
File path: nifi-nar-bundles/nifi-elasticsearch-bundle/nifi-elasticsearch-restapi-processors/src/main/java/org/apache/nifi/processors/elasticsearch/JsonQueryElasticsearch.java
##########
@@ -46,11 +48,15 @@
"Elasticsearch JSON DSL. It does not automatically paginate queries for the user. If an incoming relationship is added to this " +
"processor, it will use the flowfile's content for the query. Care should be taken on the size of the query because the entire response " +
"from Elasticsearch will be loaded into memory all at once and converted into the resulting flowfiles.")
+@DynamicProperty(
+ name = "A URL query parameter",
+ value = "The value to set it to",
+ expressionLanguageScope = ExpressionLanguageScope.FLOWFILE_ATTRIBUTES,
+ description = "Adds the specified property name/value as a query parameter in the Elasticsearch URL used for processing")
Review comment:
Same comments as above
##########
File path: nifi-nar-bundles/nifi-elasticsearch-bundle/nifi-elasticsearch-restapi-processors/src/main/java/org/apache/nifi/processors/elasticsearch/PaginatedJsonQueryElasticsearch.java
##########
@@ -48,6 +50,11 @@
@CapabilityDescription("A processor that allows the user to run a paginated query (with aggregations) written with the Elasticsearch JSON DSL. " +
"It will use the flowfile's content for the query unless the QUERY attribute is populated. " +
"Search After/Point in Time queries must include a valid \"sort\" field.")
+@DynamicProperty(
+ name = "A URL query parameter",
+ value = "The value to set it to",
+ expressionLanguageScope = ExpressionLanguageScope.FLOWFILE_ATTRIBUTES,
+ description = "Adds the specified property name/value as a query parameter in the Elasticsearch URL used for processing")
Review comment:
Same comments as above
##########
File path: nifi-nar-bundles/nifi-elasticsearch-bundle/nifi-elasticsearch-restapi-processors/src/test/groovy/org/apache/nifi/processors/elasticsearch/AbstractByQueryElasticsearchTest.groovy
##########
@@ -0,0 +1,218 @@
+/*
+ * 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.elasticsearch
+
+import org.apache.nifi.util.MockFlowFile
+import org.apache.nifi.util.TestRunner
+import org.junit.Assert
+import org.junit.Test
+
+abstract class AbstractByQueryElasticsearchTest {
Review comment:
This looks like it could be easily converted to a Java test. I know opinions are split on Groovy vs Java tests, but I'm going to recommend converting this one.
##########
File path: nifi-nar-bundles/nifi-elasticsearch-bundle/nifi-elasticsearch-restapi-processors/src/main/java/org/apache/nifi/processors/elasticsearch/PutElasticsearchRecord.java
##########
@@ -414,8 +434,8 @@ private void removeBadRecordFlowFiles(List<FlowFile> bad, ProcessSession session
bad.clear();
}
- private FlowFile indexDocuments(BulkOperation bundle, ProcessSession session, FlowFile input) throws Exception {
- IndexOperationResponse response = clientService.bulk(bundle.getOperationList());
+ private FlowFile indexDocuments(BulkOperation bundle, ProcessContext context, ProcessSession session, FlowFile input) throws Exception {
+ IndexOperationResponse response = clientService.bulk(bundle.getOperationList(), getUrlQueryParameters(context, input));
Review comment:
`final`
##########
File path: nifi-nar-bundles/nifi-elasticsearch-bundle/nifi-elasticsearch-restapi-processors/src/main/java/org/apache/nifi/processors/elasticsearch/PutElasticsearchRecord.java
##########
@@ -72,6 +74,11 @@
@InputRequirement(InputRequirement.Requirement.INPUT_REQUIRED)
@Tags({"json", "elasticsearch", "elasticsearch5", "elasticsearch6", "put", "index", "record"})
@CapabilityDescription("A record-aware Elasticsearch put processor that uses the official Elastic REST client libraries.")
+@DynamicProperty(
+ name = "A URL query parameter",
+ value = "The value to set it to",
+ expressionLanguageScope = ExpressionLanguageScope.FLOWFILE_ATTRIBUTES,
+ description = "Adds the specified property name/value as a query parameter in the Elasticsearch URL used for processing")
Review comment:
Same comments as above
##########
File path: nifi-nar-bundles/nifi-elasticsearch-bundle/nifi-elasticsearch-restapi-processors/src/main/java/org/apache/nifi/processors/elasticsearch/SearchElasticsearch.java
##########
@@ -65,6 +66,11 @@
"Search After/Point in Time queries must include a valid \"sort\" field. The processor will retrieve multiple pages of results " +
"until either no more results are available or the Pagination Keep Alive expiration is reached, after which the query will " +
"restart with the first page of results being retrieved.")
+@DynamicProperty(
+ name = "A URL query parameter",
+ value = "The value to set it to",
+ expressionLanguageScope = ExpressionLanguageScope.FLOWFILE_ATTRIBUTES,
+ description = "Adds the specified property name/value as a query parameter in the Elasticsearch URL used for processing")
Review comment:
Same comments as above
##########
File path: nifi-nar-bundles/nifi-elasticsearch-bundle/nifi-elasticsearch-client-service/src/main/java/org/apache/nifi/elasticsearch/ElasticSearchClientServiceImpl.java
##########
@@ -182,8 +181,8 @@ private Response runQuery(final String endpoint, final String query, final Strin
final HttpEntity queryEntity = new NStringEntity(query, ContentType.APPLICATION_JSON);
try {
- return client.performRequest("POST", sb.toString(), requestParameters != null ? requestParameters : Collections.emptyMap(), queryEntity);
- } catch (final Exception e) {
+ return performRequest("POST", sb.toString(), requestParameters, queryEntity);
Review comment:
Sorry, looks like I clicked the wrong line. I meant the Exception, which used to be final, but no longer is with the PR
##########
File path: nifi-nar-bundles/nifi-elasticsearch-bundle/nifi-elasticsearch-restapi-processors/src/test/groovy/org/apache/nifi/processors/elasticsearch/AbstractByQueryElasticsearchTest.groovy
##########
@@ -0,0 +1,218 @@
+/*
+ * 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.elasticsearch
+
+import org.apache.nifi.util.MockFlowFile
+import org.apache.nifi.util.TestRunner
+import org.junit.Assert
+import org.junit.Test
+
+abstract class AbstractByQueryElasticsearchTest {
Review comment:
Fair enough, sounds good.
##########
File path: nifi-nar-bundles/nifi-elasticsearch-bundle/nifi-elasticsearch-restapi-processors/src/main/java/org/apache/nifi/processors/elasticsearch/PutElasticsearchRecord.java
##########
@@ -426,22 +429,21 @@ public void onTrigger(ProcessContext context, ProcessSession session) {
session.transfer(input, REL_SUCCESS);
}
- private void removeBadRecordFlowFiles(List<FlowFile> bad, ProcessSession session) {
- for (FlowFile badFlowFile : bad) {
+ private void removeBadRecordFlowFiles(final List<FlowFile> bad, final ProcessSession session) {
+ for (final FlowFile badFlowFile : bad) {
session.remove(badFlowFile);
}
bad.clear();
}
- private FlowFile indexDocuments(BulkOperation bundle, ProcessContext context, ProcessSession session, FlowFile input) throws Exception {
- IndexOperationResponse response = clientService.bulk(bundle.getOperationList(), getUrlQueryParameters(context, input));
+ private FlowFile indexDocuments(final BulkOperation bundle, final ProcessContext context, final ProcessSession session, final FlowFile input) throws Exception {
+ final IndexOperationResponse response = clientService.bulk(bundle.getOperationList(), getUrlQueryParameters(context, input));
if (response.hasErrors()) {
if(logErrors || getLogger().isDebugEnabled()) {
- List<Map<String, Object>> errors = response.getItems();
- ObjectMapper mapper = new ObjectMapper();
- mapper.enable(SerializationFeature.INDENT_OUTPUT);
- String output = String.format("An error was encountered while processing bulk operations. Server response below:%n%n%s", mapper.writeValueAsString(errors));
+ final List<Map<String, Object>> errors = response.getItems();
+ objectMapper.enable(SerializationFeature.INDENT_OUTPUT);
Review comment:
What do you think about putting the `enable()` call in the constructor?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@nifi.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [nifi] ChrisSamo632 commented on a change in pull request #4693: NIFI-8001 Elasticsearch Dynamic Parameters
Posted by GitBox <gi...@apache.org>.
ChrisSamo632 commented on a change in pull request #4693:
URL: https://github.com/apache/nifi/pull/4693#discussion_r724388277
##########
File path: nifi-nar-bundles/nifi-elasticsearch-bundle/nifi-elasticsearch-restapi-processors/src/main/java/org/apache/nifi/processors/elasticsearch/SearchElasticsearch.java
##########
@@ -65,6 +66,11 @@
"Search After/Point in Time queries must include a valid \"sort\" field. The processor will retrieve multiple pages of results " +
"until either no more results are available or the Pagination Keep Alive expiration is reached, after which the query will " +
"restart with the first page of results being retrieved.")
+@DynamicProperty(
+ name = "A URL query parameter",
+ value = "The value to set it to",
+ expressionLanguageScope = ExpressionLanguageScope.FLOWFILE_ATTRIBUTES,
+ description = "Adds the specified property name/value as a query parameter in the Elasticsearch URL used for processing")
Review comment:
I'm confused, which other comments are you referring to (I can't make this `final` π)?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@nifi.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [nifi] asfgit closed pull request #4693: NIFI-8001 Elasticsearch Dynamic Parameters
Posted by GitBox <gi...@apache.org>.
asfgit closed pull request #4693:
URL: https://github.com/apache/nifi/pull/4693
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@nifi.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [nifi] gresockj commented on a change in pull request #4693: NIFI-8001 Elasticsearch Dynamic Parameters
Posted by GitBox <gi...@apache.org>.
gresockj commented on a change in pull request #4693:
URL: https://github.com/apache/nifi/pull/4693#discussion_r724054441
##########
File path: nifi-nar-bundles/nifi-elasticsearch-bundle/nifi-elasticsearch-client-service/src/main/java/org/apache/nifi/elasticsearch/ElasticSearchClientServiceImpl.java
##########
@@ -182,8 +181,8 @@ private Response runQuery(final String endpoint, final String query, final Strin
final HttpEntity queryEntity = new NStringEntity(query, ContentType.APPLICATION_JSON);
try {
- return client.performRequest("POST", sb.toString(), requestParameters != null ? requestParameters : Collections.emptyMap(), queryEntity);
- } catch (final Exception e) {
+ return performRequest("POST", sb.toString(), requestParameters, queryEntity);
Review comment:
Consider `final`
##########
File path: nifi-nar-bundles/nifi-elasticsearch-bundle/nifi-elasticsearch-client-service/src/main/java/org/apache/nifi/elasticsearch/ElasticSearchLookupService.java
##########
@@ -104,26 +103,25 @@ public void onEnabled(final ConfigurationContext context) {
type = context.getProperty(TYPE).evaluateAttributeExpressions().getValue();
mapper = new ObjectMapper();
- List<PropertyDescriptor> dynamic = context.getProperties().entrySet().stream()
- .filter( e -> e.getKey().isDynamic())
- .map(e -> e.getKey())
+ List<PropertyDescriptor> dynamic = context.getProperties().keySet().stream()
+ .filter(PropertyDescriptor::isDynamic)
.collect(Collectors.toList());
- Map<String, RecordPath> _temp = new HashMap<>();
+ Map<String, RecordPath> temp = new HashMap<>();
for (PropertyDescriptor desc : dynamic) {
String value = context.getProperty(desc).getValue();
Review comment:
We could add some `final` modifiers here while we're at it.
##########
File path: nifi-nar-bundles/nifi-elasticsearch-bundle/nifi-elasticsearch-client-service/src/main/java/org/apache/nifi/elasticsearch/ElasticSearchLookupService.java
##########
@@ -176,7 +174,7 @@ public PropertyDescriptor getSupportedDynamicPropertyDescriptor(String name) {
}
}
- private void validateCoordinates(Map coordinates) throws LookupFailureException {
+ private void validateCoordinates(Map<String, Object> coordinates) throws LookupFailureException {
List<String> reasons = new ArrayList<>();
Review comment:
Might as well make this `final`
##########
File path: nifi-nar-bundles/nifi-elasticsearch-bundle/nifi-elasticsearch-restapi-processors/src/main/java/org/apache/nifi/processors/elasticsearch/AbstractByQueryElasticsearch.java
##########
@@ -0,0 +1,150 @@
+/*
+ * 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.elasticsearch;
+
+import org.apache.nifi.annotation.lifecycle.OnScheduled;
+import org.apache.nifi.components.PropertyDescriptor;
+import org.apache.nifi.elasticsearch.ElasticSearchClientService;
+import org.apache.nifi.elasticsearch.OperationResponse;
+import org.apache.nifi.expression.ExpressionLanguageScope;
+import org.apache.nifi.flowfile.FlowFile;
+import org.apache.nifi.processor.AbstractProcessor;
+import org.apache.nifi.processor.ProcessContext;
+import org.apache.nifi.processor.ProcessSession;
+import org.apache.nifi.processor.Relationship;
+import org.apache.nifi.processor.util.StandardValidators;
+import org.apache.nifi.util.StringUtils;
+
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+
+public abstract class AbstractByQueryElasticsearch extends AbstractProcessor implements ElasticsearchRestProcessor {
+ public static final Relationship REL_FAILURE = new Relationship.Builder().name("failure")
+ .description("If the \"by query\" operation fails, and a flowfile was read, it will be sent to this relationship.")
+ .build();
+
+ public static final Relationship REL_SUCCESS = new Relationship.Builder().name("success")
+ .description("If the \"by query\" operation succeeds, and a flowfile was read, it will be sent to this relationship.")
+ .build();
+
+ private static final Set<Relationship> relationships;
+ private static final List<PropertyDescriptor> propertyDescriptors;
+
+ private volatile ElasticSearchClientService clientService;
+
+ static {
+ final Set<Relationship> rels = new HashSet<>();
+ rels.add(REL_SUCCESS);
+ rels.add(REL_FAILURE);
+ relationships = Collections.unmodifiableSet(rels);
+
+ final List<PropertyDescriptor> descriptors = new ArrayList<>();
+ descriptors.add(QUERY);
+ descriptors.add(QUERY_ATTRIBUTE);
+ descriptors.add(INDEX);
+ descriptors.add(TYPE);
+ descriptors.add(CLIENT_SERVICE);
+
+ propertyDescriptors = Collections.unmodifiableList(descriptors);
+ }
+
+ abstract String tookAttribute();
Review comment:
I think we should prefer a verb naming for these methods: `getTookAttribute` and `getErrorAttribute`
##########
File path: nifi-nar-bundles/nifi-elasticsearch-bundle/nifi-elasticsearch-restapi-processors/src/main/java/org/apache/nifi/processors/elasticsearch/DeleteByQueryElasticsearch.java
##########
@@ -17,129 +17,48 @@
package org.apache.nifi.processors.elasticsearch;
+import org.apache.nifi.annotation.behavior.DynamicProperty;
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.Tags;
-import org.apache.nifi.annotation.lifecycle.OnScheduled;
-import org.apache.nifi.components.PropertyDescriptor;
-import org.apache.nifi.elasticsearch.DeleteOperationResponse;
import org.apache.nifi.elasticsearch.ElasticSearchClientService;
-import org.apache.nifi.flowfile.FlowFile;
-import org.apache.nifi.processor.AbstractProcessor;
-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.util.StringUtils;
+import org.apache.nifi.elasticsearch.OperationResponse;
+import org.apache.nifi.expression.ExpressionLanguageScope;
-import java.util.ArrayList;
-import java.util.Collections;
-import java.util.HashMap;
-import java.util.HashSet;
-import java.util.List;
import java.util.Map;
-import java.util.Set;
+@WritesAttributes({
+ @WritesAttribute(attribute = "elasticsearch.delete.took", description = "The amount of time that it took to complete the delete operation in ms."),
+ @WritesAttribute(attribute = "elasticsearch.delete.error", description = "The error message provided by Elasticsearch if there is an error running the delete.")
+})
@InputRequirement(InputRequirement.Requirement.INPUT_ALLOWED)
+@Tags({ "elastic", "elasticsearch", "delete", "query"})
@CapabilityDescription("Delete from an Elasticsearch index using a query. The query can be loaded from a flowfile body " +
"or from the Query parameter.")
-@Tags({ "elastic", "elasticsearch", "delete", "query"})
-@WritesAttributes({
- @WritesAttribute(attribute = "elasticsearch.delete.took", description = "The amount of time that it took to complete the delete operation in ms."),
- @WritesAttribute(attribute = "elasticsearch.delete.error", description = "The error message provided by Elasticsearch if there is an error running the delete.")
-})
-public class DeleteByQueryElasticsearch extends AbstractProcessor implements ElasticsearchRestProcessor {
- public static final Relationship REL_FAILURE = new Relationship.Builder().name("failure")
- .description("If the delete by query fails, and a flowfile was read, it will be sent to this relationship.").build();
-
- public static final Relationship REL_SUCCESS = new Relationship.Builder().name("success")
- .description("If the delete by query succeeds, and a flowfile was read, it will be sent to this relationship.")
- .build();
-
-
+@DynamicProperty(
+ name = "A URL query parameter",
+ value = "The value to set it to",
+ expressionLanguageScope = ExpressionLanguageScope.FLOWFILE_ATTRIBUTES,
+ description = "Adds the specified property name/value as a query parameter in the Elasticsearch URL used for processing")
Review comment:
I think we should mention that these parameters will override any matching parameters in the request body, which might be provided by other non-dynamic properties.
##########
File path: nifi-nar-bundles/nifi-elasticsearch-bundle/nifi-elasticsearch-restapi-processors/src/main/java/org/apache/nifi/processors/elasticsearch/PutElasticsearchRecord.java
##########
@@ -254,14 +271,16 @@
return DESCRIPTORS;
}
- private RecordReaderFactory readerFactory;
- private RecordPathCache recordPathCache;
- private ElasticSearchClientService clientService;
- private RecordSetWriterFactory writerFactory;
- private boolean logErrors;
- private volatile String dateFormat;
- private volatile String timeFormat;
- private volatile String timestampFormat;
+ @Override
+ protected PropertyDescriptor getSupportedDynamicPropertyDescriptor(String propertyDescriptorName) {
Review comment:
`final`
##########
File path: nifi-nar-bundles/nifi-elasticsearch-bundle/nifi-elasticsearch-restapi-processors/src/main/java/org/apache/nifi/processors/elasticsearch/UpdateByQueryElasticsearch.java
##########
@@ -0,0 +1,64 @@
+/*
+ * 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.elasticsearch;
+
+import org.apache.nifi.annotation.behavior.DynamicProperty;
+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.Tags;
+import org.apache.nifi.elasticsearch.ElasticSearchClientService;
+import org.apache.nifi.elasticsearch.OperationResponse;
+import org.apache.nifi.expression.ExpressionLanguageScope;
+
+import java.util.Map;
+
+@WritesAttributes({
+ @WritesAttribute(attribute = "elasticsearch.update.took", description = "The amount of time that it took to complete the update operation in ms."),
+ @WritesAttribute(attribute = "elasticsearch.update.error", description = "The error message provided by Elasticsearch if there is an error running the update.")
+})
+@InputRequirement(InputRequirement.Requirement.INPUT_ALLOWED)
+@Tags({ "elastic", "elasticsearch", "update", "query"})
+@CapabilityDescription("Update documents in an Elasticsearch index using a query. The query can be loaded from a flowfile body " +
+ "or from the Query parameter.")
+@DynamicProperty(
+ name = "A URL query parameter",
+ value = "The value to set it to",
+ expressionLanguageScope = ExpressionLanguageScope.FLOWFILE_ATTRIBUTES,
Review comment:
Same comments as above
##########
File path: nifi-nar-bundles/nifi-elasticsearch-bundle/nifi-elasticsearch-client-service/src/main/java/org/apache/nifi/elasticsearch/ElasticSearchLookupService.java
##########
@@ -104,26 +103,25 @@ public void onEnabled(final ConfigurationContext context) {
type = context.getProperty(TYPE).evaluateAttributeExpressions().getValue();
mapper = new ObjectMapper();
- List<PropertyDescriptor> dynamic = context.getProperties().entrySet().stream()
- .filter( e -> e.getKey().isDynamic())
- .map(e -> e.getKey())
+ List<PropertyDescriptor> dynamic = context.getProperties().keySet().stream()
+ .filter(PropertyDescriptor::isDynamic)
.collect(Collectors.toList());
- Map<String, RecordPath> _temp = new HashMap<>();
+ Map<String, RecordPath> temp = new HashMap<>();
Review comment:
Let's rename this something more descriptive like `tempRecordPathMappings`
##########
File path: nifi-nar-bundles/nifi-elasticsearch-bundle/nifi-elasticsearch-client-service/src/main/java/org/apache/nifi/elasticsearch/ElasticSearchLookupService.java
##########
@@ -104,26 +103,25 @@ public void onEnabled(final ConfigurationContext context) {
type = context.getProperty(TYPE).evaluateAttributeExpressions().getValue();
mapper = new ObjectMapper();
- List<PropertyDescriptor> dynamic = context.getProperties().entrySet().stream()
- .filter( e -> e.getKey().isDynamic())
- .map(e -> e.getKey())
+ List<PropertyDescriptor> dynamic = context.getProperties().keySet().stream()
Review comment:
Let's take the opportunity to rename this `dynamicDescriptors` or something similar.
##########
File path: nifi-nar-bundles/nifi-elasticsearch-bundle/nifi-elasticsearch-client-service/src/main/java/org/apache/nifi/elasticsearch/ElasticSearchClientServiceImpl.java
##########
@@ -359,9 +358,21 @@ public DeleteOperationResponse deleteByQuery(final String query, final String in
return new DeleteOperationResponse(watch.getDuration(TimeUnit.MILLISECONDS));
}
+
+ public UpdateOperationResponse updateByQuery(final String query, final String index, final String type, final Map<String, String> requestParameters) {
+ long start = System.currentTimeMillis();
+ Response response = runQuery("_update_by_query", query, index, type, requestParameters);
+ long end = System.currentTimeMillis();
Review comment:
Though it aligns the `System.currentTimeMillis`, I think we should prefer removing the extra spaces here.
##########
File path: nifi-nar-bundles/nifi-elasticsearch-bundle/nifi-elasticsearch-client-service/src/main/java/org/apache/nifi/elasticsearch/ElasticSearchLookupService.java
##########
@@ -83,15 +83,14 @@
private String type;
private ObjectMapper mapper;
- private final List<PropertyDescriptor> DESCRIPTORS;
+ private final List<PropertyDescriptor> descriptors;
public ElasticSearchLookupService() {
- List<PropertyDescriptor> _desc = new ArrayList<>();
- _desc.addAll(super.getSupportedPropertyDescriptors());
- _desc.add(CLIENT_SERVICE);
- _desc.add(INDEX);
- _desc.add(TYPE);
- DESCRIPTORS = Collections.unmodifiableList(_desc);
+ List<PropertyDescriptor> desc = new ArrayList<>(super.getSupportedPropertyDescriptors());
+ desc.add(CLIENT_SERVICE);
+ desc.add(INDEX);
+ desc.add(TYPE);
+ descriptors = Collections.unmodifiableList(desc);
}
private volatile ConcurrentHashMap<String, RecordPath> mappings;
Review comment:
Let's move this up with the other variables, and rename it something more descriptive like `recordPathMappings`
##########
File path: nifi-nar-bundles/nifi-elasticsearch-bundle/nifi-elasticsearch-restapi-processors/src/main/java/org/apache/nifi/processors/elasticsearch/AbstractByQueryElasticsearch.java
##########
@@ -0,0 +1,150 @@
+/*
+ * 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.elasticsearch;
+
+import org.apache.nifi.annotation.lifecycle.OnScheduled;
+import org.apache.nifi.components.PropertyDescriptor;
+import org.apache.nifi.elasticsearch.ElasticSearchClientService;
+import org.apache.nifi.elasticsearch.OperationResponse;
+import org.apache.nifi.expression.ExpressionLanguageScope;
+import org.apache.nifi.flowfile.FlowFile;
+import org.apache.nifi.processor.AbstractProcessor;
+import org.apache.nifi.processor.ProcessContext;
+import org.apache.nifi.processor.ProcessSession;
+import org.apache.nifi.processor.Relationship;
+import org.apache.nifi.processor.util.StandardValidators;
+import org.apache.nifi.util.StringUtils;
+
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+
+public abstract class AbstractByQueryElasticsearch extends AbstractProcessor implements ElasticsearchRestProcessor {
+ public static final Relationship REL_FAILURE = new Relationship.Builder().name("failure")
+ .description("If the \"by query\" operation fails, and a flowfile was read, it will be sent to this relationship.")
+ .build();
+
+ public static final Relationship REL_SUCCESS = new Relationship.Builder().name("success")
+ .description("If the \"by query\" operation succeeds, and a flowfile was read, it will be sent to this relationship.")
+ .build();
+
+ private static final Set<Relationship> relationships;
+ private static final List<PropertyDescriptor> propertyDescriptors;
+
+ private volatile ElasticSearchClientService clientService;
+
+ static {
+ final Set<Relationship> rels = new HashSet<>();
+ rels.add(REL_SUCCESS);
+ rels.add(REL_FAILURE);
+ relationships = Collections.unmodifiableSet(rels);
+
+ final List<PropertyDescriptor> descriptors = new ArrayList<>();
+ descriptors.add(QUERY);
+ descriptors.add(QUERY_ATTRIBUTE);
+ descriptors.add(INDEX);
+ descriptors.add(TYPE);
+ descriptors.add(CLIENT_SERVICE);
+
+ propertyDescriptors = Collections.unmodifiableList(descriptors);
+ }
+
+ abstract String tookAttribute();
+
+ abstract String errorAttribute();
+
+ abstract OperationResponse performOperation(final ElasticSearchClientService clientService, final String query,
+ final String index, final String type,
+ final Map<String, String> requestParameters);
+
+ @Override
+ public Set<Relationship> getRelationships() {
+ return relationships;
+ }
+
+ @Override
+ public final List<PropertyDescriptor> getSupportedPropertyDescriptors() {
+ return propertyDescriptors;
+ }
+
+ @Override
+ protected PropertyDescriptor getSupportedDynamicPropertyDescriptor(String propertyDescriptorName) {
Review comment:
Could be `final`
##########
File path: nifi-nar-bundles/nifi-elasticsearch-bundle/nifi-elasticsearch-client-service/src/main/java/org/apache/nifi/elasticsearch/ElasticSearchLookupService.java
##########
@@ -290,22 +288,19 @@ private Record getByQuery(final Map<String, Object> query, Map<String, String> c
}
private Record applyMappings(Record record, Map<String, Object> source) {
- Record _rec = new MapRecord(record.getSchema(), new HashMap<>());
+ Record rec = new MapRecord(record.getSchema(), new HashMap<>());
- mappings.entrySet().forEach(entry -> {
+ mappings.forEach((key, path) -> {
try {
- Object o = JsonPath.read(source, entry.getKey());
- RecordPath path = entry.getValue();
- Optional<FieldValue> first = path.evaluate(_rec).getSelectedFields().findFirst();
- if (first.isPresent()) {
- first.get().updateValue(o);
- }
+ Object o = JsonPath.read(source, key);
Review comment:
Could use `final` here, etc.
##########
File path: nifi-nar-bundles/nifi-elasticsearch-bundle/nifi-elasticsearch-restapi-processors/src/main/java/org/apache/nifi/processors/elasticsearch/DeleteByQueryElasticsearch.java
##########
@@ -17,129 +17,48 @@
package org.apache.nifi.processors.elasticsearch;
+import org.apache.nifi.annotation.behavior.DynamicProperty;
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.Tags;
-import org.apache.nifi.annotation.lifecycle.OnScheduled;
-import org.apache.nifi.components.PropertyDescriptor;
-import org.apache.nifi.elasticsearch.DeleteOperationResponse;
import org.apache.nifi.elasticsearch.ElasticSearchClientService;
-import org.apache.nifi.flowfile.FlowFile;
-import org.apache.nifi.processor.AbstractProcessor;
-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.util.StringUtils;
+import org.apache.nifi.elasticsearch.OperationResponse;
+import org.apache.nifi.expression.ExpressionLanguageScope;
-import java.util.ArrayList;
-import java.util.Collections;
-import java.util.HashMap;
-import java.util.HashSet;
-import java.util.List;
import java.util.Map;
-import java.util.Set;
+@WritesAttributes({
+ @WritesAttribute(attribute = "elasticsearch.delete.took", description = "The amount of time that it took to complete the delete operation in ms."),
+ @WritesAttribute(attribute = "elasticsearch.delete.error", description = "The error message provided by Elasticsearch if there is an error running the delete.")
+})
@InputRequirement(InputRequirement.Requirement.INPUT_ALLOWED)
+@Tags({ "elastic", "elasticsearch", "delete", "query"})
@CapabilityDescription("Delete from an Elasticsearch index using a query. The query can be loaded from a flowfile body " +
"or from the Query parameter.")
-@Tags({ "elastic", "elasticsearch", "delete", "query"})
-@WritesAttributes({
- @WritesAttribute(attribute = "elasticsearch.delete.took", description = "The amount of time that it took to complete the delete operation in ms."),
- @WritesAttribute(attribute = "elasticsearch.delete.error", description = "The error message provided by Elasticsearch if there is an error running the delete.")
-})
-public class DeleteByQueryElasticsearch extends AbstractProcessor implements ElasticsearchRestProcessor {
- public static final Relationship REL_FAILURE = new Relationship.Builder().name("failure")
- .description("If the delete by query fails, and a flowfile was read, it will be sent to this relationship.").build();
-
- public static final Relationship REL_SUCCESS = new Relationship.Builder().name("success")
- .description("If the delete by query succeeds, and a flowfile was read, it will be sent to this relationship.")
- .build();
-
-
+@DynamicProperty(
+ name = "A URL query parameter",
Review comment:
Rephrasing suggestion:
```suggestion
name = "The name of a URL query parameter to add",
```
##########
File path: nifi-nar-bundles/nifi-elasticsearch-bundle/nifi-elasticsearch-restapi-processors/src/main/java/org/apache/nifi/processors/elasticsearch/AbstractPaginatedJsonQueryElasticsearch.java
##########
@@ -141,14 +141,21 @@ SearchResponse doQuery(final PaginatedJsonQueryParameters paginatedJsonQueryPara
if (!newQuery && PAGINATION_SCROLL.getValue().equals(paginationType)) {
response = clientService.get().scroll(queryJson);
} else {
+ Map<String, String> requestParameters = getUrlQueryParameters(context, input);
Review comment:
`final`
##########
File path: nifi-nar-bundles/nifi-elasticsearch-bundle/nifi-elasticsearch-client-service/src/main/java/org/apache/nifi/elasticsearch/ElasticSearchClientServiceImpl.java
##########
@@ -359,9 +358,21 @@ public DeleteOperationResponse deleteByQuery(final String query, final String in
return new DeleteOperationResponse(watch.getDuration(TimeUnit.MILLISECONDS));
}
+
+ public UpdateOperationResponse updateByQuery(final String query, final String index, final String type, final Map<String, String> requestParameters) {
+ long start = System.currentTimeMillis();
Review comment:
This variables could be `final`
##########
File path: nifi-nar-bundles/nifi-elasticsearch-bundle/nifi-elasticsearch-restapi-processors/src/main/java/org/apache/nifi/processors/elasticsearch/AbstractByQueryElasticsearch.java
##########
@@ -0,0 +1,150 @@
+/*
+ * 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.elasticsearch;
+
+import org.apache.nifi.annotation.lifecycle.OnScheduled;
+import org.apache.nifi.components.PropertyDescriptor;
+import org.apache.nifi.elasticsearch.ElasticSearchClientService;
+import org.apache.nifi.elasticsearch.OperationResponse;
+import org.apache.nifi.expression.ExpressionLanguageScope;
+import org.apache.nifi.flowfile.FlowFile;
+import org.apache.nifi.processor.AbstractProcessor;
+import org.apache.nifi.processor.ProcessContext;
+import org.apache.nifi.processor.ProcessSession;
+import org.apache.nifi.processor.Relationship;
+import org.apache.nifi.processor.util.StandardValidators;
+import org.apache.nifi.util.StringUtils;
+
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+
+public abstract class AbstractByQueryElasticsearch extends AbstractProcessor implements ElasticsearchRestProcessor {
+ public static final Relationship REL_FAILURE = new Relationship.Builder().name("failure")
+ .description("If the \"by query\" operation fails, and a flowfile was read, it will be sent to this relationship.")
+ .build();
+
+ public static final Relationship REL_SUCCESS = new Relationship.Builder().name("success")
+ .description("If the \"by query\" operation succeeds, and a flowfile was read, it will be sent to this relationship.")
+ .build();
+
+ private static final Set<Relationship> relationships;
+ private static final List<PropertyDescriptor> propertyDescriptors;
+
+ private volatile ElasticSearchClientService clientService;
+
+ static {
+ final Set<Relationship> rels = new HashSet<>();
+ rels.add(REL_SUCCESS);
+ rels.add(REL_FAILURE);
+ relationships = Collections.unmodifiableSet(rels);
+
+ final List<PropertyDescriptor> descriptors = new ArrayList<>();
+ descriptors.add(QUERY);
+ descriptors.add(QUERY_ATTRIBUTE);
+ descriptors.add(INDEX);
+ descriptors.add(TYPE);
+ descriptors.add(CLIENT_SERVICE);
+
+ propertyDescriptors = Collections.unmodifiableList(descriptors);
+ }
+
+ abstract String tookAttribute();
+
+ abstract String errorAttribute();
+
+ abstract OperationResponse performOperation(final ElasticSearchClientService clientService, final String query,
+ final String index, final String type,
+ final Map<String, String> requestParameters);
+
+ @Override
+ public Set<Relationship> getRelationships() {
+ return relationships;
+ }
+
+ @Override
+ public final List<PropertyDescriptor> getSupportedPropertyDescriptors() {
+ return propertyDescriptors;
+ }
+
+ @Override
+ protected PropertyDescriptor getSupportedDynamicPropertyDescriptor(String propertyDescriptorName) {
+ return new PropertyDescriptor.Builder()
+ .name(propertyDescriptorName)
+ .required(false)
+ .addValidator(StandardValidators.NON_EMPTY_VALIDATOR)
+ .expressionLanguageSupported(ExpressionLanguageScope.FLOWFILE_ATTRIBUTES)
+ .dynamic(true)
+ .build();
+ }
+
+ @OnScheduled
+ public void onScheduled(final ProcessContext context) {
+ clientService = context.getProperty(CLIENT_SERVICE).asControllerService(ElasticSearchClientService.class);
+ }
+
+ @Override
+ public void onTrigger(final ProcessContext context, final ProcessSession session) {
+ FlowFile input = null;
+ if (context.hasIncomingConnection()) {
+ input = session.get();
+
+ if (input == null && context.hasNonLoopConnection()) {
+ return;
+ }
+ }
+
+ try {
+ final String query = getQuery(input, context, session);
+ final String index = context.getProperty(INDEX).evaluateAttributeExpressions(input).getValue();
+ final String type = context.getProperty(TYPE).isSet()
+ ? context.getProperty(TYPE).evaluateAttributeExpressions(input).getValue()
+ : null;
+ final String queryAttr = context.getProperty(QUERY_ATTRIBUTE).isSet()
+ ? context.getProperty(QUERY_ATTRIBUTE).evaluateAttributeExpressions(input).getValue()
+ : null;
+
+ final OperationResponse or = performOperation(this.clientService, query, index, type, getUrlQueryParameters(context, input));
+
+ if (input == null) {
+ input = session.create();
+ }
+
+ final Map<String, String> attrs = new HashMap<>();
+ attrs.put(tookAttribute(), String.valueOf(or.getTook()));
+ if (!StringUtils.isBlank(queryAttr)) {
+ attrs.put(queryAttr, query);
+ }
+
+ input = session.putAllAttributes(input, attrs);
+
+ session.transfer(input, REL_SUCCESS);
+ } catch (Exception e) {
Review comment:
`final`
##########
File path: nifi-nar-bundles/nifi-elasticsearch-bundle/nifi-elasticsearch-restapi-processors/src/main/java/org/apache/nifi/processors/elasticsearch/DeleteByQueryElasticsearch.java
##########
@@ -17,129 +17,48 @@
package org.apache.nifi.processors.elasticsearch;
+import org.apache.nifi.annotation.behavior.DynamicProperty;
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.Tags;
-import org.apache.nifi.annotation.lifecycle.OnScheduled;
-import org.apache.nifi.components.PropertyDescriptor;
-import org.apache.nifi.elasticsearch.DeleteOperationResponse;
import org.apache.nifi.elasticsearch.ElasticSearchClientService;
-import org.apache.nifi.flowfile.FlowFile;
-import org.apache.nifi.processor.AbstractProcessor;
-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.util.StringUtils;
+import org.apache.nifi.elasticsearch.OperationResponse;
+import org.apache.nifi.expression.ExpressionLanguageScope;
-import java.util.ArrayList;
-import java.util.Collections;
-import java.util.HashMap;
-import java.util.HashSet;
-import java.util.List;
import java.util.Map;
-import java.util.Set;
+@WritesAttributes({
+ @WritesAttribute(attribute = "elasticsearch.delete.took", description = "The amount of time that it took to complete the delete operation in ms."),
+ @WritesAttribute(attribute = "elasticsearch.delete.error", description = "The error message provided by Elasticsearch if there is an error running the delete.")
+})
@InputRequirement(InputRequirement.Requirement.INPUT_ALLOWED)
+@Tags({ "elastic", "elasticsearch", "delete", "query"})
@CapabilityDescription("Delete from an Elasticsearch index using a query. The query can be loaded from a flowfile body " +
"or from the Query parameter.")
-@Tags({ "elastic", "elasticsearch", "delete", "query"})
-@WritesAttributes({
- @WritesAttribute(attribute = "elasticsearch.delete.took", description = "The amount of time that it took to complete the delete operation in ms."),
- @WritesAttribute(attribute = "elasticsearch.delete.error", description = "The error message provided by Elasticsearch if there is an error running the delete.")
-})
-public class DeleteByQueryElasticsearch extends AbstractProcessor implements ElasticsearchRestProcessor {
- public static final Relationship REL_FAILURE = new Relationship.Builder().name("failure")
- .description("If the delete by query fails, and a flowfile was read, it will be sent to this relationship.").build();
-
- public static final Relationship REL_SUCCESS = new Relationship.Builder().name("success")
- .description("If the delete by query succeeds, and a flowfile was read, it will be sent to this relationship.")
- .build();
-
-
+@DynamicProperty(
+ name = "A URL query parameter",
+ value = "The value to set it to",
Review comment:
Rephrasing suggestion:
```suggestion
value = "The value of the URL query parameter",
```
##########
File path: nifi-nar-bundles/nifi-elasticsearch-bundle/nifi-elasticsearch-restapi-processors/src/main/java/org/apache/nifi/processors/elasticsearch/JsonQueryElasticsearch.java
##########
@@ -46,11 +48,15 @@
"Elasticsearch JSON DSL. It does not automatically paginate queries for the user. If an incoming relationship is added to this " +
"processor, it will use the flowfile's content for the query. Care should be taken on the size of the query because the entire response " +
"from Elasticsearch will be loaded into memory all at once and converted into the resulting flowfiles.")
+@DynamicProperty(
+ name = "A URL query parameter",
+ value = "The value to set it to",
+ expressionLanguageScope = ExpressionLanguageScope.FLOWFILE_ATTRIBUTES,
+ description = "Adds the specified property name/value as a query parameter in the Elasticsearch URL used for processing")
Review comment:
Same comments as above
##########
File path: nifi-nar-bundles/nifi-elasticsearch-bundle/nifi-elasticsearch-restapi-processors/src/main/java/org/apache/nifi/processors/elasticsearch/PaginatedJsonQueryElasticsearch.java
##########
@@ -48,6 +50,11 @@
@CapabilityDescription("A processor that allows the user to run a paginated query (with aggregations) written with the Elasticsearch JSON DSL. " +
"It will use the flowfile's content for the query unless the QUERY attribute is populated. " +
"Search After/Point in Time queries must include a valid \"sort\" field.")
+@DynamicProperty(
+ name = "A URL query parameter",
+ value = "The value to set it to",
+ expressionLanguageScope = ExpressionLanguageScope.FLOWFILE_ATTRIBUTES,
+ description = "Adds the specified property name/value as a query parameter in the Elasticsearch URL used for processing")
Review comment:
Same comments as above
##########
File path: nifi-nar-bundles/nifi-elasticsearch-bundle/nifi-elasticsearch-restapi-processors/src/test/groovy/org/apache/nifi/processors/elasticsearch/AbstractByQueryElasticsearchTest.groovy
##########
@@ -0,0 +1,218 @@
+/*
+ * 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.elasticsearch
+
+import org.apache.nifi.util.MockFlowFile
+import org.apache.nifi.util.TestRunner
+import org.junit.Assert
+import org.junit.Test
+
+abstract class AbstractByQueryElasticsearchTest {
Review comment:
This looks like it could be easily converted to a Java test. I know opinions are split on Groovy vs Java tests, but I'm going to recommend converting this one.
##########
File path: nifi-nar-bundles/nifi-elasticsearch-bundle/nifi-elasticsearch-restapi-processors/src/main/java/org/apache/nifi/processors/elasticsearch/PutElasticsearchRecord.java
##########
@@ -414,8 +434,8 @@ private void removeBadRecordFlowFiles(List<FlowFile> bad, ProcessSession session
bad.clear();
}
- private FlowFile indexDocuments(BulkOperation bundle, ProcessSession session, FlowFile input) throws Exception {
- IndexOperationResponse response = clientService.bulk(bundle.getOperationList());
+ private FlowFile indexDocuments(BulkOperation bundle, ProcessContext context, ProcessSession session, FlowFile input) throws Exception {
+ IndexOperationResponse response = clientService.bulk(bundle.getOperationList(), getUrlQueryParameters(context, input));
Review comment:
`final`
##########
File path: nifi-nar-bundles/nifi-elasticsearch-bundle/nifi-elasticsearch-restapi-processors/src/main/java/org/apache/nifi/processors/elasticsearch/PutElasticsearchRecord.java
##########
@@ -72,6 +74,11 @@
@InputRequirement(InputRequirement.Requirement.INPUT_REQUIRED)
@Tags({"json", "elasticsearch", "elasticsearch5", "elasticsearch6", "put", "index", "record"})
@CapabilityDescription("A record-aware Elasticsearch put processor that uses the official Elastic REST client libraries.")
+@DynamicProperty(
+ name = "A URL query parameter",
+ value = "The value to set it to",
+ expressionLanguageScope = ExpressionLanguageScope.FLOWFILE_ATTRIBUTES,
+ description = "Adds the specified property name/value as a query parameter in the Elasticsearch URL used for processing")
Review comment:
Same comments as above
##########
File path: nifi-nar-bundles/nifi-elasticsearch-bundle/nifi-elasticsearch-restapi-processors/src/main/java/org/apache/nifi/processors/elasticsearch/SearchElasticsearch.java
##########
@@ -65,6 +66,11 @@
"Search After/Point in Time queries must include a valid \"sort\" field. The processor will retrieve multiple pages of results " +
"until either no more results are available or the Pagination Keep Alive expiration is reached, after which the query will " +
"restart with the first page of results being retrieved.")
+@DynamicProperty(
+ name = "A URL query parameter",
+ value = "The value to set it to",
+ expressionLanguageScope = ExpressionLanguageScope.FLOWFILE_ATTRIBUTES,
+ description = "Adds the specified property name/value as a query parameter in the Elasticsearch URL used for processing")
Review comment:
Same comments as above
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@nifi.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [nifi] gresockj commented on a change in pull request #4693: NIFI-8001 Elasticsearch Dynamic Parameters
Posted by GitBox <gi...@apache.org>.
gresockj commented on a change in pull request #4693:
URL: https://github.com/apache/nifi/pull/4693#discussion_r724494943
##########
File path: nifi-nar-bundles/nifi-elasticsearch-bundle/nifi-elasticsearch-restapi-processors/src/main/java/org/apache/nifi/processors/elasticsearch/PutElasticsearchRecord.java
##########
@@ -426,22 +429,21 @@ public void onTrigger(ProcessContext context, ProcessSession session) {
session.transfer(input, REL_SUCCESS);
}
- private void removeBadRecordFlowFiles(List<FlowFile> bad, ProcessSession session) {
- for (FlowFile badFlowFile : bad) {
+ private void removeBadRecordFlowFiles(final List<FlowFile> bad, final ProcessSession session) {
+ for (final FlowFile badFlowFile : bad) {
session.remove(badFlowFile);
}
bad.clear();
}
- private FlowFile indexDocuments(BulkOperation bundle, ProcessContext context, ProcessSession session, FlowFile input) throws Exception {
- IndexOperationResponse response = clientService.bulk(bundle.getOperationList(), getUrlQueryParameters(context, input));
+ private FlowFile indexDocuments(final BulkOperation bundle, final ProcessContext context, final ProcessSession session, final FlowFile input) throws Exception {
+ final IndexOperationResponse response = clientService.bulk(bundle.getOperationList(), getUrlQueryParameters(context, input));
if (response.hasErrors()) {
if(logErrors || getLogger().isDebugEnabled()) {
- List<Map<String, Object>> errors = response.getItems();
- ObjectMapper mapper = new ObjectMapper();
- mapper.enable(SerializationFeature.INDENT_OUTPUT);
- String output = String.format("An error was encountered while processing bulk operations. Server response below:%n%n%s", mapper.writeValueAsString(errors));
+ final List<Map<String, Object>> errors = response.getItems();
+ objectMapper.enable(SerializationFeature.INDENT_OUTPUT);
Review comment:
What do you think about putting the `enable()` call in the constructor?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@nifi.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [nifi] ChrisSamo632 commented on a change in pull request #4693: NIFI-8001 Elasticsearch Dynamic Parameters
Posted by GitBox <gi...@apache.org>.
ChrisSamo632 commented on a change in pull request #4693:
URL: https://github.com/apache/nifi/pull/4693#discussion_r724426883
##########
File path: nifi-nar-bundles/nifi-elasticsearch-bundle/nifi-elasticsearch-restapi-processors/src/main/java/org/apache/nifi/processors/elasticsearch/SearchElasticsearch.java
##########
@@ -65,6 +66,11 @@
"Search After/Point in Time queries must include a valid \"sort\" field. The processor will retrieve multiple pages of results " +
"until either no more results are available or the Pagination Keep Alive expiration is reached, after which the query will " +
"restart with the first page of results being retrieved.")
+@DynamicProperty(
+ name = "A URL query parameter",
+ value = "The value to set it to",
+ expressionLanguageScope = ExpressionLanguageScope.FLOWFILE_ATTRIBUTES,
+ description = "Adds the specified property name/value as a query parameter in the Elasticsearch URL used for processing")
Review comment:
Found the relevant comments, addressed
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@nifi.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [nifi] ChrisSamo632 commented on a change in pull request #4693: NIFI-8001 Elasticsearch Dynamic Parameters
Posted by GitBox <gi...@apache.org>.
ChrisSamo632 commented on a change in pull request #4693:
URL: https://github.com/apache/nifi/pull/4693#discussion_r724391043
##########
File path: nifi-nar-bundles/nifi-elasticsearch-bundle/nifi-elasticsearch-restapi-processors/src/main/java/org/apache/nifi/processors/elasticsearch/SearchElasticsearch.java
##########
@@ -65,6 +66,11 @@
"Search After/Point in Time queries must include a valid \"sort\" field. The processor will retrieve multiple pages of results " +
"until either no more results are available or the Pagination Keep Alive expiration is reached, after which the query will " +
"restart with the first page of results being retrieved.")
+@DynamicProperty(
+ name = "A URL query parameter",
+ value = "The value to set it to",
+ expressionLanguageScope = ExpressionLanguageScope.FLOWFILE_ATTRIBUTES,
+ description = "Adds the specified property name/value as a query parameter in the Elasticsearch URL used for processing")
Review comment:
Ah wait, I'd missed a load of comments (GitHub hides a lot by default where there are many comments... I'll look through the rest shortly)
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@nifi.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [nifi] ChrisSamo632 commented on a change in pull request #4693: NIFI-8001 Elasticsearch Dynamic Parameters
Posted by GitBox <gi...@apache.org>.
ChrisSamo632 commented on a change in pull request #4693:
URL: https://github.com/apache/nifi/pull/4693#discussion_r724388507
##########
File path: nifi-nar-bundles/nifi-elasticsearch-bundle/nifi-elasticsearch-restapi-processors/src/main/java/org/apache/nifi/processors/elasticsearch/UpdateByQueryElasticsearch.java
##########
@@ -0,0 +1,64 @@
+/*
+ * 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.elasticsearch;
+
+import org.apache.nifi.annotation.behavior.DynamicProperty;
+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.Tags;
+import org.apache.nifi.elasticsearch.ElasticSearchClientService;
+import org.apache.nifi.elasticsearch.OperationResponse;
+import org.apache.nifi.expression.ExpressionLanguageScope;
+
+import java.util.Map;
+
+@WritesAttributes({
+ @WritesAttribute(attribute = "elasticsearch.update.took", description = "The amount of time that it took to complete the update operation in ms."),
+ @WritesAttribute(attribute = "elasticsearch.update.error", description = "The error message provided by Elasticsearch if there is an error running the update.")
+})
+@InputRequirement(InputRequirement.Requirement.INPUT_ALLOWED)
+@Tags({ "elastic", "elasticsearch", "update", "query"})
+@CapabilityDescription("Update documents in an Elasticsearch index using a query. The query can be loaded from a flowfile body " +
+ "or from the Query parameter.")
+@DynamicProperty(
+ name = "A URL query parameter",
+ value = "The value to set it to",
+ expressionLanguageScope = ExpressionLanguageScope.FLOWFILE_ATTRIBUTES,
Review comment:
Which other comments?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@nifi.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org