You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@nifi.apache.org by ex...@apache.org on 2023/02/13 16:44:36 UTC
[nifi] branch main updated: NIFI-11168 Removed deprecated Elasticsearch Processors and properties
This is an automated email from the ASF dual-hosted git repository.
exceptionfactory pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/nifi.git
The following commit(s) were added to refs/heads/main by this push:
new 7393ce294e NIFI-11168 Removed deprecated Elasticsearch Processors and properties
7393ce294e is described below
commit 7393ce294e4f4c1349738b04d0e823ae2fbdc535
Author: Chris Sampson <ch...@gmail.com>
AuthorDate: Fri Feb 10 21:03:57 2023 +0000
NIFI-11168 Removed deprecated Elasticsearch Processors and properties
This closes #6942
Signed-off-by: David Handermann <ex...@apache.org>
---
nifi-assembly/pom.xml | 6 -
.../elasticsearch/ElasticSearchClientService.java | 13 -
.../ElasticSearchClientServiceImpl.java | 1 -
.../nifi-elasticsearch-nar/pom.xml | 39 -
.../src/main/resources/META-INF/LICENSE | 209 -----
.../src/main/resources/META-INF/NOTICE | 39 -
.../nifi-elasticsearch-processors/pom.xml | 129 ---
.../AbstractElasticsearchHttpProcessor.java | 338 --------
.../AbstractElasticsearchProcessor.java | 95 ---
.../elasticsearch/FetchElasticsearchHttp.java | 338 --------
.../elasticsearch/IdentifierNotFoundException.java | 43 -
.../elasticsearch/PutElasticsearchHttp.java | 399 ---------
.../elasticsearch/PutElasticsearchHttpRecord.java | 812 -------------------
.../elasticsearch/QueryElasticsearchHttp.java | 545 -------------
.../elasticsearch/RetryableException.java | 43 -
.../elasticsearch/ScrollElasticsearchHttp.java | 452 -----------
.../elasticsearch/UnretryableException.java | 43 -
.../services/org.apache.nifi.processor.Processor | 19 -
.../elasticsearch/TestFetchElasticsearchHttp.java | 417 ----------
.../elasticsearch/TestPutElasticsearchHttp.java | 511 ------------
.../TestPutElasticsearchHttpRecord.java | 898 ---------------------
.../elasticsearch/TestQueryElasticsearchHttp.java | 522 ------------
.../TestQueryElasticsearchHttpNoHits.java | 412 ----------
.../elasticsearch/TestScrollElasticsearchHttp.java | 463 -----------
.../src/test/resources/DocumentExample.json | 21 -
.../src/test/resources/query-page1.json | 59 --
.../src/test/resources/query-page2.json | 37 -
.../src/test/resources/query-page3.json | 14 -
.../src/test/resources/scroll-page1.json | 56 --
.../src/test/resources/scroll-page2.json | 36 -
.../src/test/resources/scroll-page3.json | 15 -
nifi-nar-bundles/nifi-elasticsearch-bundle/pom.xml | 2 -
32 files changed, 7026 deletions(-)
diff --git a/nifi-assembly/pom.xml b/nifi-assembly/pom.xml
index fef53d0de5..30d9980089 100644
--- a/nifi-assembly/pom.xml
+++ b/nifi-assembly/pom.xml
@@ -546,12 +546,6 @@ language governing permissions and limitations under the License. -->
<version>2.0.0-SNAPSHOT</version>
<type>nar</type>
</dependency>
- <dependency>
- <groupId>org.apache.nifi</groupId>
- <artifactId>nifi-elasticsearch-nar</artifactId>
- <version>2.0.0-SNAPSHOT</version>
- <type>nar</type>
- </dependency>
<dependency>
<groupId>org.apache.nifi</groupId>
<artifactId>nifi-elasticsearch-client-service-api-nar</artifactId>
diff --git a/nifi-nar-bundles/nifi-elasticsearch-bundle/nifi-elasticsearch-client-service-api/src/main/java/org/apache/nifi/elasticsearch/ElasticSearchClientService.java b/nifi-nar-bundles/nifi-elasticsearch-bundle/nifi-elasticsearch-client-service-api/src/main/java/org/apache/nifi/elasticsearch/ElasticSearchClientService.java
index 76e5772eaf..4664ae1a64 100644
--- a/nifi-nar-bundles/nifi-elasticsearch-bundle/nifi-elasticsearch-client-service-api/src/main/java/org/apache/nifi/elasticsearch/ElasticSearchClientService.java
+++ b/nifi-nar-bundles/nifi-elasticsearch-bundle/nifi-elasticsearch-client-service-api/src/main/java/org/apache/nifi/elasticsearch/ElasticSearchClientService.java
@@ -123,19 +123,6 @@ public interface ElasticSearchClientService extends ControllerService, Verifiabl
.defaultValue("60000")
.addValidator(StandardValidators.POSITIVE_INTEGER_VALIDATOR)
.build();
- /**
- * @deprecated this setting is no longer used and will be removed in a future version.
- * Property retained for now to prevent existing Flows with this processor from breaking upon upgrade.
- */
- @Deprecated
- PropertyDescriptor RETRY_TIMEOUT = new PropertyDescriptor.Builder()
- .name("el-cs-retry-timeout")
- .displayName("Retry timeout")
- .description("Controls the amount of time, in milliseconds, before a timeout occurs when retrying the operation.")
- .required(true)
- .defaultValue("60000")
- .addValidator(StandardValidators.POSITIVE_INTEGER_VALIDATOR)
- .build();
PropertyDescriptor CHARSET = new PropertyDescriptor.Builder()
.name("el-cs-charset")
diff --git a/nifi-nar-bundles/nifi-elasticsearch-bundle/nifi-elasticsearch-client-service/src/main/java/org/apache/nifi/elasticsearch/ElasticSearchClientServiceImpl.java b/nifi-nar-bundles/nifi-elasticsearch-bundle/nifi-elasticsearch-client-service/src/main/java/org/apache/nifi/elasticsearch/ElasticSearchClientServiceImpl.java
index ce0b200d2b..9658c2ce86 100644
--- a/nifi-nar-bundles/nifi-elasticsearch-bundle/nifi-elasticsearch-client-service/src/main/java/org/apache/nifi/elasticsearch/ElasticSearchClientServiceImpl.java
+++ b/nifi-nar-bundles/nifi-elasticsearch-bundle/nifi-elasticsearch-client-service/src/main/java/org/apache/nifi/elasticsearch/ElasticSearchClientServiceImpl.java
@@ -102,7 +102,6 @@ public class ElasticSearchClientServiceImpl extends AbstractControllerService im
props.add(PROXY_CONFIGURATION_SERVICE);
props.add(CONNECT_TIMEOUT);
props.add(SOCKET_TIMEOUT);
- props.add(RETRY_TIMEOUT);
props.add(CHARSET);
props.add(SUPPRESS_NULLS);
diff --git a/nifi-nar-bundles/nifi-elasticsearch-bundle/nifi-elasticsearch-nar/pom.xml b/nifi-nar-bundles/nifi-elasticsearch-bundle/nifi-elasticsearch-nar/pom.xml
deleted file mode 100644
index fb110c5b6b..0000000000
--- a/nifi-nar-bundles/nifi-elasticsearch-bundle/nifi-elasticsearch-nar/pom.xml
+++ /dev/null
@@ -1,39 +0,0 @@
-<?xml version="1.0" encoding="UTF-8"?>
-<!-- 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. -->
-<project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 https://maven.apache.org/xsd/maven-4.0.0.xsd">
- <modelVersion>4.0.0</modelVersion>
- <parent>
- <artifactId>nifi-elasticsearch-bundle</artifactId>
- <groupId>org.apache.nifi</groupId>
- <version>2.0.0-SNAPSHOT</version>
- </parent>
-
- <artifactId>nifi-elasticsearch-nar</artifactId>
- <packaging>nar</packaging>
- <properties>
- <maven.javadoc.skip>true</maven.javadoc.skip>
- <source.skip>true</source.skip>
- </properties>
-
- <dependencies>
- <dependency>
- <groupId>org.apache.nifi</groupId>
- <artifactId>nifi-standard-services-api-nar</artifactId>
- <version>2.0.0-SNAPSHOT</version>
- <type>nar</type>
- </dependency>
- <dependency>
- <groupId>org.apache.nifi</groupId>
- <artifactId>nifi-elasticsearch-processors</artifactId>
- </dependency>
- </dependencies>
-</project>
\ No newline at end of file
diff --git a/nifi-nar-bundles/nifi-elasticsearch-bundle/nifi-elasticsearch-nar/src/main/resources/META-INF/LICENSE b/nifi-nar-bundles/nifi-elasticsearch-bundle/nifi-elasticsearch-nar/src/main/resources/META-INF/LICENSE
deleted file mode 100644
index f3c8ecef26..0000000000
--- a/nifi-nar-bundles/nifi-elasticsearch-bundle/nifi-elasticsearch-nar/src/main/resources/META-INF/LICENSE
+++ /dev/null
@@ -1,209 +0,0 @@
-
- Apache License
- Version 2.0, January 2004
- http://www.apache.org/licenses/
-
- TERMS AND CONDITIONS FOR USE, REPRODUCTION, AND DISTRIBUTION
-
- 1. Definitions.
-
- "License" shall mean the terms and conditions for use, reproduction,
- and distribution as defined by Sections 1 through 9 of this document.
-
- "Licensor" shall mean the copyright owner or entity authorized by
- the copyright owner that is granting the License.
-
- "Legal Entity" shall mean the union of the acting entity and all
- other entities that control, are controlled by, or are under common
- control with that entity. For the purposes of this definition,
- "control" means (i) the power, direct or indirect, to cause the
- direction or management of such entity, whether by contract or
- otherwise, or (ii) ownership of fifty percent (50%) or more of the
- outstanding shares, or (iii) beneficial ownership of such entity.
-
- "You" (or "Your") shall mean an individual or Legal Entity
- exercising permissions granted by this License.
-
- "Source" form shall mean the preferred form for making modifications,
- including but not limited to software source code, documentation
- source, and configuration files.
-
- "Object" form shall mean any form resulting from mechanical
- transformation or translation of a Source form, including but
- not limited to compiled object code, generated documentation,
- and conversions to other media types.
-
- "Work" shall mean the work of authorship, whether in Source or
- Object form, made available under the License, as indicated by a
- copyright notice that is included in or attached to the work
- (an example is provided in the Appendix below).
-
- "Derivative Works" shall mean any work, whether in Source or Object
- form, that is based on (or derived from) the Work and for which the
- editorial revisions, annotations, elaborations, or other modifications
- represent, as a whole, an original work of authorship. For the purposes
- of this License, Derivative Works shall not include works that remain
- separable from, or merely link (or bind by name) to the interfaces of,
- the Work and Derivative Works thereof.
-
- "Contribution" shall mean any work of authorship, including
- the original version of the Work and any modifications or additions
- to that Work or Derivative Works thereof, that is intentionally
- submitted to Licensor for inclusion in the Work by the copyright owner
- or by an individual or Legal Entity authorized to submit on behalf of
- the copyright owner. For the purposes of this definition, "submitted"
- means any form of electronic, verbal, or written communication sent
- to the Licensor or its representatives, including but not limited to
- communication on electronic mailing lists, source code control systems,
- and issue tracking systems that are managed by, or on behalf of, the
- Licensor for the purpose of discussing and improving the Work, but
- excluding communication that is conspicuously marked or otherwise
- designated in writing by the copyright owner as "Not a Contribution."
-
- "Contributor" shall mean Licensor and any individual or Legal Entity
- on behalf of whom a Contribution has been received by Licensor and
- subsequently incorporated within the Work.
-
- 2. Grant of Copyright License. Subject to the terms and conditions of
- this License, each Contributor hereby grants to You a perpetual,
- worldwide, non-exclusive, no-charge, royalty-free, irrevocable
- copyright license to reproduce, prepare Derivative Works of,
- publicly display, publicly perform, sublicense, and distribute the
- Work and such Derivative Works in Source or Object form.
-
- 3. Grant of Patent License. Subject to the terms and conditions of
- this License, each Contributor hereby grants to You a perpetual,
- worldwide, non-exclusive, no-charge, royalty-free, irrevocable
- (except as stated in this section) patent license to make, have made,
- use, offer to sell, sell, import, and otherwise transfer the Work,
- where such license applies only to those patent claims licensable
- by such Contributor that are necessarily infringed by their
- Contribution(s) alone or by combination of their Contribution(s)
- with the Work to which such Contribution(s) was submitted. If You
- institute patent litigation against any entity (including a
- cross-claim or counterclaim in a lawsuit) alleging that the Work
- or a Contribution incorporated within the Work constitutes direct
- or contributory patent infringement, then any patent licenses
- granted to You under this License for that Work shall terminate
- as of the date such litigation is filed.
-
- 4. Redistribution. You may reproduce and distribute copies of the
- Work or Derivative Works thereof in any medium, with or without
- modifications, and in Source or Object form, provided that You
- meet the following conditions:
-
- (a) You must give any other recipients of the Work or
- Derivative Works a copy of this License; and
-
- (b) You must cause any modified files to carry prominent notices
- stating that You changed the files; and
-
- (c) You must retain, in the Source form of any Derivative Works
- that You distribute, all copyright, patent, trademark, and
- attribution notices from the Source form of the Work,
- excluding those notices that do not pertain to any part of
- the Derivative Works; and
-
- (d) If the Work includes a "NOTICE" text file as part of its
- distribution, then any Derivative Works that You distribute must
- include a readable copy of the attribution notices contained
- within such NOTICE file, excluding those notices that do not
- pertain to any part of the Derivative Works, in at least one
- of the following places: within a NOTICE text file distributed
- as part of the Derivative Works; within the Source form or
- documentation, if provided along with the Derivative Works; or,
- within a display generated by the Derivative Works, if and
- wherever such third-party notices normally appear. The contents
- of the NOTICE file are for informational purposes only and
- do not modify the License. You may add Your own attribution
- notices within Derivative Works that You distribute, alongside
- or as an addendum to the NOTICE text from the Work, provided
- that such additional attribution notices cannot be construed
- as modifying the License.
-
- You may add Your own copyright statement to Your modifications and
- may provide additional or different license terms and conditions
- for use, reproduction, or distribution of Your modifications, or
- for any such Derivative Works as a whole, provided Your use,
- reproduction, and distribution of the Work otherwise complies with
- the conditions stated in this License.
-
- 5. Submission of Contributions. Unless You explicitly state otherwise,
- any Contribution intentionally submitted for inclusion in the Work
- by You to the Licensor shall be under the terms and conditions of
- this License, without any additional terms or conditions.
- Notwithstanding the above, nothing herein shall supersede or modify
- the terms of any separate license agreement you may have executed
- with Licensor regarding such Contributions.
-
- 6. Trademarks. This License does not grant permission to use the trade
- names, trademarks, service marks, or product names of the Licensor,
- except as required for reasonable and customary use in describing the
- origin of the Work and reproducing the content of the NOTICE file.
-
- 7. Disclaimer of Warranty. Unless required by applicable law or
- agreed to in writing, Licensor provides the Work (and each
- Contributor provides its Contributions) on an "AS IS" BASIS,
- WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or
- implied, including, without limitation, any warranties or conditions
- of TITLE, NON-INFRINGEMENT, MERCHANTABILITY, or FITNESS FOR A
- PARTICULAR PURPOSE. You are solely responsible for determining the
- appropriateness of using or redistributing the Work and assume any
- risks associated with Your exercise of permissions under this License.
-
- 8. Limitation of Liability. In no event and under no legal theory,
- whether in tort (including negligence), contract, or otherwise,
- unless required by applicable law (such as deliberate and grossly
- negligent acts) or agreed to in writing, shall any Contributor be
- liable to You for damages, including any direct, indirect, special,
- incidental, or consequential damages of any character arising as a
- result of this License or out of the use or inability to use the
- Work (including but not limited to damages for loss of goodwill,
- work stoppage, computer failure or malfunction, or any and all
- other commercial damages or losses), even if such Contributor
- has been advised of the possibility of such damages.
-
- 9. Accepting Warranty or Additional Liability. While redistributing
- the Work or Derivative Works thereof, You may choose to offer,
- and charge a fee for, acceptance of support, warranty, indemnity,
- or other liability obligations and/or rights consistent with this
- License. However, in accepting such obligations, You may act only
- on Your own behalf and on Your sole responsibility, not on behalf
- of any other Contributor, and only if You agree to indemnify,
- defend, and hold each Contributor harmless for any liability
- incurred by, or claims asserted against, such Contributor by reason
- of your accepting any such warranty or additional liability.
-
- END OF TERMS AND CONDITIONS
-
- APPENDIX: How to apply the Apache License to your work.
-
- To apply the Apache License to your work, attach the following
- boilerplate notice, with the fields enclosed by brackets "[]"
- replaced with your own identifying information. (Don't include
- the brackets!) The text should be enclosed in the appropriate
- comment syntax for the file format. We also recommend that a
- file or class name and description of purpose be included on the
- same "printed page" as the copyright notice for easier
- identification within third-party archives.
-
- Copyright [yyyy] [name of copyright owner]
-
- Licensed 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.
-
-APACHE NIFI SUBCOMPONENTS:
-
-The Apache NiFi project contains subcomponents with separate copyright
-notices and license terms. Your use of the source code for the these
-subcomponents is subject to the terms and conditions of the following
-licenses.
diff --git a/nifi-nar-bundles/nifi-elasticsearch-bundle/nifi-elasticsearch-nar/src/main/resources/META-INF/NOTICE b/nifi-nar-bundles/nifi-elasticsearch-bundle/nifi-elasticsearch-nar/src/main/resources/META-INF/NOTICE
deleted file mode 100644
index 13fa655693..0000000000
--- a/nifi-nar-bundles/nifi-elasticsearch-bundle/nifi-elasticsearch-nar/src/main/resources/META-INF/NOTICE
+++ /dev/null
@@ -1,39 +0,0 @@
-nifi-elasticsearch-nar
-Copyright 2015-2020 The Apache Software Foundation
-
-This product includes software developed at
-The Apache Software Foundation (http://www.apache.org/).
-
-******************
-Apache Software License v2
-******************
-
-The following binary components are provided under the Apache Software License v2
-
- (ASLv2) Apache Commons IO
- The following NOTICE information applies:
- Apache Commons IO
- Copyright 2002-2016 The Apache Software Foundation
-
- (ASLv2) Jackson JSON processor
- The following NOTICE information applies:
- # Jackson JSON processor
-
- Jackson is a high-performance, Free/Open Source JSON processing library.
- It was originally written by Tatu Saloranta (tatu.saloranta@iki.fi), and has
- been in development since 2007.
- It is currently developed by a community of developers, as well as supported
- commercially by FasterXML.com.
-
- ## Licensing
-
- Jackson core and extension components may licensed under different licenses.
- To find the details that apply to this artifact see the accompanying LICENSE file.
- For more information, including possible other licensing options, contact
- FasterXML.com (http://fasterxml.com).
-
- ## Credits
-
- A list of contributors may be found from CREDITS file, which is included
- in some artifacts (usually source distributions); but is always available
- from the source code management (SCM) system project uses.
diff --git a/nifi-nar-bundles/nifi-elasticsearch-bundle/nifi-elasticsearch-processors/pom.xml b/nifi-nar-bundles/nifi-elasticsearch-bundle/nifi-elasticsearch-processors/pom.xml
deleted file mode 100644
index b52a47691c..0000000000
--- a/nifi-nar-bundles/nifi-elasticsearch-bundle/nifi-elasticsearch-processors/pom.xml
+++ /dev/null
@@ -1,129 +0,0 @@
-<?xml version="1.0" encoding="UTF-8"?>
-<!-- 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. -->
-<project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 https://maven.apache.org/xsd/maven-4.0.0.xsd">
- <modelVersion>4.0.0</modelVersion>
- <parent>
- <artifactId>nifi-elasticsearch-bundle</artifactId>
- <groupId>org.apache.nifi</groupId>
- <version>2.0.0-SNAPSHOT</version>
- </parent>
-
- <artifactId>nifi-elasticsearch-processors</artifactId>
- <packaging>jar</packaging>
-
- <dependencies>
- <dependency>
- <groupId>org.apache.nifi</groupId>
- <artifactId>nifi-api</artifactId>
- <scope>provided</scope>
- </dependency>
- <dependency>
- <groupId>org.apache.nifi</groupId>
- <artifactId>nifi-properties</artifactId>
- <scope>provided</scope>
- </dependency>
- <dependency>
- <groupId>org.apache.nifi</groupId>
- <artifactId>nifi-utils</artifactId>
- <version>2.0.0-SNAPSHOT</version>
- </dependency>
- <dependency>
- <groupId>org.apache.nifi</groupId>
- <artifactId>nifi-record-path</artifactId>
- <version>2.0.0-SNAPSHOT</version>
- </dependency>
- <dependency>
- <groupId>org.apache.nifi</groupId>
- <artifactId>nifi-record-serialization-service-api</artifactId>
- </dependency>
- <dependency>
- <groupId>org.apache.nifi</groupId>
- <artifactId>nifi-record</artifactId>
- </dependency>
- <dependency>
- <groupId>org.apache.nifi</groupId>
- <artifactId>nifi-proxy-configuration-api</artifactId>
- </dependency>
- <dependency>
- <groupId>org.apache.commons</groupId>
- <artifactId>commons-text</artifactId>
- </dependency>
- <dependency>
- <groupId>org.apache.nifi</groupId>
- <artifactId>nifi-mock</artifactId>
- <version>2.0.0-SNAPSHOT</version>
- <scope>test</scope>
- </dependency>
- <dependency>
- <groupId>org.apache.nifi</groupId>
- <artifactId>nifi-mock-record-utils</artifactId>
- <version>2.0.0-SNAPSHOT</version>
- <scope>test</scope>
- </dependency>
- <dependency>
- <groupId>org.apache.nifi</groupId>
- <artifactId>nifi-schema-registry-service-api</artifactId>
- <version>2.0.0-SNAPSHOT</version>
- <scope>test</scope>
- </dependency>
- <dependency>
- <groupId>org.apache.nifi</groupId>
- <artifactId>nifi-record-serialization-services</artifactId>
- <version>2.0.0-SNAPSHOT</version>
- <scope>test</scope>
- </dependency>
- <dependency>
- <groupId>com.squareup.okhttp3</groupId>
- <artifactId>okhttp</artifactId>
- </dependency>
- <dependency>
- <groupId>org.apache.nifi</groupId>
- <artifactId>nifi-ssl-context-service-api</artifactId>
- </dependency>
- <dependency>
- <groupId>commons-io</groupId>
- <artifactId>commons-io</artifactId>
- </dependency>
- <dependency>
- <groupId>org.apache.nifi</groupId>
- <artifactId>nifi-ssl-context-service</artifactId>
- <scope>test</scope>
- </dependency>
- <dependency>
- <groupId>com.fasterxml.jackson.core</groupId>
- <artifactId>jackson-core</artifactId>
- </dependency>
- <dependency>
- <groupId>com.fasterxml.jackson.core</groupId>
- <artifactId>jackson-databind</artifactId>
- </dependency>
- <dependency>
- <groupId>org.apache.nifi</groupId>
- <artifactId>nifi-standard-record-utils</artifactId>
- <version>2.0.0-SNAPSHOT</version>
- </dependency>
- </dependencies>
-
- <build>
- <plugins>
- <plugin>
- <groupId>org.apache.rat</groupId>
- <artifactId>apache-rat-plugin</artifactId>
- <configuration>
- <excludes combine.children="append">
- <exclude>src/test/resources/*.json</exclude>
- </excludes>
- </configuration>
- </plugin>
- </plugins>
- </build>
-</project>
diff --git a/nifi-nar-bundles/nifi-elasticsearch-bundle/nifi-elasticsearch-processors/src/main/java/org/apache/nifi/processors/elasticsearch/AbstractElasticsearchHttpProcessor.java b/nifi-nar-bundles/nifi-elasticsearch-bundle/nifi-elasticsearch-processors/src/main/java/org/apache/nifi/processors/elasticsearch/AbstractElasticsearchHttpProcessor.java
deleted file mode 100644
index fe56e12b72..0000000000
--- a/nifi-nar-bundles/nifi-elasticsearch-bundle/nifi-elasticsearch-processors/src/main/java/org/apache/nifi/processors/elasticsearch/AbstractElasticsearchHttpProcessor.java
+++ /dev/null
@@ -1,338 +0,0 @@
-/*
- * 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 com.fasterxml.jackson.databind.JsonNode;
-import com.fasterxml.jackson.databind.ObjectMapper;
-import java.io.IOException;
-import java.io.InputStream;
-import java.net.Proxy;
-import java.net.URL;
-import java.util.ArrayList;
-import java.util.Collection;
-import java.util.Collections;
-import java.util.List;
-import java.util.concurrent.TimeUnit;
-import java.util.concurrent.atomic.AtomicReference;
-import okhttp3.Authenticator;
-import okhttp3.Credentials;
-import okhttp3.OkHttpClient;
-import okhttp3.Request;
-import okhttp3.RequestBody;
-import okhttp3.Response;
-import okhttp3.Route;
-import org.apache.commons.text.StringEscapeUtils;
-import org.apache.nifi.components.PropertyDescriptor;
-import org.apache.nifi.components.ValidationContext;
-import org.apache.nifi.components.ValidationResult;
-import org.apache.nifi.expression.ExpressionLanguageScope;
-import org.apache.nifi.logging.ComponentLog;
-import org.apache.nifi.processor.ProcessContext;
-import org.apache.nifi.processor.exception.ProcessException;
-import org.apache.nifi.processor.util.StandardValidators;
-import org.apache.nifi.proxy.ProxyConfiguration;
-import org.apache.nifi.proxy.ProxySpec;
-import org.apache.nifi.ssl.SSLContextService;
-import org.apache.nifi.util.StringUtils;
-
-import javax.net.ssl.SSLContext;
-import javax.net.ssl.X509TrustManager;
-
-/**
- * A base class for Elasticsearch processors that use the HTTP API
- */
-@Deprecated
-public abstract class AbstractElasticsearchHttpProcessor extends AbstractElasticsearchProcessor {
- static final String SOURCE_QUERY_PARAM = "_source";
- static final String QUERY_QUERY_PARAM = "q";
- static final String SORT_QUERY_PARAM = "sort";
- static final String SIZE_QUERY_PARAM = "size";
-
-
- public static final PropertyDescriptor ES_URL = new PropertyDescriptor.Builder()
- .name("elasticsearch-http-url")
- .displayName("Elasticsearch URL")
- .description("Elasticsearch URL which will be connected to, including scheme (http, e.g.), host, and port. The default port for the REST API is 9200.")
- .required(true)
- .addValidator(StandardValidators.URL_VALIDATOR)
- .expressionLanguageSupported(ExpressionLanguageScope.VARIABLE_REGISTRY)
- .build();
-
- public static final PropertyDescriptor PROXY_HOST = new PropertyDescriptor.Builder()
- .name("elasticsearch-http-proxy-host")
- .displayName("Proxy Host")
- .description("The fully qualified hostname or IP address of the proxy server")
- .required(false)
- .expressionLanguageSupported(ExpressionLanguageScope.VARIABLE_REGISTRY)
- .addValidator(StandardValidators.NON_EMPTY_VALIDATOR)
- .build();
-
- public static final PropertyDescriptor PROXY_PORT = new PropertyDescriptor.Builder()
- .name("elasticsearch-http-proxy-port")
- .displayName("Proxy Port")
- .description("The port of the proxy server")
- .required(false)
- .expressionLanguageSupported(ExpressionLanguageScope.VARIABLE_REGISTRY)
- .addValidator(StandardValidators.PORT_VALIDATOR)
- .build();
-
- public static final PropertyDescriptor PROXY_USERNAME = new PropertyDescriptor.Builder()
- .name("proxy-username")
- .displayName("Proxy Username")
- .description("Proxy Username")
- .addValidator(StandardValidators.NON_EMPTY_VALIDATOR)
- .required(false)
- .expressionLanguageSupported(ExpressionLanguageScope.VARIABLE_REGISTRY)
- .build();
- public static final PropertyDescriptor PROXY_PASSWORD = new PropertyDescriptor.Builder()
- .name("proxy-password")
- .displayName("Proxy Password")
- .description("Proxy Password")
- .addValidator(StandardValidators.NON_EMPTY_VALIDATOR)
- .required(false)
- .sensitive(true)
- .expressionLanguageSupported(ExpressionLanguageScope.VARIABLE_REGISTRY)
- .build();
-
- public static final PropertyDescriptor CONNECT_TIMEOUT = new PropertyDescriptor.Builder()
- .name("elasticsearch-http-connect-timeout")
- .displayName("Connection Timeout")
- .description("Max wait time for the connection to the Elasticsearch REST API.")
- .required(true)
- .defaultValue("5 secs")
- .addValidator(StandardValidators.TIME_PERIOD_VALIDATOR)
- .expressionLanguageSupported(ExpressionLanguageScope.VARIABLE_REGISTRY)
- .build();
-
- public static final PropertyDescriptor RESPONSE_TIMEOUT = new PropertyDescriptor.Builder()
- .name("elasticsearch-http-response-timeout")
- .displayName("Response Timeout")
- .description("Max wait time for a response from the Elasticsearch REST API.")
- .required(true)
- .defaultValue("15 secs")
- .addValidator(StandardValidators.TIME_PERIOD_VALIDATOR)
- .expressionLanguageSupported(ExpressionLanguageScope.VARIABLE_REGISTRY)
- .build();
-
- private final AtomicReference<OkHttpClient> okHttpClientAtomicReference = new AtomicReference<>();
-
- final ObjectMapper mapper = new ObjectMapper();
-
- @Override
- protected PropertyDescriptor getSupportedDynamicPropertyDescriptor(String propertyDescriptorName) {
- return new PropertyDescriptor.Builder()
- .name(propertyDescriptorName)
- .required(false)
- .addValidator(StandardValidators.NON_EMPTY_VALIDATOR)
- .expressionLanguageSupported(ExpressionLanguageScope.VARIABLE_REGISTRY)
- .dynamic(true)
- .build();
- }
-
- private static final ProxySpec[] PROXY_SPECS = {ProxySpec.HTTP_AUTH, ProxySpec.SOCKS};
- public static final PropertyDescriptor PROXY_CONFIGURATION_SERVICE
- = ProxyConfiguration.createProxyConfigPropertyDescriptor(true, PROXY_SPECS);
- static final List<PropertyDescriptor> COMMON_PROPERTY_DESCRIPTORS;
-
- static {
- final List<PropertyDescriptor> properties = new ArrayList<>();
- properties.add(ES_URL);
- properties.add(PROP_SSL_CONTEXT_SERVICE);
- properties.add(CHARSET);
- properties.add(USERNAME);
- properties.add(PASSWORD);
- properties.add(CONNECT_TIMEOUT);
- properties.add(RESPONSE_TIMEOUT);
- properties.add(PROXY_CONFIGURATION_SERVICE);
- properties.add(PROXY_HOST);
- properties.add(PROXY_PORT);
- properties.add(PROXY_USERNAME);
- properties.add(PROXY_PASSWORD);
-
- COMMON_PROPERTY_DESCRIPTORS = Collections.unmodifiableList(properties);
- }
-
- @Override
- protected void createElasticsearchClient(ProcessContext context) throws ProcessException {
- okHttpClientAtomicReference.set(null);
-
- OkHttpClient.Builder okHttpClient = new OkHttpClient.Builder();
-
- // Add a proxy if set
- final ProxyConfiguration proxyConfig = ProxyConfiguration.getConfiguration(context, () -> {
- final String proxyHost = context.getProperty(PROXY_HOST).evaluateAttributeExpressions().getValue();
- final Integer proxyPort = context.getProperty(PROXY_PORT).evaluateAttributeExpressions().asInteger();
- if (proxyHost != null && proxyPort != null) {
- final ProxyConfiguration componentProxyConfig = new ProxyConfiguration();
- componentProxyConfig.setProxyType(Proxy.Type.HTTP);
- componentProxyConfig.setProxyServerHost(proxyHost);
- componentProxyConfig.setProxyServerPort(proxyPort);
- componentProxyConfig.setProxyUserName(context.getProperty(PROXY_USERNAME).evaluateAttributeExpressions().getValue());
- componentProxyConfig.setProxyUserPassword(context.getProperty(PROXY_PASSWORD).evaluateAttributeExpressions().getValue());
- return componentProxyConfig;
- }
- return ProxyConfiguration.DIRECT_CONFIGURATION;
- });
-
- if (!Proxy.Type.DIRECT.equals(proxyConfig.getProxyType())) {
- final Proxy proxy = proxyConfig.createProxy();
- okHttpClient.proxy(proxy);
-
- if (proxyConfig.hasCredential()) {
- okHttpClient.proxyAuthenticator(new Authenticator() {
- @Override
- public Request authenticate(Route route, Response response) throws IOException {
- final String credential = Credentials.basic(proxyConfig.getProxyUserName(), proxyConfig.getProxyUserPassword());
- return response.request().newBuilder()
- .header("Proxy-Authorization", credential)
- .build();
- }
- });
- }
- }
-
- // Set timeouts
- okHttpClient.connectTimeout((context.getProperty(CONNECT_TIMEOUT).evaluateAttributeExpressions().asTimePeriod(TimeUnit.MILLISECONDS).intValue()), TimeUnit.MILLISECONDS);
- okHttpClient.readTimeout(context.getProperty(RESPONSE_TIMEOUT).evaluateAttributeExpressions().asTimePeriod(TimeUnit.MILLISECONDS).intValue(), TimeUnit.MILLISECONDS);
-
- // Apply the TLS configuration if present
- final SSLContextService sslService = context.getProperty(PROP_SSL_CONTEXT_SERVICE).asControllerService(SSLContextService.class);
- if (sslService != null) {
- final SSLContext sslContext = sslService.createContext();
- final X509TrustManager trustManager = sslService.createTrustManager();
- okHttpClient.sslSocketFactory(sslContext.getSocketFactory(), trustManager);
- }
-
- okHttpClientAtomicReference.set(okHttpClient.build());
- }
-
- @Override
- protected Collection<ValidationResult> customValidate(ValidationContext validationContext) {
- List<ValidationResult> results = new ArrayList<>(super.customValidate(validationContext));
- if (validationContext.getProperty(PROXY_HOST).isSet() != validationContext.getProperty(PROXY_PORT).isSet()) {
- results.add(new ValidationResult.Builder()
- .valid(false)
- .explanation("Proxy Host and Proxy Port must be both set or empty")
- .subject("Proxy server configuration")
- .build());
- }
-
- ProxyConfiguration.validateProxySpec(validationContext, results, PROXY_SPECS);
-
- return results;
- }
-
- protected OkHttpClient getClient() {
- return okHttpClientAtomicReference.get();
- }
-
- protected boolean isSuccess(int statusCode) {
- return statusCode / 100 == 2;
- }
-
- protected Response sendRequestToElasticsearch(OkHttpClient client, URL url, String username, String password, String verb, RequestBody body) throws IOException {
-
- final ComponentLog log = getLogger();
- Request.Builder requestBuilder = new Request.Builder()
- .url(url);
- if ("get".equalsIgnoreCase(verb)) {
- requestBuilder = requestBuilder.get();
- } else if ("put".equalsIgnoreCase(verb)) {
- requestBuilder = requestBuilder.put(body);
- } else if ("post".equalsIgnoreCase(verb)) {
- requestBuilder = requestBuilder.post(body);
- } else {
- throw new IllegalArgumentException("Elasticsearch REST API verb not supported by this processor: " + verb);
- }
-
- if (!StringUtils.isEmpty(username) && !StringUtils.isEmpty(password)) {
- String credential = Credentials.basic(username, password);
- requestBuilder = requestBuilder.header("Authorization", credential);
- }
- Request httpRequest = requestBuilder.build();
- log.debug("Sending Elasticsearch request to {}", new Object[]{url});
-
- Response responseHttp = client.newCall(httpRequest).execute();
-
- // store the status code and message
- int statusCode = responseHttp.code();
-
- if (statusCode == 0) {
- throw new IllegalStateException("Status code unknown, connection hasn't been attempted.");
- }
-
- log.debug("Received response from Elasticsearch with status code {}", new Object[]{statusCode});
-
- return responseHttp;
- }
-
- protected JsonNode parseJsonResponse(InputStream in) throws IOException {
- return mapper.readTree(in);
- }
-
- protected void buildBulkCommand(StringBuilder sb, String index, String docType, String indexOp, String id, String jsonString) {
- if (indexOp.equalsIgnoreCase("index") || indexOp.equalsIgnoreCase("create")) {
- sb.append("{\"");
- sb.append(indexOp.toLowerCase());
- sb.append("\": { \"_index\": \"");
- sb.append(StringEscapeUtils.escapeJson(index));
- sb.append("\"");
- if (StringUtils.isNotBlank(docType)) {
- sb.append(", \"_type\": \"");
- sb.append(StringEscapeUtils.escapeJson(docType));
- sb.append("\"");
- }
- if (StringUtils.isNotBlank(id)) {
- sb.append(", \"_id\": \"");
- sb.append(StringEscapeUtils.escapeJson(id));
- sb.append("\"");
- }
- sb.append("}}\n");
- sb.append(jsonString);
- sb.append("\n");
- } else if (indexOp.equalsIgnoreCase("upsert") || indexOp.equalsIgnoreCase("update")) {
- sb.append("{\"update\": { \"_index\": \"");
- sb.append(StringEscapeUtils.escapeJson(index));
- sb.append("\"");
- if (StringUtils.isNotBlank(docType)) {
- sb.append(", \"_type\": \"");
- sb.append(StringEscapeUtils.escapeJson(docType));
- sb.append("\"");
- }
- sb.append(", \"_id\": \"");
- sb.append(StringEscapeUtils.escapeJson(id));
- sb.append("\" } }\n");
- sb.append("{\"doc\": ");
- sb.append(jsonString);
- sb.append(", \"doc_as_upsert\": ");
- sb.append(indexOp.equalsIgnoreCase("upsert"));
- sb.append(" }\n");
- } else if (indexOp.equalsIgnoreCase("delete")) {
- sb.append("{\"delete\": { \"_index\": \"");
- sb.append(StringEscapeUtils.escapeJson(index));
- sb.append("\"");
- if (StringUtils.isNotBlank(docType)) {
- sb.append(", \"_type\": \"");
- sb.append(StringEscapeUtils.escapeJson(docType));
- sb.append("\"");
- }
- sb.append(", \"_id\": \"");
- sb.append(StringEscapeUtils.escapeJson(id));
- sb.append("\" } }\n");
- }
- }
-}
diff --git a/nifi-nar-bundles/nifi-elasticsearch-bundle/nifi-elasticsearch-processors/src/main/java/org/apache/nifi/processors/elasticsearch/AbstractElasticsearchProcessor.java b/nifi-nar-bundles/nifi-elasticsearch-bundle/nifi-elasticsearch-processors/src/main/java/org/apache/nifi/processors/elasticsearch/AbstractElasticsearchProcessor.java
deleted file mode 100644
index feac3c4be9..0000000000
--- a/nifi-nar-bundles/nifi-elasticsearch-bundle/nifi-elasticsearch-processors/src/main/java/org/apache/nifi/processors/elasticsearch/AbstractElasticsearchProcessor.java
+++ /dev/null
@@ -1,95 +0,0 @@
-/*
- * 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.components.PropertyDescriptor;
-import org.apache.nifi.components.ValidationContext;
-import org.apache.nifi.components.ValidationResult;
-import org.apache.nifi.expression.ExpressionLanguageScope;
-import org.apache.nifi.processor.AbstractProcessor;
-import org.apache.nifi.processor.ProcessContext;
-import org.apache.nifi.processor.exception.ProcessException;
-import org.apache.nifi.processor.util.StandardValidators;
-import org.apache.nifi.ssl.SSLContextService;
-import org.apache.nifi.util.StringUtils;
-
-import java.util.Collection;
-import java.util.HashSet;
-import java.util.Set;
-
-/**
- * A base class for all Elasticsearch processors
- */
-public abstract class AbstractElasticsearchProcessor extends AbstractProcessor {
-
- public static final PropertyDescriptor PROP_SSL_CONTEXT_SERVICE = new PropertyDescriptor.Builder()
- .name("SSL Context Service")
- .description("The SSL Context Service used to provide client certificate information for TLS/SSL "
- + "connections. This service only applies if the Elasticsearch endpoint(s) have been secured with TLS/SSL.")
- .required(false)
- .identifiesControllerService(SSLContextService.class)
- .build();
-
- protected static final PropertyDescriptor CHARSET = new PropertyDescriptor.Builder()
- .name("Character Set")
- .description("Specifies the character set of the document data.")
- .required(true)
- .defaultValue("UTF-8")
- .addValidator(StandardValidators.CHARACTER_SET_VALIDATOR)
- .expressionLanguageSupported(ExpressionLanguageScope.FLOWFILE_ATTRIBUTES)
- .build();
-
- public static final PropertyDescriptor USERNAME = new PropertyDescriptor.Builder()
- .name("Username")
- .description("Username to access the Elasticsearch cluster")
- .required(false)
- .addValidator(StandardValidators.NON_EMPTY_VALIDATOR)
- .expressionLanguageSupported(ExpressionLanguageScope.VARIABLE_REGISTRY)
- .build();
-
- public static final PropertyDescriptor PASSWORD = new PropertyDescriptor.Builder()
- .name("Password")
- .description("Password to access the Elasticsearch cluster")
- .required(false)
- .sensitive(true)
- .addValidator(StandardValidators.NON_EMPTY_VALIDATOR)
- .expressionLanguageSupported(ExpressionLanguageScope.VARIABLE_REGISTRY)
- .build();
-
- protected abstract void createElasticsearchClient(ProcessContext context) throws ProcessException;
-
- @Override
- protected Collection<ValidationResult> customValidate(ValidationContext validationContext) {
- Set<ValidationResult> results = new HashSet<>();
-
- // Ensure that if username or password is set, then the other is too
- String userName = validationContext.getProperty(USERNAME).evaluateAttributeExpressions().getValue();
- String password = validationContext.getProperty(PASSWORD).evaluateAttributeExpressions().getValue();
- if (StringUtils.isEmpty(userName) != StringUtils.isEmpty(password)) {
- results.add(new ValidationResult.Builder().valid(false).explanation(
- "If username or password is specified, then the other must be specified as well").build());
- }
-
- return results;
- }
-
- public void setup(ProcessContext context) {
- // Create the client if one does not already exist
- createElasticsearchClient(context);
- }
-
-}
diff --git a/nifi-nar-bundles/nifi-elasticsearch-bundle/nifi-elasticsearch-processors/src/main/java/org/apache/nifi/processors/elasticsearch/FetchElasticsearchHttp.java b/nifi-nar-bundles/nifi-elasticsearch-bundle/nifi-elasticsearch-processors/src/main/java/org/apache/nifi/processors/elasticsearch/FetchElasticsearchHttp.java
deleted file mode 100644
index 9732cfde77..0000000000
--- a/nifi-nar-bundles/nifi-elasticsearch-bundle/nifi-elasticsearch-processors/src/main/java/org/apache/nifi/processors/elasticsearch/FetchElasticsearchHttp.java
+++ /dev/null
@@ -1,338 +0,0 @@
-/*
- * 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 com.fasterxml.jackson.databind.JsonNode;
-import okhttp3.HttpUrl;
-import okhttp3.OkHttpClient;
-import okhttp3.Response;
-import okhttp3.ResponseBody;
-import org.apache.commons.lang3.StringUtils;
-import org.apache.nifi.annotation.behavior.DynamicProperty;
-import org.apache.nifi.annotation.behavior.EventDriven;
-import org.apache.nifi.annotation.behavior.InputRequirement;
-import org.apache.nifi.annotation.behavior.SupportsBatching;
-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.DeprecationNotice;
-import org.apache.nifi.annotation.documentation.Tags;
-import org.apache.nifi.annotation.lifecycle.OnScheduled;
-import org.apache.nifi.components.PropertyDescriptor;
-import org.apache.nifi.expression.ExpressionLanguageScope;
-import org.apache.nifi.flowfile.FlowFile;
-import org.apache.nifi.logging.ComponentLog;
-import org.apache.nifi.processor.ProcessContext;
-import org.apache.nifi.processor.ProcessSession;
-import org.apache.nifi.processor.Relationship;
-import org.apache.nifi.processor.exception.ProcessException;
-import org.apache.nifi.processor.util.StandardValidators;
-
-import java.io.ByteArrayInputStream;
-import java.io.IOException;
-import java.net.MalformedURLException;
-import java.net.URL;
-import java.nio.charset.Charset;
-import java.util.ArrayList;
-import java.util.Collections;
-import java.util.HashSet;
-import java.util.List;
-import java.util.Map;
-import java.util.Set;
-import java.util.concurrent.TimeUnit;
-import java.util.stream.Collectors;
-import java.util.stream.Stream;
-
-@Deprecated
-@DeprecationNotice(classNames = {"org.apache.nifi.processors.elasticsearch.GetElasticsearch"},
- reason = "This processor is deprecated and may be removed in future releases.")
-@InputRequirement(InputRequirement.Requirement.INPUT_ALLOWED)
-@EventDriven
-@SupportsBatching
-@Tags({"elasticsearch", "fetch", "read", "get", "http"})
-@CapabilityDescription("Retrieves a document from Elasticsearch using the specified connection properties and the "
- + "identifier of the document to retrieve. Note that the full body of the document will be read into memory before being "
- + "written to a Flow File for transfer.")
-@WritesAttributes({
- @WritesAttribute(attribute = "filename", description = "The filename attribute is set to the document identifier"),
- @WritesAttribute(attribute = "es.index", description = "The Elasticsearch index containing the document"),
- @WritesAttribute(attribute = "es.type", description = "The Elasticsearch document type")
-})
-@DynamicProperty(
- name = "A URL query parameter",
- value = "The value to set it to",
- expressionLanguageScope = ExpressionLanguageScope.VARIABLE_REGISTRY,
- description = "Adds the specified property name/value as a query parameter in the Elasticsearch URL used for processing")
-public class FetchElasticsearchHttp extends AbstractElasticsearchHttpProcessor {
-
- public static final Relationship REL_SUCCESS = new Relationship.Builder()
- .name("success")
- .description("All FlowFiles that are read from Elasticsearch are routed to this relationship.")
- .build();
-
- public static final Relationship REL_FAILURE = new Relationship.Builder()
- .name("failure")
- .description("All FlowFiles that cannot be read from Elasticsearch are routed to this relationship. Note that only incoming "
- + "flow files will be routed to failure.")
- .build();
-
- public static final Relationship REL_RETRY = new Relationship.Builder().name("retry")
- .description("A FlowFile is routed to this relationship if the document cannot be fetched but attempting the operation again may "
- + "succeed. Note that if the processor has no incoming connections, flow files may still be sent to this relationship "
- + "based on the processor properties and the results of the fetch operation.")
- .build();
-
- public static final Relationship REL_NOT_FOUND = new Relationship.Builder().name("not found")
- .description("A FlowFile is routed to this relationship if the specified document does not exist in the Elasticsearch cluster. "
- + "Note that if the processor has no incoming connections, flow files may still be sent to this relationship based "
- + "on the processor properties and the results of the fetch operation.")
- .build();
-
- public static final PropertyDescriptor DOC_ID = new PropertyDescriptor.Builder()
- .name("fetch-es-doc-id")
- .displayName("Document Identifier")
- .description("The identifier of the document to be fetched")
- .required(true)
- .expressionLanguageSupported(ExpressionLanguageScope.FLOWFILE_ATTRIBUTES)
- .addValidator(StandardValidators.NON_EMPTY_VALIDATOR)
- .build();
-
- public static final PropertyDescriptor INDEX = new PropertyDescriptor.Builder()
- .name("fetch-es-index")
- .displayName("Index")
- .description("The name of the index to read from.")
- .required(true)
- .expressionLanguageSupported(ExpressionLanguageScope.FLOWFILE_ATTRIBUTES)
- .addValidator(StandardValidators.NON_EMPTY_VALIDATOR)
- .build();
-
- public static final PropertyDescriptor TYPE = new PropertyDescriptor.Builder()
- .name("fetch-es-type")
- .displayName("Type")
- .description("The type of document/fetch (if unset, the first document matching the "
- + "identifier across _all types will be retrieved). "
- + "This should be unset, '_doc' or '_source' for Elasticsearch 7.0+.")
- .required(false)
- .expressionLanguageSupported(ExpressionLanguageScope.FLOWFILE_ATTRIBUTES)
- .addValidator(StandardValidators.NON_EMPTY_EL_VALIDATOR)
- .build();
-
- public static final PropertyDescriptor FIELDS = new PropertyDescriptor.Builder()
- .name("fetch-es-fields")
- .displayName("Fields")
- .description("A comma-separated list of fields to retrieve from the document. If the Fields property is left blank, "
- + "then the entire document's source will be retrieved.")
- .required(false)
- .expressionLanguageSupported(ExpressionLanguageScope.FLOWFILE_ATTRIBUTES)
- .addValidator(StandardValidators.NON_EMPTY_VALIDATOR)
- .build();
-
- private static final Set<Relationship> relationships;
- private static final List<PropertyDescriptor> propertyDescriptors;
-
- static {
- final Set<Relationship> _rels = new HashSet<>();
- _rels.add(REL_SUCCESS);
- _rels.add(REL_FAILURE);
- _rels.add(REL_RETRY);
- _rels.add(REL_NOT_FOUND);
- relationships = Collections.unmodifiableSet(_rels);
-
- final List<PropertyDescriptor> descriptors = new ArrayList<>(COMMON_PROPERTY_DESCRIPTORS);
- descriptors.add(DOC_ID);
- descriptors.add(INDEX);
- descriptors.add(TYPE);
- descriptors.add(FIELDS);
-
- propertyDescriptors = Collections.unmodifiableList(descriptors);
- }
-
- @Override
- public Set<Relationship> getRelationships() {
- return relationships;
- }
-
- @Override
- public final List<PropertyDescriptor> getSupportedPropertyDescriptors() {
- return propertyDescriptors;
- }
-
- @OnScheduled
- public void setup(ProcessContext context) {
- super.setup(context);
- }
-
- @Override
- public void onTrigger(final ProcessContext context, final ProcessSession session) throws ProcessException {
-
- FlowFile flowFile = null;
- if (context.hasIncomingConnection()) {
- flowFile = session.get();
-
- // If we have no FlowFile, and all incoming connections are self-loops then we can continue on.
- // However, if we have no FlowFile and we have connections coming from other Processors, then
- // we know that we should run only if we have a FlowFile.
- if (flowFile == null && context.hasNonLoopConnection()) {
- return;
- }
- }
-
- OkHttpClient okHttpClient = getClient();
-
- if (flowFile == null) {
- flowFile = session.create();
- }
-
- final String index = context.getProperty(INDEX).evaluateAttributeExpressions(flowFile).getValue();
- final String docId = context.getProperty(DOC_ID).evaluateAttributeExpressions(flowFile).getValue();
- final String docType = context.getProperty(TYPE).evaluateAttributeExpressions(flowFile).getValue();
- final String fields = context.getProperty(FIELDS).isSet()
- ? context.getProperty(FIELDS).evaluateAttributeExpressions(flowFile).getValue()
- : null;
-
- // Authentication
- final String username = context.getProperty(USERNAME).evaluateAttributeExpressions(flowFile).getValue();
- final String password = context.getProperty(PASSWORD).evaluateAttributeExpressions().getValue();
- final Charset charset = Charset.forName(context.getProperty(CHARSET).evaluateAttributeExpressions(flowFile).getValue());
-
- final ComponentLog logger = getLogger();
-
- Response getResponse = null;
-
- try {
- logger.debug("Fetching {}/{}/{} from Elasticsearch", new Object[]{index, docType, docId});
-
- // read the url property from the context
- final String urlstr = StringUtils.trimToEmpty(context.getProperty(ES_URL).evaluateAttributeExpressions().getValue());
- final URL url = buildRequestURL(urlstr, docId, index, docType, fields, context);
- final long startNanos = System.nanoTime();
-
- getResponse = sendRequestToElasticsearch(okHttpClient, url, username, password, "GET", null);
- final int statusCode = getResponse.code();
-
- if (isSuccess(statusCode)) {
- ResponseBody body = getResponse.body();
- final byte[] bodyBytes = body.bytes();
- JsonNode responseJson = parseJsonResponse(new ByteArrayInputStream(bodyBytes));
- boolean found = responseJson.get("found").asBoolean(false);
- String retrievedIndex = responseJson.get("_index").asText();
- String retrievedType = responseJson.get("_type").asText();
- String retrievedId = responseJson.get("_id").asText();
-
- if (found) {
- JsonNode source = responseJson.get("_source");
- flowFile = session.putAttribute(flowFile, "filename", retrievedId);
- flowFile = session.putAttribute(flowFile, "es.index", retrievedIndex);
- flowFile = session.putAttribute(flowFile, "es.type", retrievedType);
- if (source != null) {
- flowFile = session.write(flowFile, out -> {
- out.write(source.toString().getBytes(charset));
- });
- }
- logger.debug("Elasticsearch document " + retrievedId + " fetched, routing to success");
-
- // emit provenance event
- final long millis = TimeUnit.NANOSECONDS.toMillis(System.nanoTime() - startNanos);
- if (context.hasNonLoopConnection()) {
- session.getProvenanceReporter().fetch(flowFile, url.toExternalForm(), millis);
- } else {
- session.getProvenanceReporter().receive(flowFile, url.toExternalForm(), millis);
- }
- session.transfer(flowFile, REL_SUCCESS);
- } else {
- logger.debug("Failed to read {}/{}/{} from Elasticsearch: Document not found",
- new Object[]{index, docType, docId});
-
- // We couldn't find the document, so send it to "not found"
- session.transfer(flowFile, REL_NOT_FOUND);
- }
- } else {
- if (statusCode == 404) {
- logger.warn("Failed to read {}/{}/{} from Elasticsearch: Document not found",
- new Object[]{index, docType, docId});
-
- // We couldn't find the document, so penalize it and send it to "not found"
- session.transfer(flowFile, REL_NOT_FOUND);
- } else {
- // 5xx -> RETRY, but a server error might last a while, so yield
- if (statusCode / 100 == 5) {
-
- logger.warn("Elasticsearch returned code {} with message {}, transferring flow file to retry. This is likely a server problem, yielding...",
- new Object[]{statusCode, getResponse.message()});
- session.transfer(flowFile, REL_RETRY);
- context.yield();
- } else if (context.hasIncomingConnection()) { // 1xx, 3xx, 4xx -> NO RETRY
- logger.warn("Elasticsearch returned code {} with message {}, transferring flow file to failure", new Object[]{statusCode, getResponse.message()});
- session.transfer(flowFile, REL_FAILURE);
- } else {
- logger.warn("Elasticsearch returned code {} with message {}", new Object[]{statusCode, getResponse.message()});
- session.remove(flowFile);
- }
- }
- }
- } catch (IOException ioe) {
- logger.error("Failed to read from Elasticsearch due to {}, this may indicate an error in configuration "
- + "(hosts, username/password, etc.). Routing to retry",
- new Object[]{ioe.getLocalizedMessage()}, ioe);
- if (context.hasIncomingConnection()) {
- session.transfer(flowFile, REL_RETRY);
- } else {
- session.remove(flowFile);
- }
- context.yield();
-
- } catch (Exception e) {
- logger.error("Failed to read {} from Elasticsearch due to {}", new Object[]{flowFile, e.getLocalizedMessage()}, e);
- if (context.hasIncomingConnection()) {
- session.transfer(flowFile, REL_FAILURE);
- } else {
- session.remove(flowFile);
- }
- context.yield();
- } finally {
- if (getResponse != null) {
- getResponse.close();
- }
- }
- }
-
- private URL buildRequestURL(String baseUrl, String docId, String index, String type, String fields, ProcessContext context) throws MalformedURLException {
- if (StringUtils.isEmpty(baseUrl)) {
- throw new MalformedURLException("Base URL cannot be null");
- }
- HttpUrl.Builder builder = HttpUrl.parse(baseUrl).newBuilder();
- builder.addPathSegment(index);
- builder.addPathSegment(StringUtils.isBlank(type) ? "_all" : type);
- builder.addPathSegment(docId);
- if (!StringUtils.isEmpty(fields)) {
- String trimmedFields = Stream.of(fields.split(",")).map(String::trim).collect(Collectors.joining(","));
- builder.addQueryParameter(SOURCE_QUERY_PARAM, trimmedFields);
- }
-
- // Find the user-added properties and set them as query parameters on the URL
- for (Map.Entry<PropertyDescriptor, String> property : context.getProperties().entrySet()) {
- PropertyDescriptor pd = property.getKey();
- if (pd.isDynamic()) {
- if (property.getValue() != null) {
- builder.addQueryParameter(pd.getName(), context.getProperty(pd).evaluateAttributeExpressions().getValue());
- }
- }
- }
-
- return builder.build().url();
- }
-}
diff --git a/nifi-nar-bundles/nifi-elasticsearch-bundle/nifi-elasticsearch-processors/src/main/java/org/apache/nifi/processors/elasticsearch/IdentifierNotFoundException.java b/nifi-nar-bundles/nifi-elasticsearch-bundle/nifi-elasticsearch-processors/src/main/java/org/apache/nifi/processors/elasticsearch/IdentifierNotFoundException.java
deleted file mode 100644
index eb61d3cb34..0000000000
--- a/nifi-nar-bundles/nifi-elasticsearch-bundle/nifi-elasticsearch-processors/src/main/java/org/apache/nifi/processors/elasticsearch/IdentifierNotFoundException.java
+++ /dev/null
@@ -1,43 +0,0 @@
-/*
- * 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;
-
-/**
- * A domain-specific exception for when a valid Elasticsearch document identifier is expected but not found
- */
-@Deprecated
-public class IdentifierNotFoundException extends Exception {
-
- public IdentifierNotFoundException() {
- }
-
- public IdentifierNotFoundException(String message) {
- super(message);
- }
-
- public IdentifierNotFoundException(String message, Throwable cause) {
- super(message, cause);
- }
-
- public IdentifierNotFoundException(Throwable cause) {
- super(cause);
- }
-
- public IdentifierNotFoundException(String message, Throwable cause, boolean enableSuppression, boolean writableStackTrace) {
- super(message, cause, enableSuppression, writableStackTrace);
- }
-}
diff --git a/nifi-nar-bundles/nifi-elasticsearch-bundle/nifi-elasticsearch-processors/src/main/java/org/apache/nifi/processors/elasticsearch/PutElasticsearchHttp.java b/nifi-nar-bundles/nifi-elasticsearch-bundle/nifi-elasticsearch-processors/src/main/java/org/apache/nifi/processors/elasticsearch/PutElasticsearchHttp.java
deleted file mode 100644
index ab54aa688b..0000000000
--- a/nifi-nar-bundles/nifi-elasticsearch-bundle/nifi-elasticsearch-processors/src/main/java/org/apache/nifi/processors/elasticsearch/PutElasticsearchHttp.java
+++ /dev/null
@@ -1,399 +0,0 @@
-/*
- * 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 com.fasterxml.jackson.databind.JsonNode;
-import com.fasterxml.jackson.databind.node.ArrayNode;
-import okhttp3.HttpUrl;
-import okhttp3.MediaType;
-import okhttp3.OkHttpClient;
-import okhttp3.RequestBody;
-import okhttp3.Response;
-import okhttp3.ResponseBody;
-import org.apache.commons.io.IOUtils;
-import org.apache.nifi.annotation.behavior.DynamicProperty;
-import org.apache.nifi.annotation.behavior.EventDriven;
-import org.apache.nifi.annotation.behavior.SystemResourceConsideration;
-import org.apache.nifi.annotation.behavior.InputRequirement;
-import org.apache.nifi.annotation.behavior.SupportsBatching;
-import org.apache.nifi.annotation.behavior.SystemResource;
-import org.apache.nifi.annotation.documentation.CapabilityDescription;
-import org.apache.nifi.annotation.documentation.DeprecationNotice;
-import org.apache.nifi.annotation.documentation.Tags;
-import org.apache.nifi.annotation.lifecycle.OnScheduled;
-import org.apache.nifi.components.PropertyDescriptor;
-import org.apache.nifi.components.ValidationContext;
-import org.apache.nifi.components.ValidationResult;
-import org.apache.nifi.expression.AttributeExpression;
-import org.apache.nifi.expression.ExpressionLanguageScope;
-import org.apache.nifi.flowfile.FlowFile;
-import org.apache.nifi.logging.ComponentLog;
-import org.apache.nifi.processor.ProcessContext;
-import org.apache.nifi.processor.ProcessSession;
-import org.apache.nifi.processor.Relationship;
-import org.apache.nifi.processor.exception.ProcessException;
-import org.apache.nifi.processor.util.StandardValidators;
-import org.apache.nifi.util.StringUtils;
-
-import java.io.ByteArrayInputStream;
-import java.io.IOException;
-import java.net.URL;
-import java.nio.charset.Charset;
-import java.util.ArrayList;
-import java.util.Collection;
-import java.util.Collections;
-import java.util.HashSet;
-import java.util.LinkedList;
-import java.util.List;
-import java.util.Map;
-import java.util.Set;
-
-@Deprecated
-@DeprecationNotice(classNames = {"org.apache.nifi.processors.elasticsearch.PutElasticsearchJson"},
- reason = "This processor is deprecated and may be removed in future releases.")
-@InputRequirement(InputRequirement.Requirement.INPUT_REQUIRED)
-@EventDriven
-@SupportsBatching
-@Tags({"elasticsearch", "insert", "update", "upsert", "delete", "write", "put", "http"})
-@CapabilityDescription("Writes the contents of a FlowFile to Elasticsearch, using the specified parameters such as "
- + "the index to insert into and the type of the document.")
-@DynamicProperty(
- name = "A URL query parameter",
- value = "The value to set it to",
- expressionLanguageScope = ExpressionLanguageScope.VARIABLE_REGISTRY,
- description = "Adds the specified property name/value as a query parameter in the Elasticsearch URL used for processing")
-@SystemResourceConsideration(resource = SystemResource.MEMORY)
-public class PutElasticsearchHttp extends AbstractElasticsearchHttpProcessor {
-
- public static final Relationship REL_SUCCESS = new Relationship.Builder().name("success")
- .description("All FlowFiles that are written to Elasticsearch are routed to this relationship").build();
-
- public static final Relationship REL_FAILURE = new Relationship.Builder().name("failure")
- .description("All FlowFiles that cannot be written to Elasticsearch are routed to this relationship").build();
-
- public static final Relationship REL_RETRY = new Relationship.Builder().name("retry")
- .description("A FlowFile is routed to this relationship if the database cannot be updated but attempting the operation again may succeed")
- .build();
-
- public static final PropertyDescriptor ID_ATTRIBUTE = new PropertyDescriptor.Builder()
- .name("put-es-id-attr")
- .displayName("Identifier Attribute")
- .description("The name of the FlowFile attribute containing the identifier for the document. If the Index Operation is \"index\", "
- + "this property may be left empty or evaluate to an empty value, in which case the document's identifier will be "
- + "auto-generated by Elasticsearch. For all other Index Operations, the attribute must evaluate to a non-empty value.")
- .required(false)
- .expressionLanguageSupported(ExpressionLanguageScope.NONE)
- .addValidator(StandardValidators.ATTRIBUTE_KEY_VALIDATOR)
- .build();
-
- public static final PropertyDescriptor INDEX = new PropertyDescriptor.Builder()
- .name("put-es-index")
- .displayName("Index")
- .description("The name of the index to insert into")
- .required(true)
- .expressionLanguageSupported(ExpressionLanguageScope.FLOWFILE_ATTRIBUTES)
- .addValidator(StandardValidators.createAttributeExpressionLanguageValidator(
- AttributeExpression.ResultType.STRING, true))
- .build();
-
- public static final PropertyDescriptor TYPE = new PropertyDescriptor.Builder()
- .name("put-es-type")
- .displayName("Type")
- .description("The type of this document (required by Elasticsearch versions < 7.0 for indexing and searching). "
- + "This must be unset or '_doc' for Elasticsearch 7.0+.")
- .required(false)
- .expressionLanguageSupported(ExpressionLanguageScope.FLOWFILE_ATTRIBUTES)
- .addValidator(StandardValidators.NON_EMPTY_EL_VALIDATOR)
- .build();
-
- public static final PropertyDescriptor INDEX_OP = new PropertyDescriptor.Builder()
- .name("put-es-index-op")
- .displayName("Index Operation")
- .description("The type of the operation used to index (create, index, update, upsert, delete)")
- .required(true)
- .expressionLanguageSupported(ExpressionLanguageScope.FLOWFILE_ATTRIBUTES)
- .addValidator(StandardValidators.NON_EMPTY_EL_VALIDATOR)
- .defaultValue("index")
- .build();
-
- public static final PropertyDescriptor BATCH_SIZE = new PropertyDescriptor.Builder()
- .name("put-es-batch-size")
- .displayName("Batch Size")
- .description("The preferred number of flow files to put to the database in a single transaction. Note that the contents of the "
- + "flow files will be stored in memory until the bulk operation is performed. Also the results should be returned in the "
- + "same order the flow files were received.")
- .required(true)
- .addValidator(StandardValidators.POSITIVE_INTEGER_VALIDATOR)
- .defaultValue("100")
- .expressionLanguageSupported(ExpressionLanguageScope.VARIABLE_REGISTRY)
- .build();
-
- private static final Set<Relationship> relationships;
- private static final List<PropertyDescriptor> propertyDescriptors;
-
- static {
- final Set<Relationship> _rels = new HashSet<>();
- _rels.add(REL_SUCCESS);
- _rels.add(REL_FAILURE);
- _rels.add(REL_RETRY);
- relationships = Collections.unmodifiableSet(_rels);
-
- final List<PropertyDescriptor> descriptors = new ArrayList<>(COMMON_PROPERTY_DESCRIPTORS);
- descriptors.add(ID_ATTRIBUTE);
- descriptors.add(INDEX);
- descriptors.add(TYPE);
- descriptors.add(BATCH_SIZE);
- descriptors.add(INDEX_OP);
-
- propertyDescriptors = Collections.unmodifiableList(descriptors);
- }
-
- @Override
- public Set<Relationship> getRelationships() {
- return relationships;
- }
-
- @Override
- public final List<PropertyDescriptor> getSupportedPropertyDescriptors() {
- return propertyDescriptors;
- }
-
- @Override
- protected Collection<ValidationResult> customValidate(ValidationContext validationContext) {
- final List<ValidationResult> problems = new ArrayList<>(super.customValidate(validationContext));
- // Since Expression Language is allowed for index operation, we can't guarantee that we can catch
- // all invalid configurations, but we should catch them as soon as we can. For example, if the
- // Identifier Attribute property is empty, the Index Operation must evaluate to "index".
- String idAttribute = validationContext.getProperty(ID_ATTRIBUTE).getValue();
- String indexOp = validationContext.getProperty(INDEX_OP).getValue();
-
- if (StringUtils.isEmpty(idAttribute)) {
- switch (indexOp.toLowerCase()) {
- case "update":
- case "upsert":
- case "delete":
- case "":
- problems.add(new ValidationResult.Builder()
- .valid(false)
- .subject(INDEX_OP.getDisplayName())
- .explanation("If Identifier Attribute is not set, Index Operation must evaluate to one of \"index\" or \"create\"")
- .build());
- break;
- default:
- break;
- }
- }
- return problems;
- }
-
- @OnScheduled
- public void setup(ProcessContext context) {
- super.setup(context);
- }
-
- @Override
- public void onTrigger(final ProcessContext context, final ProcessSession session) throws ProcessException {
- final int batchSize = context.getProperty(BATCH_SIZE).evaluateAttributeExpressions().asInteger();
-
- final List<FlowFile> flowFiles = session.get(batchSize);
- if (flowFiles.isEmpty()) {
- return;
- }
-
- final String id_attribute = context.getProperty(ID_ATTRIBUTE).getValue();
-
- // Authentication
- final String username = context.getProperty(USERNAME).evaluateAttributeExpressions().getValue();
- final String password = context.getProperty(PASSWORD).evaluateAttributeExpressions().getValue();
-
-
- OkHttpClient okHttpClient = getClient();
- final ComponentLog logger = getLogger();
-
- // Keep track of the list of flow files that need to be transferred. As they are transferred, remove them from the list.
- List<FlowFile> flowFilesToTransfer = new LinkedList<>(flowFiles);
-
- final StringBuilder sb = new StringBuilder();
- final String baseUrl = context.getProperty(ES_URL).evaluateAttributeExpressions().getValue().trim();
- if (StringUtils.isEmpty(baseUrl)) {
- throw new ProcessException("Elasticsearch URL is empty or null, this indicates an invalid Expression (missing variables, e.g.)");
- }
- HttpUrl.Builder urlBuilder = HttpUrl.parse(baseUrl).newBuilder().addPathSegment("_bulk");
-
- // Find the user-added properties and set them as query parameters on the URL
- for (Map.Entry<PropertyDescriptor, String> property : context.getProperties().entrySet()) {
- PropertyDescriptor pd = property.getKey();
- if (pd.isDynamic()) {
- if (property.getValue() != null) {
- urlBuilder = urlBuilder.addQueryParameter(pd.getName(), context.getProperty(pd).evaluateAttributeExpressions().getValue());
- }
- }
- }
- final URL url = urlBuilder.build().url();
-
- for (FlowFile file : flowFiles) {
- final String index = context.getProperty(INDEX).evaluateAttributeExpressions(file).getValue();
- final Charset charset = Charset.forName(context.getProperty(CHARSET).evaluateAttributeExpressions(file).getValue());
- if (StringUtils.isEmpty(index)) {
- logger.error("No value for index in for {}, transferring to failure", new Object[]{id_attribute, file});
- flowFilesToTransfer.remove(file);
- session.transfer(file, REL_FAILURE);
- continue;
- }
- final String docType = context.getProperty(TYPE).evaluateAttributeExpressions(file).getValue();
- String indexOp = context.getProperty(INDEX_OP).evaluateAttributeExpressions(file).getValue();
- if (StringUtils.isEmpty(indexOp)) {
- logger.error("No Index operation specified for {}, transferring to failure.", new Object[]{file});
- flowFilesToTransfer.remove(file);
- session.transfer(file, REL_FAILURE);
- continue;
- }
-
- switch (indexOp.toLowerCase()) {
- case "create":
- case "index":
- case "update":
- case "upsert":
- case "delete":
- break;
- default:
- logger.error("Index operation {} not supported for {}, transferring to failure.", new Object[]{indexOp, file});
- flowFilesToTransfer.remove(file);
- session.transfer(file, REL_FAILURE);
- continue;
- }
-
- final String id = (id_attribute != null) ? file.getAttribute(id_attribute) : null;
-
- // The ID must be valid for all operations except "index". For that case,
- // a missing ID indicates one is to be auto-generated by Elasticsearch
- if (id == null && !(indexOp.equalsIgnoreCase("index") || indexOp.equalsIgnoreCase("create"))) {
- logger.error("Index operation {} requires a valid identifier value from a flow file attribute, transferring to failure.",
- new Object[]{indexOp, file});
- flowFilesToTransfer.remove(file);
- session.transfer(file, REL_FAILURE);
- continue;
- }
-
- final StringBuilder json = new StringBuilder();
- session.read(file, in -> {
- json.append(IOUtils.toString(in, charset).replace("\r\n", " ").replace('\n', ' ').replace('\r', ' '));
- });
-
- String jsonString = json.toString();
-
- // Ensure the JSON body is well-formed
- try {
- mapper.readTree(jsonString);
- } catch (IOException e) {
- logger.error("Flow file content is not valid JSON, penalizing and transferring to failure.",
- new Object[]{indexOp, file});
- flowFilesToTransfer.remove(file);
- file = session.penalize(file);
- session.transfer(file, REL_FAILURE);
- continue;
- }
-
- buildBulkCommand(sb, index, docType, indexOp, id, jsonString);
- }
- if (!flowFilesToTransfer.isEmpty()) {
- RequestBody requestBody = RequestBody.create(MediaType.parse("application/json"), sb.toString());
- final Response getResponse;
- try {
- getResponse = sendRequestToElasticsearch(okHttpClient, url, username, password, "PUT", requestBody);
- } catch (final Exception e) {
- logger.error("Routing to {} due to exception: {}", new Object[]{REL_FAILURE.getName(), e}, e);
- flowFilesToTransfer.forEach((flowFileToTransfer) -> {
- flowFileToTransfer = session.penalize(flowFileToTransfer);
- session.transfer(flowFileToTransfer, REL_FAILURE);
- });
- flowFilesToTransfer.clear();
- return;
- }
-
- final int statusCode = getResponse.code();
-
- if (isSuccess(statusCode)) {
- ResponseBody responseBody = getResponse.body();
- try {
- final byte[] bodyBytes = responseBody.bytes();
-
- JsonNode responseJson = parseJsonResponse(new ByteArrayInputStream(bodyBytes));
- boolean errors = responseJson.get("errors").asBoolean(false);
- if (errors) {
- ArrayNode itemNodeArray = (ArrayNode) responseJson.get("items");
- if (itemNodeArray.size() > 0) {
- // All items are returned whether they succeeded or failed, so iterate through the item array
- // at the same time as the flow file list, moving each to success or failure accordingly,
- // but only keep the first error for logging
- String errorReason = null;
- for (int i = itemNodeArray.size() - 1; i >= 0; i--) {
- JsonNode itemNode = itemNodeArray.get(i);
- if (flowFilesToTransfer.size() > i) {
- FlowFile flowFile = flowFilesToTransfer.remove(i);
- int status = itemNode.findPath("status").asInt();
- if (!isSuccess(status)) {
- if (errorReason == null) {
- // Use "result" if it is present; this happens for status codes like 404 Not Found, which may not have an error/reason
- String reason = itemNode.findPath("result").asText();
- if (StringUtils.isEmpty(reason)) {
- // If there was no result, we expect an error with a string description in the "reason" field
- reason = itemNode.findPath("reason").asText();
- }
- errorReason = reason;
- logger.error("Failed to process {} due to {}, transferring to failure",
- new Object[]{flowFile, errorReason});
- }
- flowFile = session.penalize(flowFile);
- flowFile = session.putAttribute(flowFile, "reason", errorReason);
- session.transfer(flowFile, REL_FAILURE);
-
- } else {
- session.transfer(flowFile, REL_SUCCESS);
- // Record provenance event
- session.getProvenanceReporter().send(flowFile, url.toString());
- }
- }
- }
- }
- }
- // Transfer any remaining flowfiles to success
- flowFilesToTransfer.forEach(file -> {
- session.transfer(file, REL_SUCCESS);
- // Record provenance event
- session.getProvenanceReporter().send(file, url.toString());
- });
- } catch (IOException ioe) {
- // Something went wrong when parsing the response, log the error and route to failure
- logger.error("Error parsing Bulk API response: {}", new Object[]{ioe.getMessage()}, ioe);
- session.transfer(flowFilesToTransfer, REL_FAILURE);
- context.yield();
- }
- } else if (statusCode / 100 == 5) {
- // 5xx -> RETRY, but a server error might last a while, so yield
- logger.warn("Elasticsearch returned code {} with message {}, transferring flow file to retry. This is likely a server problem, yielding...",
- new Object[]{statusCode, getResponse.message()});
- session.transfer(flowFilesToTransfer, REL_RETRY);
- context.yield();
- } else { // 1xx, 3xx, 4xx, etc. -> NO RETRY
- logger.warn("Elasticsearch returned code {} with message {}, transferring flow file to failure", new Object[]{statusCode, getResponse.message()});
- session.transfer(flowFilesToTransfer, REL_FAILURE);
- }
- getResponse.close();
- }
- }
-}
diff --git a/nifi-nar-bundles/nifi-elasticsearch-bundle/nifi-elasticsearch-processors/src/main/java/org/apache/nifi/processors/elasticsearch/PutElasticsearchHttpRecord.java b/nifi-nar-bundles/nifi-elasticsearch-bundle/nifi-elasticsearch-processors/src/main/java/org/apache/nifi/processors/elasticsearch/PutElasticsearchHttpRecord.java
deleted file mode 100644
index 97e55ff45e..0000000000
--- a/nifi-nar-bundles/nifi-elasticsearch-bundle/nifi-elasticsearch-processors/src/main/java/org/apache/nifi/processors/elasticsearch/PutElasticsearchHttpRecord.java
+++ /dev/null
@@ -1,812 +0,0 @@
-/*
- * 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 com.fasterxml.jackson.core.JsonFactory;
-import com.fasterxml.jackson.core.JsonGenerator;
-import com.fasterxml.jackson.databind.JsonNode;
-import com.fasterxml.jackson.databind.node.ArrayNode;
-import okhttp3.HttpUrl;
-import okhttp3.MediaType;
-import okhttp3.OkHttpClient;
-import okhttp3.RequestBody;
-import okhttp3.Response;
-import okhttp3.ResponseBody;
-import org.apache.nifi.annotation.behavior.DynamicProperty;
-import org.apache.nifi.annotation.behavior.EventDriven;
-import org.apache.nifi.annotation.behavior.InputRequirement;
-import org.apache.nifi.annotation.behavior.WritesAttribute;
-import org.apache.nifi.annotation.behavior.WritesAttributes;
-import org.apache.nifi.annotation.documentation.CapabilityDescription;
-import org.apache.nifi.annotation.documentation.DeprecationNotice;
-import org.apache.nifi.annotation.documentation.Tags;
-import org.apache.nifi.annotation.lifecycle.OnScheduled;
-import org.apache.nifi.components.AllowableValue;
-import org.apache.nifi.components.PropertyDescriptor;
-import org.apache.nifi.components.ValidationContext;
-import org.apache.nifi.components.ValidationResult;
-import org.apache.nifi.expression.AttributeExpression;
-import org.apache.nifi.expression.ExpressionLanguageScope;
-import org.apache.nifi.flowfile.FlowFile;
-import org.apache.nifi.logging.ComponentLog;
-import org.apache.nifi.processor.ProcessContext;
-import org.apache.nifi.processor.ProcessSession;
-import org.apache.nifi.processor.Relationship;
-import org.apache.nifi.processor.exception.ProcessException;
-import org.apache.nifi.processor.util.StandardValidators;
-import org.apache.nifi.record.path.FieldValue;
-import org.apache.nifi.record.path.RecordPath;
-import org.apache.nifi.record.path.util.RecordPathCache;
-import org.apache.nifi.record.path.validation.RecordPathValidator;
-import org.apache.nifi.schema.access.SchemaNotFoundException;
-import org.apache.nifi.serialization.MalformedRecordException;
-import org.apache.nifi.serialization.RecordReader;
-import org.apache.nifi.serialization.RecordReaderFactory;
-import org.apache.nifi.serialization.RecordSetWriter;
-import org.apache.nifi.serialization.RecordSetWriterFactory;
-import org.apache.nifi.serialization.SimpleDateFormatValidator;
-import org.apache.nifi.serialization.record.DataType;
-import org.apache.nifi.serialization.record.Record;
-import org.apache.nifi.serialization.record.RecordField;
-import org.apache.nifi.serialization.record.RecordFieldType;
-import org.apache.nifi.serialization.record.RecordSchema;
-import org.apache.nifi.serialization.record.type.ArrayDataType;
-import org.apache.nifi.serialization.record.type.ChoiceDataType;
-import org.apache.nifi.serialization.record.type.MapDataType;
-import org.apache.nifi.serialization.record.util.DataTypeUtils;
-import org.apache.nifi.util.StringUtils;
-
-import java.io.ByteArrayInputStream;
-import java.io.ByteArrayOutputStream;
-import java.io.IOException;
-import java.io.InputStream;
-import java.io.OutputStream;
-import java.math.BigInteger;
-import java.net.URL;
-import java.nio.charset.Charset;
-import java.util.ArrayList;
-import java.util.Collection;
-import java.util.Collections;
-import java.util.HashSet;
-import java.util.List;
-import java.util.Map;
-import java.util.Objects;
-import java.util.Optional;
-import java.util.Set;
-
-@Deprecated
-@DeprecationNotice(classNames = {"org.apache.nifi.processors.elasticsearch.PutElasticsearchRecord"},
- reason = "This processor is deprecated and may be removed in future releases.")
-@InputRequirement(InputRequirement.Requirement.INPUT_REQUIRED)
-@EventDriven
-@Tags({"elasticsearch", "insert", "update", "upsert", "delete", "write", "put", "http", "record"})
-@CapabilityDescription("Writes the records from a FlowFile into to Elasticsearch, using the specified parameters such as "
- + "the index to insert into and the type of the document, as well as the operation type (index, upsert, delete, etc.). Note: The Bulk API is used to "
- + "send the records. This means that the entire contents of the incoming flow file are read into memory, and each record is transformed into a JSON document "
- + "which is added to a single HTTP request body. For very large flow files (files with a large number of records, e.g.), this could cause memory usage issues.")
-@WritesAttributes({
- @WritesAttribute(attribute="record.count", description="The number of records in an outgoing FlowFile. This is only populated on the 'success' relationship."),
- @WritesAttribute(attribute="failure.count", description="The number of records found by Elasticsearch to have errors. This is only populated on the 'failure' relationship.")
-})
-@DynamicProperty(
- name = "A URL query parameter",
- value = "The value to set it to",
- expressionLanguageScope = ExpressionLanguageScope.VARIABLE_REGISTRY,
- description = "Adds the specified property name/value as a query parameter in the Elasticsearch URL used for processing")
-public class PutElasticsearchHttpRecord extends AbstractElasticsearchHttpProcessor {
-
- public static final Relationship REL_SUCCESS = new Relationship.Builder().name("success")
- .description("All FlowFiles that are written to Elasticsearch are routed to this relationship").build();
-
- public static final Relationship REL_FAILURE = new Relationship.Builder().name("failure")
- .description("All FlowFiles that cannot be written to Elasticsearch are routed to this relationship").build();
-
- public static final Relationship REL_RETRY = new Relationship.Builder().name("retry")
- .description("A FlowFile is routed to this relationship if the database cannot be updated but attempting the operation again may succeed")
- .build();
-
- static final PropertyDescriptor RECORD_READER = new PropertyDescriptor.Builder()
- .name("put-es-record-record-reader")
- .displayName("Record Reader")
- .description("Specifies the Controller Service to use for parsing incoming data and determining the data's schema.")
- .identifiesControllerService(RecordReaderFactory.class)
- .required(true)
- .build();
-
- static final PropertyDescriptor RECORD_WRITER = new PropertyDescriptor.Builder()
- .name("put-es-record-record-writer")
- .displayName("Record Writer")
- .description("After sending a batch of records, Elasticsearch will report if individual records failed to insert. As an example, this can happen if the record doesn't match the mapping" +
- "for the index it is being inserted into. This property specifies the Controller Service to use for writing out those individual records sent to 'failure'. If this is not set, " +
- "then the whole FlowFile will be routed to failure (including any records which may have been inserted successfully). Note that this will only be used if Elasticsearch reports " +
- "that individual records failed and that in the event that the entire FlowFile fails (e.g. in the event ES is down), the FF will be routed to failure without being interpreted " +
- "by this record writer. If there is an error while attempting to route the failures, the entire FlowFile will be routed to Failure. Also if every record failed individually, " +
- "the entire FlowFile will be routed to Failure without being parsed by the writer.")
- .identifiesControllerService(RecordSetWriterFactory.class)
- .required(false)
- .build();
-
- static final PropertyDescriptor LOG_ALL_ERRORS = new PropertyDescriptor.Builder()
- .name("put-es-record-log-all-errors")
- .displayName("Log all errors in batch")
- .description("After sending a batch of records, Elasticsearch will report if individual records failed to insert. As an example, this can happen if the record doesn't match the mapping " +
- "for the index it is being inserted into. If this is set to true, the processor will log the failure reason for the every failed record. When set to false only the first error " +
- "in the batch will be logged.")
- .addValidator(StandardValidators.BOOLEAN_VALIDATOR)
- .required(false)
- .defaultValue("false")
- .allowableValues("true", "false")
- .build();
-
- static final PropertyDescriptor ID_RECORD_PATH = new PropertyDescriptor.Builder()
- .name("put-es-record-id-path")
- .displayName("Identifier Record Path")
- .description("A RecordPath pointing to a field in the record(s) that contains the identifier for the document. If the Index Operation is \"index\" or \"create\", "
- + "this property may be left empty or evaluate to an empty value, in which case the document's identifier will be "
- + "auto-generated by Elasticsearch. For all other Index Operations, the field's value must be non-empty.")
- .required(false)
- .addValidator(new RecordPathValidator())
- .expressionLanguageSupported(ExpressionLanguageScope.FLOWFILE_ATTRIBUTES)
- .build();
-
- static final PropertyDescriptor INDEX = new PropertyDescriptor.Builder()
- .name("put-es-record-index")
- .displayName("Index")
- .description("The name of the index to insert into")
- .required(true)
- .expressionLanguageSupported(ExpressionLanguageScope.FLOWFILE_ATTRIBUTES)
- .addValidator(StandardValidators.createAttributeExpressionLanguageValidator(
- AttributeExpression.ResultType.STRING, true))
- .build();
-
- static final PropertyDescriptor TYPE = new PropertyDescriptor.Builder()
- .name("put-es-record-type")
- .displayName("Type")
- .description("The type of this document (required by Elasticsearch versions < 7.0 for indexing and searching). "
- + "This must be unset or '_doc' for Elasticsearch 7.0+.")
- .required(false)
- .expressionLanguageSupported(ExpressionLanguageScope.FLOWFILE_ATTRIBUTES)
- .addValidator(StandardValidators.NON_EMPTY_EL_VALIDATOR)
- .build();
-
- static final PropertyDescriptor INDEX_OP = new PropertyDescriptor.Builder()
- .name("put-es-record-index-op")
- .displayName("Index Operation")
- .description("The type of the operation used to index (create, index, update, upsert, delete)")
- .required(true)
- .expressionLanguageSupported(ExpressionLanguageScope.FLOWFILE_ATTRIBUTES)
- .addValidator(StandardValidators.NON_EMPTY_EL_VALIDATOR)
- .defaultValue("index")
- .build();
-
- static final AllowableValue ALWAYS_SUPPRESS = new AllowableValue("always-suppress", "Always Suppress",
- "Fields that are missing (present in the schema but not in the record), or that have a value of null, will not be written out");
-
- static final AllowableValue NEVER_SUPPRESS = new AllowableValue("never-suppress", "Never Suppress",
- "Fields that are missing (present in the schema but not in the record), or that have a value of null, will be written out as a null value");
-
- static final AllowableValue SUPPRESS_MISSING = new AllowableValue("suppress-missing", "Suppress Missing Values",
- "When a field has a value of null, it will be written out. However, if a field is defined in the schema and not present in the record, the field will not be written out.");
-
- static final PropertyDescriptor SUPPRESS_NULLS = new PropertyDescriptor.Builder()
- .name("suppress-nulls")
- .displayName("Suppress Null Values")
- .description("Specifies how the writer should handle a null field")
- .allowableValues(NEVER_SUPPRESS, ALWAYS_SUPPRESS, SUPPRESS_MISSING)
- .defaultValue(NEVER_SUPPRESS.getValue())
- .required(true)
- .build();
-
- static final PropertyDescriptor AT_TIMESTAMP = new PropertyDescriptor.Builder()
- .name("put-es-record-at-timestamp")
- .displayName("@timestamp Value")
- .description("The value to use as the @timestamp field (required for Elasticsearch Data Streams)")
- .required(false)
- .expressionLanguageSupported(ExpressionLanguageScope.FLOWFILE_ATTRIBUTES)
- .addValidator(StandardValidators.NON_EMPTY_EL_VALIDATOR)
- .build();
-
- static final PropertyDescriptor AT_TIMESTAMP_RECORD_PATH = new PropertyDescriptor.Builder()
- .name("put-es-record-at-timestamp-path")
- .displayName("@timestamp Record Path")
- .description("A RecordPath pointing to a field in the record(s) that contains the @timestamp for the document. " +
- "If left blank the @timestamp will be determined using the main @timestamp property")
- .required(false)
- .addValidator(new RecordPathValidator())
- .expressionLanguageSupported(ExpressionLanguageScope.FLOWFILE_ATTRIBUTES)
- .build();
-
- static final PropertyDescriptor DATE_FORMAT = new PropertyDescriptor.Builder()
- .name("Date Format")
- .description("Specifies the format to use when reading/writing Date fields. "
- + "If not specified, the default format '" + RecordFieldType.DATE.getDefaultFormat() + "' is used. "
- + "If specified, the value must match the Java Simple Date Format (for example, MM/dd/yyyy for a two-digit month, followed by "
- + "a two-digit day, followed by a four-digit year, all separated by '/' characters, as in 01/25/2017).")
- .expressionLanguageSupported(ExpressionLanguageScope.VARIABLE_REGISTRY)
- .addValidator(new SimpleDateFormatValidator())
- .required(false)
- .build();
- static final PropertyDescriptor TIME_FORMAT = new PropertyDescriptor.Builder()
- .name("Time Format")
- .description("Specifies the format to use when reading/writing Time fields. "
- + "If not specified, the default format '" + RecordFieldType.TIME.getDefaultFormat() + "' is used. "
- + "If specified, the value must match the Java Simple Date Format (for example, HH:mm:ss for a two-digit hour in 24-hour format, followed by "
- + "a two-digit minute, followed by a two-digit second, all separated by ':' characters, as in 18:04:15).")
- .expressionLanguageSupported(ExpressionLanguageScope.VARIABLE_REGISTRY)
- .addValidator(new SimpleDateFormatValidator())
- .required(false)
- .build();
- static final PropertyDescriptor TIMESTAMP_FORMAT = new PropertyDescriptor.Builder()
- .name("Timestamp Format")
- .description("Specifies the format to use when reading/writing Timestamp fields. "
- + "If not specified, the default format '" + RecordFieldType.TIMESTAMP.getDefaultFormat() + "' is used. "
- + "If specified, the value must match the Java Simple Date Format (for example, MM/dd/yyyy HH:mm:ss for a two-digit month, followed by "
- + "a two-digit day, followed by a four-digit year, all separated by '/' characters; and then followed by a two-digit hour in 24-hour format, followed by "
- + "a two-digit minute, followed by a two-digit second, all separated by ':' characters, as in 01/25/2017 18:04:15).")
- .expressionLanguageSupported(ExpressionLanguageScope.VARIABLE_REGISTRY)
- .addValidator(new SimpleDateFormatValidator())
- .required(false)
- .build();
-
- private static final Set<Relationship> relationships;
- private static final List<PropertyDescriptor> propertyDescriptors;
-
- private volatile RecordPathCache recordPathCache;
-
- private final JsonFactory factory = new JsonFactory();
-
- private volatile String nullSuppression;
- private volatile String dateFormat;
- private volatile String timeFormat;
- private volatile String timestampFormat;
- private volatile Boolean logAllErrors;
-
- static {
- final Set<Relationship> _rels = new HashSet<>();
- _rels.add(REL_SUCCESS);
- _rels.add(REL_FAILURE);
- _rels.add(REL_RETRY);
- relationships = Collections.unmodifiableSet(_rels);
-
- final List<PropertyDescriptor> descriptors = new ArrayList<>(COMMON_PROPERTY_DESCRIPTORS);
- descriptors.add(RECORD_READER);
- descriptors.add(RECORD_WRITER);
- descriptors.add(LOG_ALL_ERRORS);
- descriptors.add(ID_RECORD_PATH);
- descriptors.add(AT_TIMESTAMP_RECORD_PATH);
- descriptors.add(AT_TIMESTAMP);
- descriptors.add(INDEX);
- descriptors.add(TYPE);
- descriptors.add(INDEX_OP);
- descriptors.add(SUPPRESS_NULLS);
- descriptors.add(DATE_FORMAT);
- descriptors.add(TIME_FORMAT);
- descriptors.add(TIMESTAMP_FORMAT);
-
- propertyDescriptors = Collections.unmodifiableList(descriptors);
- }
-
- @Override
- public Set<Relationship> getRelationships() {
- return relationships;
- }
-
- @Override
- public final List<PropertyDescriptor> getSupportedPropertyDescriptors() {
- return propertyDescriptors;
- }
-
- @Override
- protected Collection<ValidationResult> customValidate(ValidationContext validationContext) {
- final List<ValidationResult> problems = new ArrayList<>(super.customValidate(validationContext));
- // Since Expression Language is allowed for index operation, we can't guarantee that we can catch
- // all invalid configurations, but we should catch them as soon as we can. For example, if the
- // Identifier Record Path property is empty, the Index Operation must evaluate to "index" or "create".
- String idPath = validationContext.getProperty(ID_RECORD_PATH).getValue();
- String indexOp = validationContext.getProperty(INDEX_OP).getValue();
-
- if (StringUtils.isEmpty(idPath)) {
- switch (indexOp.toLowerCase()) {
- case "update":
- case "upsert":
- case "delete":
- case "":
- problems.add(new ValidationResult.Builder()
- .valid(false)
- .subject(INDEX_OP.getDisplayName())
- .explanation("If Identifier Record Path is not set, Index Operation must evaluate to one of \"index\" or \"create\"")
- .build());
- break;
- default:
- break;
- }
- }
- return problems;
- }
-
- @OnScheduled
- public void setup(ProcessContext context) {
- super.setup(context);
- recordPathCache = new RecordPathCache(10);
- this.dateFormat = context.getProperty(DATE_FORMAT).evaluateAttributeExpressions().getValue();
- if (this.dateFormat == null) {
- this.dateFormat = RecordFieldType.DATE.getDefaultFormat();
- }
- this.timeFormat = context.getProperty(TIME_FORMAT).evaluateAttributeExpressions().getValue();
- if (this.timeFormat == null) {
- this.timeFormat = RecordFieldType.TIME.getDefaultFormat();
- }
- this.timestampFormat = context.getProperty(TIMESTAMP_FORMAT).evaluateAttributeExpressions().getValue();
- if (this.timestampFormat == null) {
- this.timestampFormat = RecordFieldType.TIMESTAMP.getDefaultFormat();
- }
-
- logAllErrors = context.getProperty(LOG_ALL_ERRORS).asBoolean();
- }
-
- @Override
- public void onTrigger(final ProcessContext context, final ProcessSession session) throws ProcessException {
-
- FlowFile flowFile = session.get();
- if (flowFile == null) {
- return;
- }
-
- final RecordReaderFactory readerFactory = context.getProperty(RECORD_READER).asControllerService(RecordReaderFactory.class);
- final Optional<RecordSetWriterFactory> writerFactoryOptional;
-
- if (context.getProperty(RECORD_WRITER).isSet()) {
- writerFactoryOptional = Optional.of(context.getProperty(RECORD_WRITER).asControllerService(RecordSetWriterFactory.class));
- } else {
- writerFactoryOptional = Optional.empty();
- }
-
- // Authentication
- final String username = context.getProperty(USERNAME).evaluateAttributeExpressions(flowFile).getValue();
- final String password = context.getProperty(PASSWORD).evaluateAttributeExpressions(flowFile).getValue();
-
- OkHttpClient okHttpClient = getClient();
- final ComponentLog logger = getLogger();
-
- final String baseUrl = context.getProperty(ES_URL).evaluateAttributeExpressions().getValue().trim();
- if (StringUtils.isEmpty(baseUrl)) {
- throw new ProcessException("Elasticsearch URL is empty or null, this indicates an invalid Expression (missing variables, e.g.)");
- }
- HttpUrl.Builder urlBuilder = Objects.requireNonNull(HttpUrl.parse(baseUrl)).newBuilder().addPathSegment("_bulk");
-
- // Find the user-added properties and set them as query parameters on the URL
- for (Map.Entry<PropertyDescriptor, String> property : context.getProperties().entrySet()) {
- PropertyDescriptor pd = property.getKey();
- if (pd.isDynamic()) {
- if (property.getValue() != null) {
- urlBuilder = urlBuilder.addQueryParameter(pd.getName(), context.getProperty(pd).evaluateAttributeExpressions().getValue());
- }
- }
- }
- final URL url = urlBuilder.build().url();
-
- final String index = context.getProperty(INDEX).evaluateAttributeExpressions(flowFile).getValue();
- if (StringUtils.isEmpty(index)) {
- logger.error("No value for index in for {}, transferring to failure", flowFile);
- session.transfer(flowFile, REL_FAILURE);
- return;
- }
- final String docType = context.getProperty(TYPE).evaluateAttributeExpressions(flowFile).getValue();
- String indexOp = context.getProperty(INDEX_OP).evaluateAttributeExpressions(flowFile).getValue();
- if (StringUtils.isEmpty(indexOp)) {
- logger.error("No Index operation specified for {}, transferring to failure.", flowFile);
- session.transfer(flowFile, REL_FAILURE);
- return;
- }
-
- switch (indexOp.toLowerCase()) {
- case "create":
- case "index":
- case "update":
- case "upsert":
- case "delete":
- break;
- default:
- logger.error("Index operation {} not supported for {}, transferring to failure.", indexOp, flowFile);
- session.transfer(flowFile, REL_FAILURE);
- return;
- }
-
- this.nullSuppression = context.getProperty(SUPPRESS_NULLS).getValue();
-
- final String idPath = context.getProperty(ID_RECORD_PATH).evaluateAttributeExpressions(flowFile).getValue();
- final RecordPath recordPath = StringUtils.isEmpty(idPath) ? null : recordPathCache.getCompiled(idPath);
- final StringBuilder sb = new StringBuilder();
- final Charset charset = Charset.forName(context.getProperty(CHARSET).evaluateAttributeExpressions(flowFile).getValue());
-
- final String atTimestamp = context.getProperty(AT_TIMESTAMP).evaluateAttributeExpressions(flowFile).getValue();
- final String atTimestampPath = context.getProperty(AT_TIMESTAMP_RECORD_PATH).evaluateAttributeExpressions(flowFile).getValue();
- final RecordPath atPath = StringUtils.isEmpty(atTimestampPath) ? null : recordPathCache.getCompiled(atTimestampPath);
-
- int recordCount = 0;
- try (final InputStream in = session.read(flowFile);
- final RecordReader reader = readerFactory.createRecordReader(flowFile, in, getLogger())) {
-
- Record record;
- while ((record = reader.nextRecord()) != null) {
-
- final String id;
- if (recordPath != null) {
- Optional<FieldValue> idPathValue = recordPath.evaluate(record).getSelectedFields().findFirst();
- if (!idPathValue.isPresent() || idPathValue.get().getValue() == null) {
- throw new IdentifierNotFoundException("Identifier Record Path specified but no value was found, transferring {} to failure.");
- }
- id = idPathValue.get().getValue().toString();
- } else {
- id = null;
- }
-
- final Object timestamp;
- if (atPath != null) {
- final Optional<FieldValue> atPathValue = atPath.evaluate(record).getSelectedFields().findFirst();
- timestamp = !atPathValue.isPresent() || atPathValue.get().getValue() == null ? atTimestamp : atPathValue.get();
- } else {
- timestamp = atTimestamp;
- }
-
- // The ID must be valid for all operations except "index" or "create". For that case,
- // a missing ID indicates one is to be auto-generated by Elasticsearch
- if (id == null && !(indexOp.equalsIgnoreCase("index") || indexOp.equalsIgnoreCase("create"))) {
- throw new IdentifierNotFoundException("Index operation {} requires a valid identifier value from a flow file attribute, transferring to failure.");
- }
-
- final StringBuilder json = new StringBuilder();
-
- ByteArrayOutputStream out = new ByteArrayOutputStream();
- JsonGenerator generator = factory.createGenerator(out);
- writeRecord(record, generator, timestamp);
- generator.flush();
- generator.close();
- json.append(out.toString(charset.name()));
-
- buildBulkCommand(sb, index, docType, indexOp, id, json.toString());
- recordCount++;
- }
- } catch (IdentifierNotFoundException infe) {
- logger.error(infe.getMessage(), flowFile);
- flowFile = session.penalize(flowFile);
- session.transfer(flowFile, REL_FAILURE);
- return;
-
- } catch (final IOException | SchemaNotFoundException | MalformedRecordException e) {
- logger.error("Could not parse incoming data", e);
- flowFile = session.penalize(flowFile);
- session.transfer(flowFile, REL_FAILURE);
- return;
- }
-
- RequestBody requestBody = RequestBody.create(sb.toString(), MediaType.parse("application/json"));
- final Response getResponse;
- try {
- getResponse = sendRequestToElasticsearch(okHttpClient, url, username, password, "PUT", requestBody);
- } catch (final Exception e) {
- logger.error("Routing to {} due to exception: {}", new Object[]{REL_FAILURE.getName(), e}, e);
- flowFile = session.penalize(flowFile);
- session.transfer(flowFile, REL_FAILURE);
- return;
- }
- final int statusCode = getResponse.code();
-
- final Set<Integer> failures = new HashSet<>();
-
- if (isSuccess(statusCode)) {
- try (final ResponseBody responseBody = getResponse.body()) {
- if (responseBody != null) {
- final byte[] bodyBytes = responseBody.bytes();
-
- JsonNode responseJson = parseJsonResponse(new ByteArrayInputStream(bodyBytes));
- boolean errors = responseJson.get("errors").asBoolean(false);
- // ES has no rollback, so if errors occur, log them and route the whole flow file to failure
- if (errors) {
- ArrayNode itemNodeArray = (ArrayNode) responseJson.get("items");
- if (itemNodeArray != null) {
- if (itemNodeArray.size() > 0) {
- // All items are returned whether they succeeded or failed, so iterate through the item array
- // at the same time as the flow file list, moving each to success or failure accordingly,
- // but only keep the first error for logging
- String errorReason = null;
- for (int i = itemNodeArray.size() - 1; i >= 0; i--) {
- JsonNode itemNode = itemNodeArray.get(i);
- int status = itemNode.findPath("status").asInt();
- if (!isSuccess(status)) {
- if (errorReason == null || logAllErrors) {
- // Use "result" if it is present; this happens for status codes like 404 Not Found, which may not have an error/reason
- String reason = itemNode.findPath("result").asText();
- if (StringUtils.isEmpty(reason)) {
- // If there was no result, we expect an error with a string description in the "reason" field
- reason = itemNode.findPath("reason").asText();
- }
- errorReason = reason;
-
- logger.error("Failed to process record {} in FlowFile {} due to {}, transferring to failure",
- i, flowFile, errorReason);
- }
- failures.add(i);
- }
- }
- }
- }
- } else {
- // Everything succeeded, route FF and end
- flowFile = session.putAttribute(flowFile, "record.count", Integer.toString(recordCount));
- session.transfer(flowFile, REL_SUCCESS);
- session.getProvenanceReporter().send(flowFile, url.toString());
- return;
- }
- }
- } catch (IOException ioe) {
- // Something went wrong when parsing the response, log the error and route to failure
- logger.error("Error parsing Bulk API response: {}", new Object[]{ioe.getMessage()}, ioe);
- session.transfer(flowFile, REL_FAILURE);
- context.yield();
- return;
- } finally {
- getResponse.close();
- }
- } else if (statusCode / 100 == 5) {
- // 5xx -> RETRY, but a server error might last a while, so yield
- logger.warn("Elasticsearch returned code {} with message {}, transferring flow file to retry. This is likely a server problem, yielding...",
- statusCode, getResponse.message());
- session.transfer(flowFile, REL_RETRY);
- context.yield();
- return;
- } else { // 1xx, 3xx, 4xx, etc. -> NO RETRY
- logger.warn("Elasticsearch returned code {} with message {}, transferring flow file to failure", statusCode, getResponse.message());
- session.transfer(flowFile, REL_FAILURE);
- return;
- }
-
- // If everything failed or we don't have a writer factory, route the entire original FF to failure.
- if ((!failures.isEmpty() && failures.size() == recordCount ) || !writerFactoryOptional.isPresent()) {
- flowFile = session.putAttribute(flowFile, "failure.count", Integer.toString(failures.size()));
- session.transfer(flowFile, REL_FAILURE);
-
- } else if (!failures.isEmpty()) {
- // Some of the records failed and we have a writer, handle the failures individually.
- final RecordSetWriterFactory writerFactory = writerFactoryOptional.get();
-
- // We know there are a mixture of successes and failures, create FFs for each and rename input FF to avoid confusion.
- final FlowFile successFlowFile = session.create(flowFile);
- final FlowFile failedFlowFile = session.create(flowFile);
-
- // Set up the reader and writers
- try (final OutputStream successOut = session.write(successFlowFile);
- final OutputStream failedOut = session.write(failedFlowFile);
- final InputStream in = session.read(flowFile);
- final RecordReader reader = readerFactory.createRecordReader(flowFile, in, getLogger())) {
-
- final RecordSchema schema = writerFactory.getSchema(flowFile.getAttributes(), reader.getSchema());
-
- try (final RecordSetWriter successWriter = writerFactory.createWriter(getLogger(), schema, successOut, successFlowFile);
- final RecordSetWriter failedWriter = writerFactory.createWriter(getLogger(), schema, failedOut, failedFlowFile)) {
-
- successWriter.beginRecordSet();
- failedWriter.beginRecordSet();
-
- // For each record, if it's in the failure set write it to the failure FF, otherwise it succeeded.
- Record record;
- int i = 0;
- while ((record = reader.nextRecord(false, false)) != null) {
- if (failures.contains(i)) {
- failedWriter.write(record);
- } else {
- successWriter.write(record);
- }
- i++;
- }
- }
- } catch (final IOException | SchemaNotFoundException | MalformedRecordException e) {
- // We failed while handling individual failures. Not much else we can do other than log, and route the whole thing to failure.
- getLogger().error("Failed to process {} during individual record failure handling; route whole FF to failure", flowFile, e);
- session.transfer(flowFile, REL_FAILURE);
- if (successFlowFile != null) {
- session.remove(successFlowFile);
- }
- if (failedFlowFile != null) {
- session.remove(failedFlowFile);
- }
- return;
- }
- session.putAttribute(successFlowFile, "record.count", Integer.toString(recordCount - failures.size()));
-
- // Normal behavior is to output with record.count. In order to not break backwards compatibility, set both here.
- session.putAttribute(failedFlowFile, "record.count", Integer.toString(failures.size()));
- session.putAttribute(failedFlowFile, "failure.count", Integer.toString(failures.size()));
- session.transfer(successFlowFile, REL_SUCCESS);
- session.transfer(failedFlowFile, REL_FAILURE);
- session.remove(flowFile);
- }
- }
-
- private void writeRecord(final Record record, final JsonGenerator generator, final Object atTimestamp) throws IOException {
- final RecordSchema schema = record.getSchema();
-
- generator.writeStartObject();
-
- if (atTimestamp != null && !(atTimestamp instanceof String && StringUtils.isBlank((String) atTimestamp))) {
- final DataType atDataType;
- final Object atValue;
- if (atTimestamp instanceof FieldValue) {
- final FieldValue atField = (FieldValue) atTimestamp;
- atDataType = atField.getField().getDataType();
- atValue = atField.getValue();
- } else {
- atDataType = RecordFieldType.STRING.getDataType();
- atValue = atTimestamp.toString();
- }
-
- final Object outputValue = RecordFieldType.STRING.getDataType().equals(atDataType) ? coerceTimestampStringToLong(atValue.toString()) : atValue;
- final DataType outputDataType = outputValue.equals(atValue) ? atDataType : RecordFieldType.LONG.getDataType();
-
- generator.writeFieldName("@timestamp");
- writeValue(generator, outputValue, "@timestamp", outputDataType);
- }
-
- for (int i = 0; i < schema.getFieldCount(); i++) {
- final RecordField field = schema.getField(i);
- final String fieldName = field.getFieldName();
- final Object value = record.getValue(field);
- if (value == null) {
- if (nullSuppression.equals(NEVER_SUPPRESS.getValue()) || (nullSuppression.equals(SUPPRESS_MISSING.getValue())) && record.getRawFieldNames().contains(fieldName)) {
- generator.writeNullField(fieldName);
- }
-
- continue;
- }
-
- generator.writeFieldName(fieldName);
- final DataType dataType = schema.getDataType(fieldName).get();
-
- writeValue(generator, value, fieldName, dataType);
- }
- generator.writeEndObject();
- }
-
- private Object coerceTimestampStringToLong(final String stringValue) {
- return DataTypeUtils.isLongTypeCompatible(stringValue)
- ? DataTypeUtils.toLong(stringValue, "@timestamp")
- : stringValue;
- }
-
- @SuppressWarnings("unchecked")
- private void writeValue(final JsonGenerator generator, final Object value, final String fieldName, final DataType dataType) throws IOException {
- if (value == null) {
- if (nullSuppression.equals(NEVER_SUPPRESS.getValue()) || ((nullSuppression.equals(SUPPRESS_MISSING.getValue())) && fieldName != null && !fieldName.equals(""))) {
- generator.writeNullField(fieldName);
- }
-
- return;
- }
-
- final DataType chosenDataType = dataType.getFieldType() == RecordFieldType.CHOICE ? DataTypeUtils.chooseDataType(value, (ChoiceDataType) dataType) : dataType;
- final Object coercedValue = DataTypeUtils.convertType(value, chosenDataType, fieldName);
- if (coercedValue == null) {
- generator.writeNull();
- return;
- }
-
- switch (chosenDataType.getFieldType()) {
- case DATE: {
- final String stringValue = DataTypeUtils.toString(coercedValue, () -> DataTypeUtils.getDateFormat(this.dateFormat));
- if (DataTypeUtils.isLongTypeCompatible(stringValue)) {
- generator.writeNumber(DataTypeUtils.toLong(coercedValue, fieldName));
- } else {
- generator.writeString(stringValue);
- }
- break;
- }
- case TIME: {
- final String stringValue = DataTypeUtils.toString(coercedValue, () -> DataTypeUtils.getDateFormat(this.timeFormat));
- if (DataTypeUtils.isLongTypeCompatible(stringValue)) {
- generator.writeNumber(DataTypeUtils.toLong(coercedValue, fieldName));
- } else {
- generator.writeString(stringValue);
- }
- break;
- }
- case TIMESTAMP: {
- final String stringValue = DataTypeUtils.toString(coercedValue, () -> DataTypeUtils.getDateFormat(this.timestampFormat));
- if (DataTypeUtils.isLongTypeCompatible(stringValue)) {
- generator.writeNumber(DataTypeUtils.toLong(coercedValue, fieldName));
- } else {
- generator.writeString(stringValue);
- }
- break;
- }
- case DOUBLE:
- generator.writeNumber(DataTypeUtils.toDouble(coercedValue, fieldName));
- break;
- case FLOAT:
- generator.writeNumber(DataTypeUtils.toFloat(coercedValue, fieldName));
- break;
- case LONG:
- generator.writeNumber(DataTypeUtils.toLong(coercedValue, fieldName));
- break;
- case INT:
- case BYTE:
- case SHORT:
- generator.writeNumber(DataTypeUtils.toInteger(coercedValue, fieldName));
- break;
- case CHAR:
- case STRING:
- generator.writeString(coercedValue.toString());
- break;
- case BIGINT:
- if (coercedValue instanceof Long) {
- generator.writeNumber((Long) coercedValue);
- } else {
- generator.writeNumber((BigInteger) coercedValue);
- }
- break;
- case DECIMAL:
- generator.writeNumber(DataTypeUtils.toBigDecimal(coercedValue, fieldName));
- break;
- case BOOLEAN:
- final String stringValue = coercedValue.toString();
- if ("true".equalsIgnoreCase(stringValue)) {
- generator.writeBoolean(true);
- } else if ("false".equalsIgnoreCase(stringValue)) {
- generator.writeBoolean(false);
- } else {
- generator.writeString(stringValue);
- }
- break;
- case RECORD:
- writeRecord((Record) coercedValue, generator, null);
- break;
- case MAP: {
- final MapDataType mapDataType = (MapDataType) chosenDataType;
- final DataType valueDataType = mapDataType.getValueType();
- final Map<String, ?> map = (Map<String, ?>) coercedValue;
- generator.writeStartObject();
- for (final Map.Entry<String, ?> entry : map.entrySet()) {
- final String mapKey = entry.getKey();
- final Object mapValue = entry.getValue();
- generator.writeFieldName(mapKey);
- writeValue(generator, mapValue, fieldName + "." + mapKey, valueDataType);
- }
- generator.writeEndObject();
- break;
- }
- case ARRAY:
- default:
- if (coercedValue instanceof Object[]) {
- final Object[] values = (Object[]) coercedValue;
- final ArrayDataType arrayDataType = (ArrayDataType) chosenDataType;
- final DataType elementType = arrayDataType.getElementType();
- writeArray(values, fieldName, generator, elementType);
- } else {
- generator.writeString(coercedValue.toString());
- }
- break;
- }
- }
-
- private void writeArray(final Object[] values, final String fieldName, final JsonGenerator generator, final DataType elementType) throws IOException {
- generator.writeStartArray();
- for (final Object element : values) {
- writeValue(generator, element, fieldName, elementType);
- }
- generator.writeEndArray();
- }
-}
-
diff --git a/nifi-nar-bundles/nifi-elasticsearch-bundle/nifi-elasticsearch-processors/src/main/java/org/apache/nifi/processors/elasticsearch/QueryElasticsearchHttp.java b/nifi-nar-bundles/nifi-elasticsearch-bundle/nifi-elasticsearch-processors/src/main/java/org/apache/nifi/processors/elasticsearch/QueryElasticsearchHttp.java
deleted file mode 100644
index bc87ac8432..0000000000
--- a/nifi-nar-bundles/nifi-elasticsearch-bundle/nifi-elasticsearch-processors/src/main/java/org/apache/nifi/processors/elasticsearch/QueryElasticsearchHttp.java
+++ /dev/null
@@ -1,545 +0,0 @@
-/*
- * 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 static org.apache.nifi.flowfile.attributes.CoreAttributes.MIME_TYPE;
-
-import com.fasterxml.jackson.databind.JsonNode;
-import java.util.Arrays;
-import okhttp3.HttpUrl;
-import okhttp3.OkHttpClient;
-import okhttp3.Response;
-import okhttp3.ResponseBody;
-import org.apache.commons.lang3.StringUtils;
-import org.apache.nifi.annotation.behavior.DynamicProperty;
-import org.apache.nifi.annotation.behavior.EventDriven;
-import org.apache.nifi.annotation.behavior.InputRequirement;
-import org.apache.nifi.annotation.behavior.SupportsBatching;
-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.DeprecationNotice;
-import org.apache.nifi.annotation.documentation.Tags;
-import org.apache.nifi.annotation.lifecycle.OnScheduled;
-import org.apache.nifi.components.AllowableValue;
-import org.apache.nifi.components.PropertyDescriptor;
-import org.apache.nifi.expression.ExpressionLanguageScope;
-import org.apache.nifi.flowfile.FlowFile;
-import org.apache.nifi.logging.ComponentLog;
-import org.apache.nifi.processor.ProcessContext;
-import org.apache.nifi.processor.ProcessSession;
-import org.apache.nifi.processor.Relationship;
-import org.apache.nifi.processor.exception.ProcessException;
-import org.apache.nifi.processor.util.StandardValidators;
-
-import java.io.ByteArrayInputStream;
-import java.io.IOException;
-import java.net.MalformedURLException;
-import java.net.URL;
-import java.nio.charset.Charset;
-import java.util.ArrayList;
-import java.util.Collections;
-import java.util.HashMap;
-import java.util.HashSet;
-import java.util.Iterator;
-import java.util.List;
-import java.util.Map;
-import java.util.Map.Entry;
-import java.util.Set;
-import java.util.concurrent.TimeUnit;
-import java.util.stream.Collectors;
-import java.util.stream.Stream;
-
-@Deprecated
-@DeprecationNotice(classNames = {"org.apache.nifi.processors.elasticsearch.PaginatedJsonQueryElasticsearch"},
- reason = "This processor is deprecated and may be removed in future releases.")
-@InputRequirement(InputRequirement.Requirement.INPUT_ALLOWED)
-@EventDriven
-@SupportsBatching
-@Tags({ "elasticsearch", "query", "read", "get", "http" })
-@CapabilityDescription("Queries Elasticsearch using the specified connection properties. "
- + "Note that the full body of each page of documents will be read into memory before being "
- + "written to Flow Files for transfer. Also note that the Elasticsearch max_result_window index "
- + "setting is the upper bound on the number of records that can be retrieved using this query. "
- + "To retrieve more records, use the ScrollElasticsearchHttp processor.")
-@WritesAttributes({
- @WritesAttribute(attribute = "filename", description = "The filename attribute is set to the document identifier"),
- @WritesAttribute(attribute = "es.query.hitcount", description = "The number of hits for a query"),
- @WritesAttribute(attribute = "es.id", description = "The Elasticsearch document identifier"),
- @WritesAttribute(attribute = "es.index", description = "The Elasticsearch index containing the document"),
- @WritesAttribute(attribute = "es.query.url", description = "The Elasticsearch query that was built"),
- @WritesAttribute(attribute = "es.type", description = "The Elasticsearch document type"),
- @WritesAttribute(attribute = "es.result.*", description = "If Target is 'Flow file attributes', the JSON attributes of "
- + "each result will be placed into corresponding attributes with this prefix.") })
-@DynamicProperty(
- name = "A URL query parameter",
- value = "The value to set it to",
- expressionLanguageScope = ExpressionLanguageScope.VARIABLE_REGISTRY,
- description = "Adds the specified property name/value as a query parameter in the Elasticsearch URL used for processing")
-public class QueryElasticsearchHttp extends AbstractElasticsearchHttpProcessor {
-
- public enum QueryInfoRouteStrategy {
- NEVER,
- ALWAYS,
- NOHIT,
- APPEND_AS_ATTRIBUTES
- }
-
- private static final String FROM_QUERY_PARAM = "from";
-
- public static final String TARGET_FLOW_FILE_CONTENT = "Flow file content";
- public static final String TARGET_FLOW_FILE_ATTRIBUTES = "Flow file attributes";
- private static final String ATTRIBUTE_PREFIX = "es.result.";
-
- static final AllowableValue ALWAYS = new AllowableValue(QueryInfoRouteStrategy.ALWAYS.name(), "Always", "Always route Query Info");
- static final AllowableValue NEVER = new AllowableValue(QueryInfoRouteStrategy.NEVER.name(), "Never", "Never route Query Info");
- static final AllowableValue NO_HITS = new AllowableValue(QueryInfoRouteStrategy.NOHIT.name(), "No Hits", "Route Query Info if the Query returns no hits");
- static final AllowableValue APPEND_AS_ATTRIBUTES = new AllowableValue(QueryInfoRouteStrategy.APPEND_AS_ATTRIBUTES.name(), "Append as Attributes",
- "Always append Query Info as attributes, using the existing relationships (does not add the Query Info relationship).");
- public static final Relationship REL_SUCCESS = new Relationship.Builder()
- .name("success")
- .description(
- "All FlowFiles that are read from Elasticsearch are routed to this relationship.")
- .build();
-
- public static final Relationship REL_FAILURE = new Relationship.Builder()
- .name("failure")
- .description(
- "All FlowFiles that cannot be read from Elasticsearch are routed to this relationship. Note that only incoming "
- + "flow files will be routed to failure.").build();
-
- public static final Relationship REL_RETRY = new Relationship.Builder()
- .name("retry")
- .description(
- "A FlowFile is routed to this relationship if the document cannot be fetched but attempting the operation again may "
- + "succeed. Note that if the processor has no incoming connections, flow files may still be sent to this relationship "
- + "based on the processor properties and the results of the fetch operation.")
- .build();
-
- public static final Relationship REL_QUERY_INFO = new Relationship.Builder()
- .name("query-info")
- .description(
- "Depending on the setting of the Routing Strategy for Query Info property, a FlowFile is routed to this relationship with " +
- "the incoming FlowFile's attributes (if present), the number of hits, and the Elasticsearch query")
- .build();
-
- public static final PropertyDescriptor QUERY = new PropertyDescriptor.Builder()
- .name("query-es-query")
- .displayName("Query")
- .description("The Lucene-style query to run against ElasticSearch (e.g., genre:blues AND -artist:muddy)")
- .required(true)
- .expressionLanguageSupported(ExpressionLanguageScope.FLOWFILE_ATTRIBUTES)
- .addValidator(StandardValidators.NON_EMPTY_VALIDATOR)
- .build();
-
- public static final PropertyDescriptor INDEX = new PropertyDescriptor.Builder()
- .name("query-es-index")
- .displayName("Index")
- .description("The name of the index to read from. If the property is unset or set "
- + "to _all, the query will match across all indexes.")
- .required(true)
- .expressionLanguageSupported(ExpressionLanguageScope.FLOWFILE_ATTRIBUTES)
- .addValidator(StandardValidators.NON_EMPTY_VALIDATOR)
- .build();
-
- public static final PropertyDescriptor TYPE = new PropertyDescriptor.Builder()
- .name("query-es-type")
- .displayName("Type")
- .description("The type of document (if unset, the query will be against all types in the _index). "
- + "This should be unset or '_doc' for Elasticsearch 7.0+.")
- .required(false)
- .expressionLanguageSupported(ExpressionLanguageScope.FLOWFILE_ATTRIBUTES)
- .addValidator(StandardValidators.NON_EMPTY_EL_VALIDATOR)
- .build();
-
- public static final PropertyDescriptor FIELDS = new PropertyDescriptor.Builder()
- .name("query-es-fields")
- .displayName("Fields")
- .description(
- "A comma-separated list of fields to retrieve from the document. If the Fields property is left blank, "
- + "then the entire document's source will be retrieved.")
- .required(false)
- .expressionLanguageSupported(ExpressionLanguageScope.FLOWFILE_ATTRIBUTES)
- .addValidator(StandardValidators.NON_EMPTY_VALIDATOR)
- .build();
-
- public static final PropertyDescriptor SORT = new PropertyDescriptor.Builder()
- .name("query-es-sort")
- .displayName("Sort")
- .description(
- "A sort parameter (e.g., timestamp:asc). If the Sort property is left blank, "
- + "then the results will be retrieved in document order.")
- .required(false)
- .expressionLanguageSupported(ExpressionLanguageScope.FLOWFILE_ATTRIBUTES)
- .addValidator(StandardValidators.NON_EMPTY_VALIDATOR)
- .build();
-
- public static final PropertyDescriptor PAGE_SIZE = new PropertyDescriptor.Builder()
- .name("query-es-size")
- .displayName("Page Size")
- .defaultValue("20")
- .description("Determines how many documents to return per page during scrolling.")
- .required(true)
- .expressionLanguageSupported(ExpressionLanguageScope.FLOWFILE_ATTRIBUTES)
- .addValidator(StandardValidators.POSITIVE_INTEGER_VALIDATOR)
- .build();
-
- public static final PropertyDescriptor LIMIT = new PropertyDescriptor.Builder()
- .name("query-es-limit")
- .displayName("Limit")
- .description("If set, limits the number of results that will be returned.")
- .required(false)
- .expressionLanguageSupported(ExpressionLanguageScope.FLOWFILE_ATTRIBUTES)
- .addValidator(StandardValidators.POSITIVE_INTEGER_VALIDATOR)
- .build();
-
- public static final PropertyDescriptor TARGET = new PropertyDescriptor.Builder()
- .name("query-es-target")
- .displayName("Target")
- .description(
- "Indicates where the results should be placed. In the case of 'Flow file content', the JSON "
- + "response will be written as the content of the flow file. In the case of 'Flow file attributes', "
- + "the original flow file (if applicable) will be cloned for each result, and all return fields will be placed "
- + "in a flow file attribute of the same name, but prefixed by 'es.result.'")
- .required(true)
- .expressionLanguageSupported(ExpressionLanguageScope.NONE)
- .defaultValue(TARGET_FLOW_FILE_CONTENT)
- .allowableValues(TARGET_FLOW_FILE_CONTENT, TARGET_FLOW_FILE_ATTRIBUTES)
- .addValidator(StandardValidators.NON_EMPTY_VALIDATOR)
- .build();
-
- public static final PropertyDescriptor ROUTING_QUERY_INFO_STRATEGY = new PropertyDescriptor.Builder()
- .name("routing-query-info-strategy")
- .displayName("Routing Strategy for Query Info")
- .description("Specifies when to generate and route Query Info after a successful query")
- .expressionLanguageSupported(ExpressionLanguageScope.NONE)
- .allowableValues(ALWAYS, NEVER, NO_HITS, APPEND_AS_ATTRIBUTES)
- .defaultValue(NEVER.getValue())
- .required(false)
- .build();
-
- private volatile Set<Relationship> relationships = new HashSet<>(Arrays.asList(new Relationship[] {REL_SUCCESS, REL_FAILURE, REL_RETRY}));
- private static final List<PropertyDescriptor> propertyDescriptors;
- private QueryInfoRouteStrategy queryInfoRouteStrategy = QueryInfoRouteStrategy.NEVER;
-
- static {
- final List<PropertyDescriptor> descriptors = new ArrayList<>(COMMON_PROPERTY_DESCRIPTORS);
- descriptors.add(QUERY);
- descriptors.add(PAGE_SIZE);
- descriptors.add(INDEX);
- descriptors.add(TYPE);
- descriptors.add(FIELDS);
- descriptors.add(SORT);
- descriptors.add(LIMIT);
- descriptors.add(TARGET);
- descriptors.add(ROUTING_QUERY_INFO_STRATEGY);
-
- propertyDescriptors = Collections.unmodifiableList(descriptors);
- }
-
- @Override
- public Set<Relationship> getRelationships() {
- return relationships;
- }
-
- @Override
- public final List<PropertyDescriptor> getSupportedPropertyDescriptors() {
- return propertyDescriptors;
- }
-
- @OnScheduled
- public void setup(ProcessContext context) {
- super.setup(context);
- }
-
- @Override
- public void onPropertyModified(final PropertyDescriptor descriptor, final String oldValue, final String newValue) {
-
- if (ROUTING_QUERY_INFO_STRATEGY.equals(descriptor)) {
- final Set<Relationship> relationshipSet = new HashSet<>();
- relationshipSet.add(REL_SUCCESS);
- relationshipSet.add(REL_FAILURE);
- relationshipSet.add(REL_RETRY);
-
- if (ALWAYS.getValue().equalsIgnoreCase(newValue) || NO_HITS.getValue().equalsIgnoreCase(newValue)) {
- relationshipSet.add(REL_QUERY_INFO);
- }
- this.queryInfoRouteStrategy = QueryInfoRouteStrategy.valueOf(newValue);
- this.relationships = relationshipSet;
- }
- }
-
- @Override
- public void onTrigger(final ProcessContext context, final ProcessSession session)
- throws ProcessException {
-
- FlowFile flowFile = null;
- if (context.hasIncomingConnection()) {
- flowFile = session.get();
-
- // If we have no FlowFile, and all incoming connections are self-loops then we can
- // continue on.
- // However, if we have no FlowFile and we have connections coming from other Processors,
- // then
- // we know that we should run only if we have a FlowFile.
- if (flowFile == null && context.hasNonLoopConnection()) {
- return;
- }
- }
-
- OkHttpClient okHttpClient = getClient();
-
- final String index = context.getProperty(INDEX).evaluateAttributeExpressions(flowFile)
- .getValue();
- final String query = context.getProperty(QUERY).evaluateAttributeExpressions(flowFile)
- .getValue();
- final String docType = context.getProperty(TYPE).evaluateAttributeExpressions(flowFile)
- .getValue();
- final int pageSize = context.getProperty(PAGE_SIZE).evaluateAttributeExpressions(flowFile)
- .asInteger();
- final Integer limit = context.getProperty(LIMIT).isSet() ? context.getProperty(LIMIT)
- .evaluateAttributeExpressions(flowFile).asInteger() : null;
- final String fields = context.getProperty(FIELDS).isSet() ? context.getProperty(FIELDS)
- .evaluateAttributeExpressions(flowFile).getValue() : null;
- final String sort = context.getProperty(SORT).isSet() ? context.getProperty(SORT)
- .evaluateAttributeExpressions(flowFile).getValue() : null;
- final boolean targetIsContent = context.getProperty(TARGET).getValue()
- .equals(TARGET_FLOW_FILE_CONTENT);
-
- // Authentication
- final String username = context.getProperty(USERNAME).evaluateAttributeExpressions().getValue();
- final String password = context.getProperty(PASSWORD).evaluateAttributeExpressions().getValue();
- final Charset charset = Charset.forName(context.getProperty(CHARSET).evaluateAttributeExpressions(flowFile).getValue());
-
- final ComponentLog logger = getLogger();
-
- int fromIndex = 0;
- int numResults = 0;
-
- try {
- logger.debug("Querying {}/{} from Elasticsearch: {}", new Object[] { index, docType,
- query });
-
- final long startNanos = System.nanoTime();
- // read the url property from the context
- final String urlstr = StringUtils.trimToEmpty(context.getProperty(ES_URL).evaluateAttributeExpressions().getValue());
-
- boolean hitLimit = false;
- do {
- int mPageSize = pageSize;
- if (limit != null && limit <= (fromIndex + pageSize)) {
- mPageSize = limit - fromIndex;
- hitLimit = true;
- }
-
- final URL queryUrl = buildRequestURL(urlstr, query, index, docType, fields, sort,
- mPageSize, fromIndex, context);
-
- final Response getResponse = sendRequestToElasticsearch(okHttpClient, queryUrl,
- username, password, "GET", null);
- numResults = this.getPage(getResponse, queryUrl, context, session, flowFile,
- logger, startNanos, targetIsContent, numResults, charset);
- fromIndex += pageSize;
- getResponse.close();
- }
- while (numResults > 0 && !hitLimit);
-
- if (flowFile != null) {
- session.remove(flowFile);
- }
- } catch (IOException ioe) {
- logger.error(
- "Failed to read from Elasticsearch due to {}, this may indicate an error in configuration "
- + "(hosts, username/password, etc.). Routing to retry",
- new Object[] { ioe.getLocalizedMessage() }, ioe);
- if (flowFile != null) {
- session.transfer(flowFile, REL_RETRY);
- }
- context.yield();
-
- } catch (RetryableException e) {
- logger.error(e.getMessage(), new Object[] { e.getLocalizedMessage() }, e);
- if (flowFile != null) {
- session.transfer(flowFile, REL_RETRY);
- }
- context.yield();
- } catch (Exception e) {
- logger.error("Failed to read {} from Elasticsearch due to {}", new Object[] { flowFile,
- e.getLocalizedMessage() }, e);
- if (flowFile != null) {
- session.transfer(flowFile, REL_FAILURE);
- }
- context.yield();
- }
- }
-
- private int getPage(final Response getResponse, final URL url, final ProcessContext context,
- final ProcessSession session, FlowFile flowFile, final ComponentLog logger,
- final long startNanos, boolean targetIsContent, int priorResultCount, Charset charset)
- throws IOException {
- List<FlowFile> page = new ArrayList<>();
- final int statusCode = getResponse.code();
-
- if (isSuccess(statusCode)) {
- ResponseBody body = getResponse.body();
- final byte[] bodyBytes = body.bytes();
- JsonNode responseJson = parseJsonResponse(new ByteArrayInputStream(bodyBytes));
- JsonNode hits = responseJson.get("hits").get("hits");
-
- // if there are no hits, and there have never been any hits in this run ( priorResultCount ) and
- // we are in NOHIT or ALWAYS, send the query info
- if ( (hits.size() == 0 && priorResultCount == 0 && queryInfoRouteStrategy == QueryInfoRouteStrategy.NOHIT)
- || queryInfoRouteStrategy == QueryInfoRouteStrategy.ALWAYS) {
- FlowFile queryInfo = flowFile == null ? session.create() : session.create(flowFile);
- queryInfo = session.putAttribute(queryInfo, "es.query.url", url.toExternalForm());
- queryInfo = session.putAttribute(queryInfo, "es.query.hitcount", String.valueOf(hits.size()));
- queryInfo = session.putAttribute(queryInfo, MIME_TYPE.key(), "application/json");
- session.transfer(queryInfo,REL_QUERY_INFO);
- }
-
- for(int i = 0; i < hits.size(); i++) {
- JsonNode hit = hits.get(i);
- String retrievedId = hit.get("_id").asText();
- String retrievedIndex = hit.get("_index").asText();
- String retrievedType = hit.get("_type").asText();
-
- FlowFile documentFlowFile = null;
- if (flowFile != null) {
- documentFlowFile = targetIsContent ? session.create(flowFile) : session.clone(flowFile);
- } else {
- documentFlowFile = session.create();
- }
-
- if (queryInfoRouteStrategy == QueryInfoRouteStrategy.APPEND_AS_ATTRIBUTES) {
- documentFlowFile = session.putAttribute(documentFlowFile, "es.query.hitcount", String.valueOf(hits.size()));
- }
-
- JsonNode source = hit.get("_source");
- documentFlowFile = session.putAttribute(documentFlowFile, "es.id", retrievedId);
- documentFlowFile = session.putAttribute(documentFlowFile, "es.index", retrievedIndex);
- documentFlowFile = session.putAttribute(documentFlowFile, "es.type", retrievedType);
- documentFlowFile = session.putAttribute(documentFlowFile, "es.query.url", url.toExternalForm());
-
- if (targetIsContent) {
- documentFlowFile = session.putAttribute(documentFlowFile, "filename", retrievedId);
- documentFlowFile = session.putAttribute(documentFlowFile, "mime.type", "application/json");
- documentFlowFile = session.write(documentFlowFile, out -> {
- out.write(source.toString().getBytes(charset));
- });
- } else {
- Map<String, String> attributes = new HashMap<>();
- for(Iterator<Entry<String, JsonNode>> it = source.fields(); it.hasNext(); ) {
- Entry<String, JsonNode> entry = it.next();
-
- String textValue = "";
- if(entry.getValue().isArray()){
- ArrayList<String> text_values = new ArrayList<String>();
- for(Iterator<JsonNode> items = entry.getValue().iterator(); items.hasNext(); ) {
- text_values.add(items.next().asText());
- }
- textValue = StringUtils.join(text_values, ',');
- } else {
- textValue = entry.getValue().asText();
- }
- attributes.put(ATTRIBUTE_PREFIX + entry.getKey(), textValue);
- }
- documentFlowFile = session.putAllAttributes(documentFlowFile, attributes);
- }
- page.add(documentFlowFile);
- }
-
- logger.debug("Elasticsearch retrieved " + responseJson.size() + " documents, routing to success");
- // If we want to append query info as attributes but there were no hits,
- // pass along the original, if present.
- if (queryInfoRouteStrategy == QueryInfoRouteStrategy.APPEND_AS_ATTRIBUTES && page.isEmpty()
- && flowFile != null) {
- FlowFile documentFlowFile = null;
- documentFlowFile = targetIsContent ? session.create(flowFile) : session.clone(flowFile);
- documentFlowFile = session.putAttribute(documentFlowFile, "es.query.hitcount", String.valueOf(hits.size()));
- documentFlowFile = session.putAttribute(documentFlowFile, "es.query.url", url.toExternalForm());
- session.transfer(documentFlowFile, REL_SUCCESS);
- } else {
- session.transfer(page, REL_SUCCESS);
- }
- } else {
- try {
- // 5xx -> RETRY, but a server error might last a while, so yield
- if (statusCode / 100 == 5) {
- throw new RetryableException(String.format("Elasticsearch returned code %s with message %s, transferring flow file to retry. This is likely a server problem, yielding...",
- statusCode, getResponse.message()));
- } else if (context.hasIncomingConnection()) { // 1xx, 3xx, 4xx -> NO RETRY
- throw new UnretryableException(String.format("Elasticsearch returned code %s with message %s, transferring flow file to failure",
- statusCode, getResponse.message()));
- } else {
- logger.warn("Elasticsearch returned code {} with message {}", new Object[]{statusCode, getResponse.message()});
- }
- } finally {
- if (!page.isEmpty()) {
- session.remove(page);
- page.clear();
- }
- }
- }
-
- // emit provenance event
- final long millis = TimeUnit.NANOSECONDS.toMillis(System.nanoTime() - startNanos);
- if (!page.isEmpty()) {
- if (context.hasNonLoopConnection()) {
- page.forEach(f -> session.getProvenanceReporter().fetch(f, url.toExternalForm(), millis));
- } else {
- page.forEach(f -> session.getProvenanceReporter().receive(f, url.toExternalForm(), millis));
- }
- }
- return page.size();
- }
-
- private URL buildRequestURL(String baseUrl, String query, String index, String type, String fields,
- String sort, int pageSize, int fromIndex, ProcessContext context) throws MalformedURLException {
- if (StringUtils.isEmpty(baseUrl)) {
- throw new MalformedURLException("Base URL cannot be null");
- }
- HttpUrl.Builder builder = HttpUrl.parse(baseUrl).newBuilder();
- builder.addPathSegment((StringUtils.isEmpty(index)) ? "_all" : index);
- if (StringUtils.isNotBlank(type)) {
- builder.addPathSegment(type);
- }
- builder.addPathSegment("_search");
- builder.addQueryParameter(QUERY_QUERY_PARAM, query);
- builder.addQueryParameter(SIZE_QUERY_PARAM, String.valueOf(pageSize));
- builder.addQueryParameter(FROM_QUERY_PARAM, String.valueOf(fromIndex));
- if (!StringUtils.isEmpty(fields)) {
- String trimmedFields = Stream.of(fields.split(",")).map(String::trim).collect(Collectors.joining(","));
- builder.addQueryParameter(SOURCE_QUERY_PARAM, trimmedFields);
- }
- if (!StringUtils.isEmpty(sort)) {
- String trimmedFields = Stream.of(sort.split(",")).map(String::trim).collect(Collectors.joining(","));
- builder.addQueryParameter(SORT_QUERY_PARAM, trimmedFields);
- }
-
- // Find the user-added properties and set them as query parameters on the URL
- for (Map.Entry<PropertyDescriptor, String> property : context.getProperties().entrySet()) {
- PropertyDescriptor pd = property.getKey();
- if (pd.isDynamic()) {
- if (property.getValue() != null) {
- builder.addQueryParameter(pd.getName(), context.getProperty(pd).evaluateAttributeExpressions().getValue());
- }
- }
- }
-
- return builder.build().url();
- }
-}
diff --git a/nifi-nar-bundles/nifi-elasticsearch-bundle/nifi-elasticsearch-processors/src/main/java/org/apache/nifi/processors/elasticsearch/RetryableException.java b/nifi-nar-bundles/nifi-elasticsearch-bundle/nifi-elasticsearch-processors/src/main/java/org/apache/nifi/processors/elasticsearch/RetryableException.java
deleted file mode 100644
index 0ecec5d7b3..0000000000
--- a/nifi-nar-bundles/nifi-elasticsearch-bundle/nifi-elasticsearch-processors/src/main/java/org/apache/nifi/processors/elasticsearch/RetryableException.java
+++ /dev/null
@@ -1,43 +0,0 @@
-/*
- * 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;
-
-/**
- * Represents a retryable exception from ElasticSearch.
- */
-@Deprecated
-public class RetryableException extends RuntimeException {
-
- private static final long serialVersionUID = -2755015600102381620L;
-
- public RetryableException() {
- super();
- }
-
- public RetryableException(String message, Throwable cause) {
- super(message, cause);
- }
-
- public RetryableException(String message) {
- super(message);
- }
-
- public RetryableException(Throwable cause) {
- super(cause);
- }
-
-}
diff --git a/nifi-nar-bundles/nifi-elasticsearch-bundle/nifi-elasticsearch-processors/src/main/java/org/apache/nifi/processors/elasticsearch/ScrollElasticsearchHttp.java b/nifi-nar-bundles/nifi-elasticsearch-bundle/nifi-elasticsearch-processors/src/main/java/org/apache/nifi/processors/elasticsearch/ScrollElasticsearchHttp.java
deleted file mode 100644
index 455e14afd4..0000000000
--- a/nifi-nar-bundles/nifi-elasticsearch-bundle/nifi-elasticsearch-processors/src/main/java/org/apache/nifi/processors/elasticsearch/ScrollElasticsearchHttp.java
+++ /dev/null
@@ -1,452 +0,0 @@
-/*
- * 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 com.fasterxml.jackson.databind.JsonNode;
-import okhttp3.HttpUrl;
-import okhttp3.MediaType;
-import okhttp3.OkHttpClient;
-import okhttp3.RequestBody;
-import okhttp3.Response;
-import okhttp3.ResponseBody;
-import org.apache.commons.lang3.StringUtils;
-import org.apache.nifi.annotation.behavior.DynamicProperty;
-import org.apache.nifi.annotation.behavior.EventDriven;
-import org.apache.nifi.annotation.behavior.InputRequirement;
-import org.apache.nifi.annotation.behavior.Stateful;
-import org.apache.nifi.annotation.behavior.SupportsBatching;
-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.DeprecationNotice;
-import org.apache.nifi.annotation.documentation.Tags;
-import org.apache.nifi.annotation.lifecycle.OnScheduled;
-import org.apache.nifi.components.PropertyDescriptor;
-import org.apache.nifi.components.state.Scope;
-import org.apache.nifi.components.state.StateManager;
-import org.apache.nifi.components.state.StateMap;
-import org.apache.nifi.expression.ExpressionLanguageScope;
-import org.apache.nifi.flowfile.FlowFile;
-import org.apache.nifi.logging.ComponentLog;
-import org.apache.nifi.processor.ProcessContext;
-import org.apache.nifi.processor.ProcessSession;
-import org.apache.nifi.processor.Relationship;
-import org.apache.nifi.processor.exception.ProcessException;
-import org.apache.nifi.processor.util.StandardValidators;
-
-import java.io.ByteArrayInputStream;
-import java.io.IOException;
-import java.net.MalformedURLException;
-import java.net.URL;
-import java.nio.charset.Charset;
-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;
-import java.util.concurrent.TimeUnit;
-import java.util.regex.Pattern;
-import java.util.stream.Collectors;
-import java.util.stream.Stream;
-
-@Deprecated
-@DeprecationNotice(classNames = {"org.apache.nifi.processors.elasticsearch.SearchElasticsearch"},
- reason = "This processor is deprecated and may be removed in future releases.")
-@InputRequirement(InputRequirement.Requirement.INPUT_FORBIDDEN)
-@EventDriven
-@SupportsBatching
-@Tags({ "elasticsearch", "query", "scroll", "read", "get", "http" })
-@CapabilityDescription("Scrolls through an Elasticsearch query using the specified connection properties. "
- + "This processor is intended to be run on the primary node, and is designed for scrolling through "
- + "huge result sets, as in the case of a reindex. The state must be cleared before another query "
- + "can be run. Each page of results is returned, wrapped in a JSON object like so: { \"hits\" : [ <doc1>, <doc2>, <docn> ] }. "
- + "Note that the full body of each page of documents will be read into memory before being "
- + "written to a Flow File for transfer.")
-@WritesAttributes({
- @WritesAttribute(attribute = "es.index", description = "The Elasticsearch index containing the document"),
- @WritesAttribute(attribute = "es.type", description = "The Elasticsearch document type") })
-@DynamicProperty(
- name = "A URL query parameter",
- value = "The value to set it to",
- expressionLanguageScope = ExpressionLanguageScope.VARIABLE_REGISTRY,
- description = "Adds the specified property name/value as a query parameter in the Elasticsearch URL used for processing")
-@Stateful(description = "After each successful scroll page, the latest scroll_id is persisted in scrollId as input for the next scroll call. "
- + "Once the entire query is complete, finishedQuery state will be set to true, and the processor will not execute unless this is cleared.", scopes = { Scope.LOCAL })
-public class ScrollElasticsearchHttp extends AbstractElasticsearchHttpProcessor {
-
- private static final String FINISHED_QUERY_STATE = "finishedQuery";
- private static final String SCROLL_ID_STATE = "scrollId";
- private static final String SCROLL_QUERY_PARAM = "scroll";
-
- public static final Relationship REL_SUCCESS = new Relationship.Builder()
- .name("success")
- .description(
- "All FlowFiles that are read from Elasticsearch are routed to this relationship.")
- .build();
-
- public static final Relationship REL_FAILURE = new Relationship.Builder()
- .name("failure")
- .description(
- "All FlowFiles that cannot be read from Elasticsearch are routed to this relationship. Note that only incoming "
- + "flow files will be routed to failure.").build();
-
- public static final PropertyDescriptor QUERY = new PropertyDescriptor.Builder()
- .name("scroll-es-query")
- .displayName("Query")
- .description("The Lucene-style query to run against ElasticSearch (e.g., genre:blues AND -artist:muddy)")
- .required(true)
- .expressionLanguageSupported(ExpressionLanguageScope.FLOWFILE_ATTRIBUTES)
- .addValidator(StandardValidators.NON_EMPTY_VALIDATOR)
- .build();
-
- public static final PropertyDescriptor SCROLL_DURATION = new PropertyDescriptor.Builder()
- .name("scroll-es-scroll")
- .displayName("Scroll Duration")
- .description("The scroll duration is how long each search context is kept in memory.")
- .defaultValue("1m")
- .required(true)
- .expressionLanguageSupported(ExpressionLanguageScope.FLOWFILE_ATTRIBUTES)
- .addValidator(
- StandardValidators.createRegexMatchingValidator(Pattern.compile("[0-9]+(m|h)")))
- .build();
-
- public static final PropertyDescriptor INDEX = new PropertyDescriptor.Builder()
- .name("scroll-es-index")
- .displayName("Index")
- .description("The name of the index to read from. If the property is set "
- + "to _all, the query will match across all indexes.")
- .required(true)
- .expressionLanguageSupported(ExpressionLanguageScope.FLOWFILE_ATTRIBUTES)
- .addValidator(StandardValidators.NON_EMPTY_VALIDATOR)
- .build();
-
- public static final PropertyDescriptor TYPE = new PropertyDescriptor.Builder()
- .name("scroll-es-type")
- .displayName("Type")
- .description("The type of document (if unset, the query will be against all types in the _index). "
- + "This should be unset or '_doc' for Elasticsearch 7.0+.")
- .required(false)
- .expressionLanguageSupported(ExpressionLanguageScope.FLOWFILE_ATTRIBUTES)
- .addValidator(StandardValidators.NON_EMPTY_EL_VALIDATOR)
- .build();
-
- public static final PropertyDescriptor FIELDS = new PropertyDescriptor.Builder()
- .name("scroll-es-fields")
- .displayName("Fields")
- .description(
- "A comma-separated list of fields to retrieve from the document. If the Fields property is left blank, "
- + "then the entire document's source will be retrieved.")
- .required(false)
- .expressionLanguageSupported(ExpressionLanguageScope.FLOWFILE_ATTRIBUTES)
- .addValidator(StandardValidators.NON_EMPTY_VALIDATOR)
- .build();
-
- public static final PropertyDescriptor SORT = new PropertyDescriptor.Builder()
- .name("scroll-es-sort")
- .displayName("Sort")
- .description(
- "A sort parameter (e.g., timestamp:asc). If the Sort property is left blank, "
- + "then the results will be retrieved in document order.")
- .required(false)
- .expressionLanguageSupported(ExpressionLanguageScope.FLOWFILE_ATTRIBUTES)
- .addValidator(StandardValidators.NON_EMPTY_VALIDATOR)
- .build();
-
- public static final PropertyDescriptor PAGE_SIZE = new PropertyDescriptor.Builder()
- .name("scroll-es-size")
- .displayName("Page Size")
- .defaultValue("20")
- .description("Determines how many documents to return per page during scrolling.")
- .required(true)
- .expressionLanguageSupported(ExpressionLanguageScope.FLOWFILE_ATTRIBUTES)
- .addValidator(StandardValidators.POSITIVE_INTEGER_VALIDATOR)
- .build();
-
- private static final Set<Relationship> relationships;
- private static final List<PropertyDescriptor> propertyDescriptors;
-
- 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<>(COMMON_PROPERTY_DESCRIPTORS);
- descriptors.add(QUERY);
- descriptors.add(SCROLL_DURATION);
- descriptors.add(PAGE_SIZE);
- descriptors.add(INDEX);
- descriptors.add(TYPE);
- descriptors.add(FIELDS);
- descriptors.add(SORT);
-
- propertyDescriptors = Collections.unmodifiableList(descriptors);
- }
-
- @Override
- public Set<Relationship> getRelationships() {
- return relationships;
- }
-
- @Override
- public final List<PropertyDescriptor> getSupportedPropertyDescriptors() {
- return propertyDescriptors;
- }
-
- @OnScheduled
- public void setup(ProcessContext context) {
- super.setup(context);
- }
-
- @Override
- public void onTrigger(final ProcessContext context, final ProcessSession session)
- throws ProcessException {
-
- try {
- if (isQueryFinished(session)) {
- getLogger().trace(
- "Query has been marked finished in the state manager. "
- + "To run another query, clear the state.");
- return;
- }
- } catch (IOException e) {
- throw new ProcessException("Could not retrieve state", e);
- }
-
- OkHttpClient okHttpClient = getClient();
-
- FlowFile flowFile = session.create();
-
- final String index = context.getProperty(INDEX).evaluateAttributeExpressions(flowFile)
- .getValue();
- final String query = context.getProperty(QUERY).evaluateAttributeExpressions(flowFile)
- .getValue();
- final String docType = context.getProperty(TYPE).evaluateAttributeExpressions(flowFile)
- .getValue();
- final int pageSize = context.getProperty(PAGE_SIZE).evaluateAttributeExpressions(flowFile)
- .asInteger();
- final String fields = context.getProperty(FIELDS).isSet() ? context.getProperty(FIELDS)
- .evaluateAttributeExpressions(flowFile).getValue() : null;
- final String sort = context.getProperty(SORT).isSet() ? context.getProperty(SORT)
- .evaluateAttributeExpressions(flowFile).getValue() : null;
- final String scroll = context.getProperty(SCROLL_DURATION).isSet() ? context
- .getProperty(SCROLL_DURATION).evaluateAttributeExpressions(flowFile).getValue() : null;
-
- // Authentication
- final String username = context.getProperty(USERNAME).evaluateAttributeExpressions().getValue();
- final String password = context.getProperty(PASSWORD).evaluateAttributeExpressions().getValue();
- final Charset charset = Charset.forName(context.getProperty(CHARSET).evaluateAttributeExpressions(flowFile).getValue());
-
- final ComponentLog logger = getLogger();
-
- try {
- String scrollId = loadScrollId(session);
-
- // read the url property from the context
- final String urlstr = StringUtils.trimToEmpty(context.getProperty(ES_URL).evaluateAttributeExpressions()
- .getValue());
- if (scrollId != null) {
- final URL scrollurl = buildRequestURL(urlstr, query, index, docType, fields, sort,
- scrollId, pageSize, scroll, context);
- final long startNanos = System.nanoTime();
-
- final String scrollBody = String.format("{ \"scroll\": \"%s\", \"scroll_id\": \"%s\" }", scroll,
- scrollId);
-
- final RequestBody body = RequestBody.create(MediaType.parse("application/json"), scrollBody);
-
- final Response getResponse = sendRequestToElasticsearch(okHttpClient, scrollurl,
- username, password, "POST", body);
- this.getPage(getResponse, scrollurl, context, session, flowFile, logger, startNanos, charset);
- getResponse.close();
- } else {
- logger.debug("Querying {}/{} from Elasticsearch: {}", new Object[] { index,
- docType, query });
-
- // read the url property from the context
- final URL queryUrl = buildRequestURL(urlstr, query, index, docType, fields, sort,
- scrollId, pageSize, scroll, context);
- final long startNanos = System.nanoTime();
-
- final Response getResponse = sendRequestToElasticsearch(okHttpClient, queryUrl,
- username, password, "GET", null);
- this.getPage(getResponse, queryUrl, context, session, flowFile, logger, startNanos, charset);
- getResponse.close();
- }
-
- } catch (IOException ioe) {
- logger.error(
- "Failed to read from Elasticsearch due to {}, this may indicate an error in configuration "
- + "(hosts, username/password, etc.).",
- new Object[] { ioe.getLocalizedMessage() }, ioe);
- session.remove(flowFile);
- context.yield();
-
- } catch (Exception e) {
- logger.error("Failed to read {} from Elasticsearch due to {}", new Object[] { flowFile,
- e.getLocalizedMessage() }, e);
- session.transfer(flowFile, REL_FAILURE);
- context.yield();
- }
- }
-
- private void getPage(final Response getResponse, final URL url, final ProcessContext context,
- final ProcessSession session, FlowFile flowFile, final ComponentLog logger, final long startNanos, Charset charset)
- throws IOException {
- final int statusCode = getResponse.code();
-
- if (isSuccess(statusCode)) {
- ResponseBody body = getResponse.body();
- final byte[] bodyBytes = body.bytes();
- JsonNode responseJson = parseJsonResponse(new ByteArrayInputStream(bodyBytes));
- String scrollId = responseJson.get("_scroll_id").asText();
-
- StringBuilder builder = new StringBuilder();
-
- builder.append("{ \"hits\" : [");
-
- JsonNode hits = responseJson.get("hits").get("hits");
- if (hits.size() == 0) {
- finishQuery(context.getStateManager());
- session.remove(flowFile);
- return;
- }
-
- for(int i = 0; i < hits.size(); i++) {
- JsonNode hit = hits.get(i);
- String retrievedIndex = hit.get("_index").asText();
- String retrievedType = hit.get("_type").asText();
-
- JsonNode source = hit.get("_source");
- flowFile = session.putAttribute(flowFile, "es.index", retrievedIndex);
- flowFile = session.putAttribute(flowFile, "es.type", retrievedType);
- flowFile = session.putAttribute(flowFile, "mime.type", "application/json");
-
- builder.append(source.toString());
- if (i < hits.size() - 1) {
- builder.append(", ");
- }
- }
- builder.append("] }");
- logger.debug("Elasticsearch retrieved " + responseJson.size() + " documents, routing to success");
-
- flowFile = session.write(flowFile, out -> {
- out.write(builder.toString().getBytes(charset));
- });
- session.transfer(flowFile, REL_SUCCESS);
-
- session.setState(Collections.singletonMap(SCROLL_ID_STATE, scrollId), Scope.LOCAL);
-
- // emit provenance event
- final long millis = TimeUnit.NANOSECONDS.toMillis(System.nanoTime() - startNanos);
- session.getProvenanceReporter().receive(flowFile, url.toExternalForm(), millis);
- } else {
- // 5xx -> RETRY, but a server error might last a while, so yield
- if (statusCode / 100 == 5) {
-
- logger.warn("Elasticsearch returned code {} with message {}, removing the flow file. This is likely a server problem, yielding...",
- new Object[]{statusCode, getResponse.message()});
- session.remove(flowFile);
- context.yield();
- } else {
- logger.warn("Elasticsearch returned code {} with message {}", new Object[]{statusCode, getResponse.message()});
- session.remove(flowFile);
- }
- }
- }
-
- private boolean isQueryFinished(final ProcessSession session) throws IOException {
- final StateMap stateMap = session.getState(Scope.LOCAL);
-
- if (stateMap.getVersion() < 0) {
- getLogger().debug("No previous state found");
- return false;
- }
-
- final String isQueryFinished = stateMap.get(FINISHED_QUERY_STATE);
- getLogger().debug("Loaded state with finishedQuery = {}", new Object[] { isQueryFinished });
-
- return "true".equals(isQueryFinished);
- }
-
- private String loadScrollId(final ProcessSession session) throws IOException {
- final StateMap stateMap = session.getState(Scope.LOCAL);
-
- if (stateMap.getVersion() < 0) {
- getLogger().debug("No previous state found");
- return null;
- }
-
- final String scrollId = stateMap.get(SCROLL_ID_STATE);
- getLogger().debug("Loaded state with scrollId {}", new Object[] { scrollId });
-
- return scrollId;
- }
-
- private void finishQuery(StateManager stateManager) throws IOException {
-
- Map<String, String> state = new HashMap<>(2);
- state.put(FINISHED_QUERY_STATE, "true");
-
- getLogger().debug("Saving state with finishedQuery = true");
- stateManager.setState(state, Scope.LOCAL);
- }
-
- private URL buildRequestURL(String baseUrl, String query, String index, String type, String fields,
- String sort, String scrollId, int pageSize, String scroll, ProcessContext context) throws MalformedURLException {
- if (StringUtils.isEmpty(baseUrl)) {
- throw new MalformedURLException("Base URL cannot be null");
- }
- HttpUrl.Builder builder = HttpUrl.parse(baseUrl).newBuilder();
- if (!StringUtils.isEmpty(scrollId)) {
- builder.addPathSegment("_search");
- builder.addPathSegment("scroll");
- } else {
- builder.addPathSegment((StringUtils.isEmpty(index)) ? "_all" : index);
- if (StringUtils.isNotBlank(type)) {
- builder.addPathSegment(type);
- }
- builder.addPathSegment("_search");
- builder.addQueryParameter(QUERY_QUERY_PARAM, query);
- builder.addQueryParameter(SIZE_QUERY_PARAM, String.valueOf(pageSize));
- if (!StringUtils.isEmpty(fields)) {
- String trimmedFields = Stream.of(fields.split(",")).map(String::trim).collect(Collectors.joining(","));
- builder.addQueryParameter(SOURCE_QUERY_PARAM, trimmedFields);
- }
- if (!StringUtils.isEmpty(sort)) {
- String trimmedFields = Stream.of(sort.split(",")).map(String::trim).collect(Collectors.joining(","));
- builder.addQueryParameter(SORT_QUERY_PARAM, trimmedFields);
- }
- builder.addQueryParameter(SCROLL_QUERY_PARAM, scroll);
- }
-
- // Find the user-added properties and set them as query parameters on the URL
- for (Map.Entry<PropertyDescriptor, String> property : context.getProperties().entrySet()) {
- PropertyDescriptor pd = property.getKey();
- if (pd.isDynamic()) {
- if (property.getValue() != null) {
- builder.addQueryParameter(pd.getName(), context.getProperty(pd).evaluateAttributeExpressions().getValue());
- }
- }
- }
-
-
- return builder.build().url();
- }
-}
diff --git a/nifi-nar-bundles/nifi-elasticsearch-bundle/nifi-elasticsearch-processors/src/main/java/org/apache/nifi/processors/elasticsearch/UnretryableException.java b/nifi-nar-bundles/nifi-elasticsearch-bundle/nifi-elasticsearch-processors/src/main/java/org/apache/nifi/processors/elasticsearch/UnretryableException.java
deleted file mode 100644
index d6071f705b..0000000000
--- a/nifi-nar-bundles/nifi-elasticsearch-bundle/nifi-elasticsearch-processors/src/main/java/org/apache/nifi/processors/elasticsearch/UnretryableException.java
+++ /dev/null
@@ -1,43 +0,0 @@
-/*
- * 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;
-
-/**
- * Represents an unrecoverable error from ElasticSearch.
- * @author jgresock
- */
-@Deprecated
-public class UnretryableException extends RuntimeException {
- private static final long serialVersionUID = -4528006567211380914L;
-
- public UnretryableException() {
- super();
- }
-
- public UnretryableException(String message, Throwable cause) {
- super(message, cause);
- }
-
- public UnretryableException(String message) {
- super(message);
- }
-
- public UnretryableException(Throwable cause) {
- super(cause);
- }
-
-}
diff --git a/nifi-nar-bundles/nifi-elasticsearch-bundle/nifi-elasticsearch-processors/src/main/resources/META-INF/services/org.apache.nifi.processor.Processor b/nifi-nar-bundles/nifi-elasticsearch-bundle/nifi-elasticsearch-processors/src/main/resources/META-INF/services/org.apache.nifi.processor.Processor
deleted file mode 100644
index d6e98f153f..0000000000
--- a/nifi-nar-bundles/nifi-elasticsearch-bundle/nifi-elasticsearch-processors/src/main/resources/META-INF/services/org.apache.nifi.processor.Processor
+++ /dev/null
@@ -1,19 +0,0 @@
-# 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.
-org.apache.nifi.processors.elasticsearch.FetchElasticsearchHttp
-org.apache.nifi.processors.elasticsearch.PutElasticsearchHttp
-org.apache.nifi.processors.elasticsearch.PutElasticsearchHttpRecord
-org.apache.nifi.processors.elasticsearch.QueryElasticsearchHttp
-org.apache.nifi.processors.elasticsearch.ScrollElasticsearchHttp
diff --git a/nifi-nar-bundles/nifi-elasticsearch-bundle/nifi-elasticsearch-processors/src/test/java/org/apache/nifi/processors/elasticsearch/TestFetchElasticsearchHttp.java b/nifi-nar-bundles/nifi-elasticsearch-bundle/nifi-elasticsearch-processors/src/test/java/org/apache/nifi/processors/elasticsearch/TestFetchElasticsearchHttp.java
deleted file mode 100644
index 54afba3332..0000000000
--- a/nifi-nar-bundles/nifi-elasticsearch-bundle/nifi-elasticsearch-processors/src/test/java/org/apache/nifi/processors/elasticsearch/TestFetchElasticsearchHttp.java
+++ /dev/null
@@ -1,417 +0,0 @@
-/*
- * 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 okhttp3.Call;
-import okhttp3.MediaType;
-import okhttp3.OkHttpClient;
-import okhttp3.Protocol;
-import okhttp3.Request;
-import okhttp3.RequestBody;
-import okhttp3.Response;
-import okhttp3.ResponseBody;
-import org.apache.nifi.processor.ProcessContext;
-import org.apache.nifi.processor.exception.ProcessException;
-import org.apache.nifi.ssl.SSLContextService;
-import org.apache.nifi.util.MockFlowFile;
-import org.apache.nifi.util.TestRunner;
-import org.apache.nifi.util.TestRunners;
-import org.junit.jupiter.api.AfterEach;
-import org.junit.jupiter.api.BeforeEach;
-import org.junit.jupiter.api.Test;
-import org.mockito.stubbing.Answer;
-
-import java.io.IOException;
-import java.io.InputStream;
-import java.net.URL;
-import java.util.HashMap;
-
-import static org.junit.jupiter.api.Assertions.assertEquals;
-import static org.junit.jupiter.api.Assertions.assertNotNull;
-import static org.junit.jupiter.api.Assertions.assertTrue;
-import static org.mockito.ArgumentMatchers.any;
-import static org.mockito.Mockito.mock;
-import static org.mockito.Mockito.when;
-
-public class TestFetchElasticsearchHttp {
-
- private InputStream docExample;
- private TestRunner runner;
-
- @BeforeEach
- public void setUp() {
- ClassLoader classloader = Thread.currentThread().getContextClassLoader();
- docExample = classloader.getResourceAsStream("DocumentExample.json");
- }
-
- @AfterEach
- public void teardown() {
- runner = null;
- }
-
- @Test
- public void testFetchElasticsearchOnTriggerEL() {
- runner = TestRunners.newTestRunner(new FetchElasticsearchHttpTestProcessor(true)); // all docs are found
- runner.setProperty(AbstractElasticsearchHttpProcessor.ES_URL, "${es.url}");
-
- runner.setProperty(FetchElasticsearchHttp.INDEX, "doc");
- runner.assertNotValid();
- runner.setProperty(FetchElasticsearchHttp.DOC_ID, "${doc_id}");
- runner.removeProperty(FetchElasticsearchHttp.TYPE);
- runner.assertValid();
- runner.setProperty(FetchElasticsearchHttp.TYPE, "");
- runner.assertNotValid();
- runner.setProperty(FetchElasticsearchHttp.TYPE, "status");
- runner.assertValid();
- runner.setProperty(FetchElasticsearchHttp.TYPE, "${type}");
- runner.assertValid();
- runner.setProperty(FetchElasticsearchHttp.TYPE, "_doc");
- runner.assertValid(); // Valid because type can be _doc for 7.0+
- runner.setProperty(AbstractElasticsearchHttpProcessor.CONNECT_TIMEOUT, "${connect.timeout}");
- runner.assertValid();
-
- runner.setVariable("es.url", "http://127.0.0.1:9200");
- runner.setVariable("connect.timeout", "5s");
-
- runner.enqueue(docExample, new HashMap<String, String>() {{
- put("doc_id", "28039652140");
- }});
- runner.run(1, true, true);
-
- runner.assertAllFlowFilesTransferred(FetchElasticsearchHttp.REL_SUCCESS, 1);
- final MockFlowFile out = runner.getFlowFilesForRelationship(FetchElasticsearchHttp.REL_SUCCESS).get(0);
- assertNotNull(out);
- out.assertAttributeEquals("doc_id", "28039652140");
- }
-
- @Test
- public void testFetchElasticsearchOnTrigger() {
- runner = TestRunners.newTestRunner(new FetchElasticsearchHttpTestProcessor(true)); // all docs are found
- runner.setProperty(AbstractElasticsearchHttpProcessor.ES_URL, "http://127.0.0.1:9200");
-
- runner.setProperty(FetchElasticsearchHttp.INDEX, "doc");
- runner.assertNotValid();
- runner.setProperty(FetchElasticsearchHttp.TYPE, "status");
- runner.assertNotValid();
- runner.setProperty(FetchElasticsearchHttp.DOC_ID, "${doc_id}");
- runner.assertValid();
-
- runner.enqueue(docExample, new HashMap<String, String>() {{
- put("doc_id", "28039652140");
- }});
- runner.run(1, true, true);
-
- runner.assertAllFlowFilesTransferred(FetchElasticsearchHttp.REL_SUCCESS, 1);
- final MockFlowFile out = runner.getFlowFilesForRelationship(FetchElasticsearchHttp.REL_SUCCESS).get(0);
- assertNotNull(out);
- out.assertAttributeEquals("doc_id", "28039652140");
- }
-
- @Test
- public void testFetchElasticsearchOnTriggerNoType() {
- final String ES_URL = "http://127.0.0.1:9200";
- final String DOC_ID = "28039652140";
- FetchElasticsearchHttpTestProcessor processor = new FetchElasticsearchHttpTestProcessor(true);
- runner = TestRunners.newTestRunner(processor); // all docs are found
- runner.setProperty(AbstractElasticsearchHttpProcessor.ES_URL, ES_URL);
-
- runner.setProperty(FetchElasticsearchHttp.INDEX, "doc");
- runner.removeProperty(FetchElasticsearchHttp.TYPE);
- runner.assertNotValid();
- runner.setProperty(FetchElasticsearchHttp.DOC_ID, "${doc_id}");
- runner.assertValid();
-
- runner.enqueue(docExample, new HashMap<String, String>() {{
- put("doc_id", DOC_ID);
- }});
- runner.run(1, true, true);
-
- runner.assertAllFlowFilesTransferred(FetchElasticsearchHttp.REL_SUCCESS, 1);
- final MockFlowFile out = runner.getFlowFilesForRelationship(FetchElasticsearchHttp.REL_SUCCESS).get(0);
- assertNotNull(out);
- out.assertAttributeEquals("doc_id", DOC_ID);
- assertEquals("http://127.0.0.1:9200" + "/doc/_all/" + DOC_ID,
- processor.getURL().toString(), "URL doesn't match expected value when type is not supplied");
- }
-
- @Test
- public void testFetchElasticsearchOnTriggerWithFields() {
- runner = TestRunners.newTestRunner(new FetchElasticsearchHttpTestProcessor(true)); // all docs are found
- runner.setProperty(AbstractElasticsearchHttpProcessor.ES_URL, "http://127.0.0.1:9200");
-
- runner.setProperty(FetchElasticsearchHttp.INDEX, "doc");
- runner.assertNotValid();
- runner.setProperty(FetchElasticsearchHttp.TYPE, "status");
- runner.assertNotValid();
- runner.setProperty(FetchElasticsearchHttp.DOC_ID, "${doc_id}");
- runner.assertValid();
- runner.setProperty(FetchElasticsearchHttp.FIELDS, "id,, userinfo.location");
- runner.assertValid();
-
- runner.enqueue(docExample, new HashMap<String, String>() {{
- put("doc_id", "28039652140");
- }});
- runner.run(1, true, true);
-
- runner.assertAllFlowFilesTransferred(FetchElasticsearchHttp.REL_SUCCESS, 1);
- final MockFlowFile out = runner.getFlowFilesForRelationship(FetchElasticsearchHttp.REL_SUCCESS).get(0);
- assertNotNull(out);
- out.assertAttributeEquals("doc_id", "28039652140");
- }
-
- @Test
- public void testFetchElasticsearchOnTriggerWithDocNotFound() {
- runner = TestRunners.newTestRunner(new FetchElasticsearchHttpTestProcessor(false)); // simulate doc not found
- runner.setProperty(AbstractElasticsearchHttpProcessor.ES_URL, "http://127.0.0.1:9200");
- runner.setProperty(FetchElasticsearchHttp.INDEX, "doc");
- runner.removeProperty(FetchElasticsearchHttp.TYPE);
- runner.setProperty(FetchElasticsearchHttp.DOC_ID, "${doc_id}");
-
- runner.setIncomingConnection(true);
- runner.enqueue(docExample, new HashMap<String, String>() {{
- put("doc_id", "28039652140");
- }});
- runner.run(1, true, true);
-
- // This test generates a "document not found"
- runner.assertAllFlowFilesTransferred(FetchElasticsearchHttp.REL_NOT_FOUND, 1);
- final MockFlowFile out = runner.getFlowFilesForRelationship(FetchElasticsearchHttp.REL_NOT_FOUND).get(0);
- assertNotNull(out);
- out.assertAttributeEquals("doc_id", "28039652140");
- }
-
- @Test
- public void testFetchElasticsearchOnTriggerWithServerErrorRetry() {
- FetchElasticsearchHttpTestProcessor processor = new FetchElasticsearchHttpTestProcessor(false);
- processor.setStatus(500, "Server error");
- runner = TestRunners.newTestRunner(processor); // simulate doc not found
- runner.setProperty(AbstractElasticsearchHttpProcessor.ES_URL, "http://127.0.0.1:9200");
- runner.setProperty(FetchElasticsearchHttp.INDEX, "doc");
- runner.setProperty(FetchElasticsearchHttp.TYPE, "status");
- runner.setProperty(FetchElasticsearchHttp.DOC_ID, "${doc_id}");
-
- runner.enqueue(docExample, new HashMap<String, String>() {{
- put("doc_id", "28039652140");
- }});
- runner.run(1, true, true);
-
- // This test generates a HTTP 500 "Server error"
- runner.assertAllFlowFilesTransferred(FetchElasticsearchHttp.REL_RETRY, 1);
- final MockFlowFile out = runner.getFlowFilesForRelationship(FetchElasticsearchHttp.REL_RETRY).get(0);
- assertNotNull(out);
- out.assertAttributeEquals("doc_id", "28039652140");
- }
-
- @Test
- public void testFetchElasticsearchOnTriggerWithServerFail() {
- FetchElasticsearchHttpTestProcessor processor = new FetchElasticsearchHttpTestProcessor(false);
- processor.setStatus(100, "Should fail");
- runner = TestRunners.newTestRunner(processor); // simulate doc not found
- runner.setProperty(AbstractElasticsearchHttpProcessor.ES_URL, "http://127.0.0.1:9200");
- runner.setProperty(FetchElasticsearchHttp.INDEX, "doc");
- runner.setProperty(FetchElasticsearchHttp.TYPE, "status");
- runner.setProperty(FetchElasticsearchHttp.DOC_ID, "${doc_id}");
-
- runner.enqueue(docExample, new HashMap<String, String>() {{
- put("doc_id", "28039652140");
- }});
- runner.run(1, true, true);
-
- // This test generates a HTTP 100
- runner.assertAllFlowFilesTransferred(FetchElasticsearchHttp.REL_FAILURE, 1);
- final MockFlowFile out = runner.getFlowFilesForRelationship(FetchElasticsearchHttp.REL_FAILURE).get(0);
- assertNotNull(out);
- out.assertAttributeEquals("doc_id", "28039652140");
- }
-
- @Test
- public void testFetchElasticsearchOnTriggerWithServerFailNoIncomingFlowFile() {
- FetchElasticsearchHttpTestProcessor processor = new FetchElasticsearchHttpTestProcessor(false);
- processor.setStatus(100, "Should fail");
- runner = TestRunners.newTestRunner(processor); // simulate doc not found
- runner.setProperty(AbstractElasticsearchHttpProcessor.ES_URL, "http://127.0.0.1:9200");
- runner.setProperty(FetchElasticsearchHttp.INDEX, "doc");
- runner.setProperty(FetchElasticsearchHttp.TYPE, "status");
- runner.setProperty(FetchElasticsearchHttp.DOC_ID, "${doc_id}");
-
- runner.setIncomingConnection(false);
- runner.run(1, true, true);
-
- // This test generates a HTTP 100 with no incoming flow file, so nothing should be transferred
- processor.getRelationships().forEach(relationship -> runner.assertTransferCount(relationship, 0));
- runner.assertTransferCount(FetchElasticsearchHttp.REL_FAILURE, 0);
- }
-
- @Test
- public void testFetchElasticsearchWithBadHosts() {
- runner = TestRunners.newTestRunner(new FetchElasticsearchHttpTestProcessor(false)); // simulate doc not found
- runner.setProperty(AbstractElasticsearchHttpProcessor.ES_URL, "127.0.0.1:9200");
- runner.setProperty(FetchElasticsearchHttp.INDEX, "doc");
- runner.setProperty(FetchElasticsearchHttp.TYPE, "status");
- runner.setProperty(FetchElasticsearchHttp.DOC_ID, "${doc_id}");
-
- runner.assertNotValid();
- }
-
- @Test
- public void testSetupSecureClient() throws Exception {
- FetchElasticsearchHttpTestProcessor processor = new FetchElasticsearchHttpTestProcessor(true);
- runner = TestRunners.newTestRunner(processor);
- SSLContextService sslService = mock(SSLContextService.class);
- when(sslService.getIdentifier()).thenReturn("ssl-context");
- runner.addControllerService("ssl-context", sslService);
- runner.enableControllerService(sslService);
- runner.setProperty(FetchElasticsearchHttp.PROP_SSL_CONTEXT_SERVICE, "ssl-context");
- runner.setProperty(AbstractElasticsearchHttpProcessor.ES_URL, "http://127.0.0.1:9200");
- runner.setProperty(FetchElasticsearchHttp.INDEX, "doc");
- runner.removeProperty(FetchElasticsearchHttp.TYPE);
- runner.setProperty(FetchElasticsearchHttp.DOC_ID, "${doc_id}");
-
- // Allow time for the controller service to fully initialize
- Thread.sleep(500);
-
- runner.enqueue(docExample, new HashMap<String, String>() {{
- put("doc_id", "28039652140");
- }});
- runner.run(1, true, true);
-
- }
-
- @Test
- public void testFetchElasticsearchOnTriggerQueryParameter() {
- FetchElasticsearchHttpTestProcessor p = new FetchElasticsearchHttpTestProcessor(true); // all docs are found
- p.setExpectedUrl("http://127.0.0.1:9200/doc/status/28039652140?_source=id&myparam=myvalue");
- runner = TestRunners.newTestRunner(p);
- runner.setProperty(AbstractElasticsearchHttpProcessor.ES_URL, "http://127.0.0.1:9200");
-
- runner.setProperty(FetchElasticsearchHttp.INDEX, "doc");
- runner.setProperty(FetchElasticsearchHttp.TYPE, "status");
- runner.setProperty(FetchElasticsearchHttp.DOC_ID, "${doc_id}");
- runner.setProperty(FetchElasticsearchHttp.FIELDS, "id");
-
- // Set dynamic property, to be added to the URL as a query parameter
- runner.setProperty("myparam", "myvalue");
-
- runner.enqueue(docExample, new HashMap<String, String>() {{
- put("doc_id", "28039652140");
- }});
- runner.run(1, true, true);
-
- runner.assertAllFlowFilesTransferred(FetchElasticsearchHttp.REL_SUCCESS, 1);
- final MockFlowFile out = runner.getFlowFilesForRelationship(FetchElasticsearchHttp.REL_SUCCESS).get(0);
- assertNotNull(out);
- out.assertAttributeEquals("doc_id", "28039652140");
- }
-
- @Test
- public void testFetchElasticsearchOnTriggerQueryParameterNoType() {
- FetchElasticsearchHttpTestProcessor p = new FetchElasticsearchHttpTestProcessor(true); // all docs are found
- p.setExpectedUrl("http://127.0.0.1:9200/doc/_all/28039652140?_source=id&myparam=myvalue");
- runner = TestRunners.newTestRunner(p);
- runner.setProperty(AbstractElasticsearchHttpProcessor.ES_URL, "http://127.0.0.1:9200");
-
- runner.setProperty(FetchElasticsearchHttp.INDEX, "doc");
- runner.removeProperty(FetchElasticsearchHttp.TYPE);
- runner.setProperty(FetchElasticsearchHttp.DOC_ID, "${doc_id}");
- runner.setProperty(FetchElasticsearchHttp.FIELDS, "id");
-
- // Set dynamic property, to be added to the URL as a query parameter
- runner.setProperty("myparam", "myvalue");
-
- runner.enqueue(docExample, new HashMap<String, String>() {{
- put("doc_id", "28039652140");
- }});
- runner.run(1, true, true);
-
- runner.assertAllFlowFilesTransferred(FetchElasticsearchHttp.REL_SUCCESS, 1);
- final MockFlowFile out = runner.getFlowFilesForRelationship(FetchElasticsearchHttp.REL_SUCCESS).get(0);
- assertNotNull(out);
- out.assertAttributeEquals("doc_id", "28039652140");
- }
-
- /**
- * A Test class that extends the processor in order to inject/mock behavior
- */
- private static class FetchElasticsearchHttpTestProcessor extends FetchElasticsearchHttp {
- boolean documentExists = true;
- Exception exceptionToThrow = null;
- OkHttpClient client;
- int statusCode = 200;
- String statusMessage = "OK";
- URL url = null;
- String expectedUrl = null;
-
- FetchElasticsearchHttpTestProcessor(boolean documentExists) {
- this.documentExists = documentExists;
- }
-
- public void setExceptionToThrow(Exception exceptionToThrow) {
- this.exceptionToThrow = exceptionToThrow;
- }
-
- void setStatus(int code, String message) {
- statusCode = code;
- statusMessage = message;
- }
-
- void setExpectedUrl(String url) {
- expectedUrl = url;
- }
-
- @Override
- protected void createElasticsearchClient(ProcessContext context) throws ProcessException {
- client = mock(OkHttpClient.class);
-
- when(client.newCall(any(Request.class))).thenAnswer((Answer<Call>) invocationOnMock -> {
- Request realRequest = (Request) invocationOnMock.getArguments()[0];
- assertTrue((expectedUrl == null) || (expectedUrl.equals(realRequest.url().toString())));
- StringBuilder sb = new StringBuilder("{\"_index\":\"randomuser.me\",\"_type\":\"user\",\"_id\":\"0\",\"_version\":2,");
- if (documentExists) {
- sb.append("\"found\":true,\"_source\":{\"gender\":\"female\",\"name\":{\"title\":\"Ms\",\"first\":\"Joan\",\"last\":\"Smith\"}}");
- } else {
- sb.append("\"found\": false");
- }
- sb.append("}");
- Response mockResponse = new Response.Builder()
- .request(realRequest)
- .protocol(Protocol.HTTP_1_1)
- .code(statusCode)
- .message(statusMessage)
- .body(ResponseBody.create(MediaType.parse("application/json"), sb.toString()))
- .build();
- final Call call = mock(Call.class);
- when(call.execute()).thenReturn(mockResponse);
- return call;
- });
- }
-
- @Override
- protected Response sendRequestToElasticsearch(OkHttpClient client, URL url, String username, String password, String verb, RequestBody body) throws IOException {
- this.url = url;
- return super.sendRequestToElasticsearch(client, url, username, password, verb, body);
- }
-
- public URL getURL() {
- return url;
- }
-
- @Override
- protected OkHttpClient getClient() {
- return client;
- }
- }
-}
diff --git a/nifi-nar-bundles/nifi-elasticsearch-bundle/nifi-elasticsearch-processors/src/test/java/org/apache/nifi/processors/elasticsearch/TestPutElasticsearchHttp.java b/nifi-nar-bundles/nifi-elasticsearch-bundle/nifi-elasticsearch-processors/src/test/java/org/apache/nifi/processors/elasticsearch/TestPutElasticsearchHttp.java
deleted file mode 100644
index 3d2e4292aa..0000000000
--- a/nifi-nar-bundles/nifi-elasticsearch-bundle/nifi-elasticsearch-processors/src/test/java/org/apache/nifi/processors/elasticsearch/TestPutElasticsearchHttp.java
+++ /dev/null
@@ -1,511 +0,0 @@
-/*
- * 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 okhttp3.Call;
-import okhttp3.MediaType;
-import okhttp3.OkHttpClient;
-import okhttp3.Protocol;
-import okhttp3.Request;
-import okhttp3.Response;
-import okhttp3.ResponseBody;
-import org.apache.commons.io.IOUtils;
-import org.apache.nifi.processor.ProcessContext;
-import org.apache.nifi.processor.exception.ProcessException;
-import org.apache.nifi.util.MockFlowFile;
-import org.apache.nifi.util.TestRunner;
-import org.apache.nifi.util.TestRunners;
-import org.junit.jupiter.api.AfterEach;
-import org.junit.jupiter.api.BeforeEach;
-import org.junit.jupiter.api.Test;
-import org.mockito.stubbing.Answer;
-
-import java.io.IOException;
-import java.net.ConnectException;
-import java.nio.charset.StandardCharsets;
-import java.util.HashMap;
-
-import static org.junit.jupiter.api.Assertions.assertNotNull;
-import static org.junit.jupiter.api.Assertions.assertThrows;
-import static org.junit.jupiter.api.Assertions.assertTrue;
-import static org.mockito.ArgumentMatchers.any;
-import static org.mockito.Mockito.mock;
-import static org.mockito.Mockito.when;
-
-public class TestPutElasticsearchHttp {
-
- private static byte[] docExample;
- private TestRunner runner;
-
- @BeforeEach
- public void once() throws IOException {
- ClassLoader classloader = Thread.currentThread().getContextClassLoader();
- docExample = IOUtils.toString(classloader.getResourceAsStream("DocumentExample.json"), StandardCharsets.UTF_8).getBytes();
- }
-
- @AfterEach
- public void teardown() {
- runner = null;
- }
-
- @Test
- public void testPutElasticSearchOnTriggerIndex() {
- runner = TestRunners.newTestRunner(new PutElasticsearchTestProcessor(false)); // no failures
- runner.setProperty(AbstractElasticsearchHttpProcessor.ES_URL, "http://127.0.0.1:9200");
-
- runner.setProperty(PutElasticsearchHttp.INDEX, "doc");
- runner.setProperty(PutElasticsearchHttp.TYPE, "status");
- runner.setProperty(PutElasticsearchHttp.BATCH_SIZE, "1");
- runner.setProperty(PutElasticsearchHttp.ID_ATTRIBUTE, "doc_id");
-
- runner.enqueue(docExample, new HashMap<String, String>() {{
- put("doc_id", "28039652140");
- }});
- runner.run(1, true, true);
-
- runner.assertAllFlowFilesTransferred(PutElasticsearchHttp.REL_SUCCESS, 1);
- final MockFlowFile out = runner.getFlowFilesForRelationship(PutElasticsearchHttp.REL_SUCCESS).get(0);
- assertNotNull(out);
- out.assertAttributeEquals("doc_id", "28039652140");
- }
-
- @Test
- public void testPutElasticSearchOnTriggerCreate() {
- runner = TestRunners.newTestRunner(new PutElasticsearchTestProcessor(false)); // no failures
- runner.setProperty(AbstractElasticsearchHttpProcessor.ES_URL, "http://127.0.0.1:9200");
-
- runner.setProperty(PutElasticsearchHttp.INDEX, "doc");
- runner.setProperty(PutElasticsearchHttp.TYPE, "status");
- runner.setProperty(PutElasticsearchHttp.BATCH_SIZE, "1");
- runner.setProperty(PutElasticsearchHttp.ID_ATTRIBUTE, "doc_id");
- runner.setProperty(PutElasticsearchHttp.INDEX_OP, "create");
-
- runner.enqueue(docExample, new HashMap<String, String>() {{
- put("doc_id", "28039652140");
- }});
- runner.run(1, true, true);
-
- runner.assertAllFlowFilesTransferred(PutElasticsearchHttp.REL_SUCCESS, 1);
- final MockFlowFile out = runner.getFlowFilesForRelationship(PutElasticsearchHttp.REL_SUCCESS).get(0);
- assertNotNull(out);
- out.assertAttributeEquals("doc_id", "28039652140");
- }
-
- @Test
- public void testPutElasticSearchOnTriggerIndex_withoutType() {
- runner = TestRunners.newTestRunner(new PutElasticsearchTestProcessor(false)); // no failures
- runner.setProperty(AbstractElasticsearchHttpProcessor.ES_URL, "http://127.0.0.1:9200");
-
- runner.setProperty(PutElasticsearchHttp.INDEX, "doc");
- runner.removeProperty(PutElasticsearchHttp.TYPE);
- runner.setProperty(PutElasticsearchHttp.BATCH_SIZE, "1");
- runner.setProperty(PutElasticsearchHttp.ID_ATTRIBUTE, "doc_id");
-
- runner.enqueue(docExample, new HashMap<String, String>() {{
- put("doc_id", "28039652140");
- }});
- runner.run(1, true, true);
-
- runner.assertAllFlowFilesTransferred(PutElasticsearchHttp.REL_SUCCESS, 1);
- final MockFlowFile out = runner.getFlowFilesForRelationship(PutElasticsearchHttp.REL_SUCCESS).get(0);
- assertNotNull(out);
- out.assertAttributeEquals("doc_id", "28039652140");
- }
-
- @Test
- public void testPutElasticSearchOnTriggerUpdate() {
- runner = TestRunners.newTestRunner(new PutElasticsearchTestProcessor(false)); // no failures
- runner.setProperty(AbstractElasticsearchHttpProcessor.ES_URL, "http://127.0.0.1:9200");
-
- runner.setProperty(PutElasticsearchHttp.INDEX, "doc");
- runner.setProperty(PutElasticsearchHttp.TYPE, "status");
- runner.setProperty(PutElasticsearchHttp.BATCH_SIZE, "1");
- runner.setProperty(PutElasticsearchHttp.ID_ATTRIBUTE, "doc_id");
- runner.setProperty(PutElasticsearchHttp.INDEX_OP, "Update");
- runner.enqueue(docExample, new HashMap<String, String>() {{
- put("doc_id", "28039652140");
- }});
- runner.run(1, true, true);
-
- runner.assertAllFlowFilesTransferred(PutElasticsearchHttp.REL_SUCCESS, 1);
- final MockFlowFile out = runner.getFlowFilesForRelationship(PutElasticsearchHttp.REL_SUCCESS).get(0);
- assertNotNull(out);
- out.assertAttributeEquals("doc_id", "28039652140");
- }
-
- @Test
- public void testPutElasticSearchOnTriggerDelete() {
- runner = TestRunners.newTestRunner(new PutElasticsearchTestProcessor(false)); // no failures
- runner.setProperty(AbstractElasticsearchHttpProcessor.ES_URL, "http://127.0.0.1:9200");
-
- runner.setProperty(PutElasticsearchHttp.INDEX, "doc");
- runner.setProperty(PutElasticsearchHttp.TYPE, "status");
- runner.setProperty(PutElasticsearchHttp.BATCH_SIZE, "1");
- runner.setProperty(PutElasticsearchHttp.ID_ATTRIBUTE, "doc_id");
- runner.setProperty(PutElasticsearchHttp.INDEX_OP, "DELETE");
- runner.enqueue(docExample, new HashMap<String, String>() {{
- put("doc_id", "28039652140");
- }});
- runner.run(1, true, true);
-
- runner.assertAllFlowFilesTransferred(PutElasticsearchHttp.REL_SUCCESS, 1);
- final MockFlowFile out = runner.getFlowFilesForRelationship(PutElasticsearchHttp.REL_SUCCESS).get(0);
- assertNotNull(out);
- out.assertAttributeEquals("doc_id", "28039652140");
- }
-
- @Test
- public void testPutElasticSearchOnTriggerEL() {
- runner = TestRunners.newTestRunner(new PutElasticsearchTestProcessor(false)); // no failures
- runner.setProperty(AbstractElasticsearchHttpProcessor.ES_URL, "${es.url}");
-
- runner.setProperty(PutElasticsearchHttp.INDEX, "doc");
- runner.setProperty(PutElasticsearchHttp.TYPE, "status");
- runner.setProperty(PutElasticsearchHttp.BATCH_SIZE, "1");
- runner.setProperty(PutElasticsearchHttp.ID_ATTRIBUTE, "doc_id");
- runner.setProperty(AbstractElasticsearchHttpProcessor.CONNECT_TIMEOUT, "${connect.timeout}");
- runner.assertValid();
-
- runner.setVariable("es.url", "http://127.0.0.1:9200");
- runner.setVariable("connect.timeout", "5s");
-
- runner.enqueue(docExample, new HashMap<String, String>() {{
- put("doc_id", "28039652140");
- }});
- runner.run(1, true, true);
-
- runner.assertAllFlowFilesTransferred(PutElasticsearchHttp.REL_SUCCESS, 1);
- final MockFlowFile out = runner.getFlowFilesForRelationship(PutElasticsearchHttp.REL_SUCCESS).get(0);
- assertNotNull(out);
- out.assertAttributeEquals("doc_id", "28039652140");
- }
-
- @Test
- public void testPutElasticSearchOnTriggerBadIndexOp() {
- runner = TestRunners.newTestRunner(new PutElasticsearchTestProcessor(false)); // no failures
- runner.setProperty(AbstractElasticsearchHttpProcessor.ES_URL, "http://127.0.0.1:9200");
-
- runner.setProperty(PutElasticsearchHttp.INDEX, "doc");
- runner.setProperty(PutElasticsearchHttp.TYPE, "status");
- runner.setProperty(PutElasticsearchHttp.BATCH_SIZE, "1");
- runner.setProperty(PutElasticsearchHttp.ID_ATTRIBUTE, "doc_id");
- runner.setProperty(PutElasticsearchHttp.INDEX_OP, "${no.attr}");
- runner.enqueue(docExample, new HashMap<String, String>() {{
- put("doc_id", "28039652140");
- }});
- runner.run(1, true, true);
-
- runner.assertAllFlowFilesTransferred(PutElasticsearchHttp.REL_FAILURE, 1);
- final MockFlowFile out = runner.getFlowFilesForRelationship(PutElasticsearchHttp.REL_FAILURE).get(0);
- assertNotNull(out);
- out.assertAttributeEquals("doc_id", "28039652140");
- }
-
- @Test
- public void testPutElasticSearchInvalidConfig() {
- runner = TestRunners.newTestRunner(new PutElasticsearchTestProcessor(false)); // no failures
- runner.setProperty(AbstractElasticsearchHttpProcessor.ES_URL, "http://127.0.0.1:9200");
-
- runner.assertNotValid();
- runner.setProperty(PutElasticsearchHttp.INDEX, "doc");
- runner.setProperty(PutElasticsearchHttp.TYPE, "status");
- runner.setProperty(PutElasticsearchHttp.BATCH_SIZE, "1");
- runner.assertValid();
- runner.setProperty(PutElasticsearchHttp.INDEX_OP, "");
- runner.assertNotValid();
- runner.setProperty(PutElasticsearchHttp.INDEX_OP, "index");
- runner.assertValid();
- runner.setProperty(PutElasticsearchHttp.INDEX_OP, "upsert");
- runner.assertNotValid();
- }
-
- @Test
- public void testPutElasticSearchOnTriggerWithFailures() {
- PutElasticsearchTestProcessor processor = new PutElasticsearchTestProcessor(true);
- processor.setStatus(100, "Should fail");
- runner = TestRunners.newTestRunner(processor); // simulate failures
- runner.setProperty(AbstractElasticsearchHttpProcessor.ES_URL, "http://127.0.0.1:9200");
- runner.setProperty(PutElasticsearchHttp.INDEX, "doc");
- runner.setProperty(PutElasticsearchHttp.TYPE, "status");
- runner.setProperty(PutElasticsearchHttp.BATCH_SIZE, "1");
- runner.setProperty(PutElasticsearchHttp.ID_ATTRIBUTE, "doc_id");
-
- runner.enqueue(docExample, new HashMap<String, String>() {{
- put("doc_id", "28039652140");
- }});
- runner.run(1, true, true);
- runner.assertAllFlowFilesTransferred(PutElasticsearchHttp.REL_FAILURE, 1);
- runner.clearTransferState();
-
- processor.setStatus(500, "Should retry");
- runner.enqueue(docExample, new HashMap<String, String>() {{
- put("doc_id", "28039652140");
- }});
- runner.run(1, true, true);
- runner.assertAllFlowFilesTransferred(PutElasticsearchHttp.REL_RETRY, 1);
- }
-
- @Test
- public void testPutElasticSearchOnTriggerWithConnectException() {
- PutElasticsearchTestProcessor processor = new PutElasticsearchTestProcessor(true);
- processor.setStatus(-1, "Connection Exception");
- runner = TestRunners.newTestRunner(processor); // simulate failures
- runner.setProperty(AbstractElasticsearchHttpProcessor.ES_URL, "http://127.0.0.1:9200");
- runner.setProperty(PutElasticsearchHttp.INDEX, "doc");
- runner.setProperty(PutElasticsearchHttp.TYPE, "status");
- runner.setProperty(PutElasticsearchHttp.BATCH_SIZE, "1");
- runner.setProperty(PutElasticsearchHttp.ID_ATTRIBUTE, "doc_id");
-
- runner.enqueue(docExample, new HashMap<String, String>() {{
- put("doc_id", "28039652140");
- }});
- runner.run(1, true, true);
- runner.assertAllFlowFilesTransferred(PutElasticsearchHttp.REL_FAILURE, 1);
- }
-
- @Test
- public void testPutElasticsearchOnTriggerWithNoIdAttribute() {
- runner = TestRunners.newTestRunner(new PutElasticsearchTestProcessor(true)); // simulate failures
- runner.setProperty(AbstractElasticsearchHttpProcessor.ES_URL, "http://127.0.0.1:9200");
- runner.setProperty(PutElasticsearchHttp.INDEX, "doc");
- runner.setProperty(PutElasticsearchHttp.TYPE, "status");
- runner.setProperty(PutElasticsearchHttp.BATCH_SIZE, "2");
- runner.setProperty(PutElasticsearchHttp.ID_ATTRIBUTE, "doc_id");
-
- runner.enqueue(docExample);
- runner.enqueue(docExample);
- runner.run(1, true, true);
-
- runner.assertTransferCount(PutElasticsearchHttp.REL_FAILURE, 1);
- runner.assertTransferCount(PutElasticsearchHttp.REL_SUCCESS, 1);
- final MockFlowFile out = runner.getFlowFilesForRelationship(PutElasticsearchHttp.REL_FAILURE).get(0);
- assertNotNull(out);
- }
-
- @Test
- public void testPutElasticsearchOnTriggerWithIndexFromAttribute() {
- runner = TestRunners.newTestRunner(new PutElasticsearchTestProcessor(false));
- runner.setProperty(AbstractElasticsearchHttpProcessor.ES_URL, "http://127.0.0.1:9200");
- runner.setProperty(PutElasticsearchHttp.INDEX, "${i}");
- runner.setProperty(PutElasticsearchHttp.TYPE, "${type}");
- runner.setProperty(PutElasticsearchHttp.BATCH_SIZE, "1");
- runner.setProperty(PutElasticsearchHttp.ID_ATTRIBUTE, "doc_id");
-
- runner.enqueue(docExample, new HashMap<String, String>() {{
- put("doc_id", "28039652144");
- put("i", "doc");
- put("type", "status");
- }});
- runner.run(1, true, true);
-
- runner.assertAllFlowFilesTransferred(PutElasticsearchHttp.REL_SUCCESS, 1);
- final MockFlowFile out = runner.getFlowFilesForRelationship(PutElasticsearchHttp.REL_SUCCESS).get(0);
- assertNotNull(out);
- runner.clearTransferState();
-
- // Now try an empty attribute value, should fail
- runner.enqueue(docExample, new HashMap<String, String>() {{
- put("doc_id", "28039652144");
- put("type", "status");
- }});
- runner.run(1, true, true);
-
- runner.assertAllFlowFilesTransferred(PutElasticsearchHttp.REL_FAILURE, 1);
- final MockFlowFile out2 = runner.getFlowFilesForRelationship(PutElasticsearchHttp.REL_FAILURE).get(0);
- assertNotNull(out2);
- }
-
- @Test
- public void testPutElasticSearchOnTriggerWithInvalidIndexOp() {
- runner = TestRunners.newTestRunner(new PutElasticsearchTestProcessor(false)); // no failures
- runner.setProperty(AbstractElasticsearchHttpProcessor.ES_URL, "http://127.0.0.1:9200");
-
- runner.assertNotValid();
- runner.setProperty(PutElasticsearchHttp.INDEX, "doc");
- runner.setProperty(PutElasticsearchHttp.TYPE, "");
- runner.assertNotValid();
- runner.setProperty(PutElasticsearchHttp.TYPE, " ");
- runner.assertValid();
- runner.removeProperty(PutElasticsearchHttp.TYPE);
- runner.assertValid();
- runner.setProperty(PutElasticsearchHttp.TYPE, "status");
- runner.assertValid();
- runner.setProperty(PutElasticsearchHttp.TYPE, "${type}");
- runner.assertValid();
- runner.setProperty(PutElasticsearchHttp.TYPE, "_doc");
- runner.assertValid();
- runner.setProperty(PutElasticsearchHttp.BATCH_SIZE, "1");
- runner.assertValid();
- runner.setProperty(PutElasticsearchHttp.INDEX_OP, "index");
- runner.assertValid();
- runner.setProperty(PutElasticsearchHttp.INDEX_OP, "create");
- runner.assertValid();
- runner.setProperty(PutElasticsearchHttp.ID_ATTRIBUTE, "doc_id");
- runner.assertValid();
-
- runner.setProperty(PutElasticsearchHttp.INDEX_OP, "index_fail");
- runner.assertValid();
-
- runner.enqueue(docExample, new HashMap<String, String>() {{
- put("doc_id", "28039652140");
- }});
- runner.run(1, true, true);
-
- runner.assertAllFlowFilesTransferred(PutElasticsearchHttp.REL_FAILURE, 1);
- final MockFlowFile out = runner.getFlowFilesForRelationship(PutElasticsearchHttp.REL_FAILURE).get(0);
- assertNotNull(out);
- }
-
- @Test
- public void testPutElasticSearchOnTriggerQueryParameter() {
- PutElasticsearchTestProcessor p = new PutElasticsearchTestProcessor(false); // no failures
- p.setExpectedUrl("http://127.0.0.1:9200/_bulk?pipeline=my-pipeline");
-
- runner = TestRunners.newTestRunner(p);
- runner.setProperty(AbstractElasticsearchHttpProcessor.ES_URL, "http://127.0.0.1:9200");
- runner.setProperty(PutElasticsearchHttp.INDEX, "doc");
- runner.setProperty(PutElasticsearchHttp.TYPE, "status");
- runner.setProperty(PutElasticsearchHttp.BATCH_SIZE, "1");
- runner.setProperty(PutElasticsearchHttp.ID_ATTRIBUTE, "doc_id");
-
- // Set dynamic property, to be added to the URL as a query parameter
- runner.setProperty("pipeline", "my-pipeline");
-
- runner.enqueue(docExample, new HashMap<String, String>() {{
- put("doc_id", "28039652140");
- }});
- runner.run(1, true, true);
-
- runner.assertAllFlowFilesTransferred(PutElasticsearchHttp.REL_SUCCESS, 1);
- final MockFlowFile out = runner.getFlowFilesForRelationship(PutElasticsearchHttp.REL_SUCCESS).get(0);
- assertNotNull(out);
- }
-
- @Test
- public void testPutElasticSearchOnTriggerWithDocumentNotFound() {
- PutElasticsearchTestProcessor processor = new PutElasticsearchTestProcessor(true);
- processor.setResultField("not_found");
- runner = TestRunners.newTestRunner(processor); // simulate failures
- runner.setProperty(AbstractElasticsearchHttpProcessor.ES_URL, "http://127.0.0.1:9200");
- runner.setProperty(PutElasticsearchHttp.INDEX_OP, "delete");
- runner.setProperty(PutElasticsearchHttp.INDEX, "doc");
- runner.setProperty(PutElasticsearchHttp.TYPE, "status");
- runner.setProperty(PutElasticsearchHttp.BATCH_SIZE, "1");
- runner.setProperty(PutElasticsearchHttp.ID_ATTRIBUTE, "doc_id");
-
- runner.enqueue(docExample, new HashMap<String, String>() {{
- put("doc_id", "28039652140");
- }});
- runner.run(1, true, true);
- runner.assertAllFlowFilesTransferred(PutElasticsearchHttp.REL_FAILURE, 1);
- runner.clearTransferState();
- }
-
- /**
- * A Test class that extends the processor in order to inject/mock behavior
- */
- private static class PutElasticsearchTestProcessor extends PutElasticsearchHttp {
- boolean responseHasFailures = false;
- OkHttpClient client;
- int statusCode = 200;
- String statusMessage = "OK";
- String expectedUrl = null;
- String resultField = null;
-
- PutElasticsearchTestProcessor(boolean responseHasFailures) {
- this.responseHasFailures = responseHasFailures;
- }
-
- void setStatus(int code, String message) {
- statusCode = code;
- statusMessage = message;
- }
-
- void setExpectedUrl(String url) {
- expectedUrl = url;
- }
-
- public void setResultField(String resultField) {
- this.resultField = resultField;
- }
-
- @Override
- protected void createElasticsearchClient(ProcessContext context) throws ProcessException {
- client = mock(OkHttpClient.class);
-
- when(client.newCall(any(Request.class))).thenAnswer((Answer<Call>) invocationOnMock -> {
- final Call call = mock(Call.class);
- if (statusCode != -1) {
- Request realRequest = (Request) invocationOnMock.getArguments()[0];
- assertTrue((expectedUrl == null) || (expectedUrl.equals(realRequest.url().toString())));
- StringBuilder sb = new StringBuilder("{\"took\": 1, \"errors\": \"");
- sb.append(responseHasFailures);
- sb.append("\", \"items\": [");
- if (responseHasFailures) {
- // This case is for a status code of 200 for the bulk response itself, but with an error (of 400) inside
- sb.append("{\"index\":{\"_index\":\"doc\",\"_type\":\"status\",\"_id\":\"28039652140\",\"status\":\"400\",");
- if(resultField != null) {
- sb.append("\"result\":{\"not_found\",");
- } else {
- sb.append("\"error\":{\"type\":\"mapper_parsing_exception\",\"reason\":\"failed to parse [gender]\",");
- }
- sb.append("\"caused_by\":{\"type\":\"json_parse_exception\",\"reason\":\"Unexpected end-of-input in VALUE_STRING\\n at ");
- sb.append("[Source: org.elasticsearch.common.io.stream.InputStreamStreamInput@1a2e3ac4; line: 1, column: 39]\"}}}},");
- }
- sb.append("{\"index\":{\"_index\":\"doc\",\"_type\":\"status\",\"_id\":\"28039652140\",\"status\":");
- sb.append(statusCode);
- sb.append(",\"_source\":{\"text\": \"This is a test document\"}}}");
-
- sb.append("]}");
- Response mockResponse = new Response.Builder()
- .request(realRequest)
- .protocol(Protocol.HTTP_1_1)
- .code(statusCode)
- .message(statusMessage)
- .body(ResponseBody.create(MediaType.parse("application/json"), sb.toString()))
- .build();
-
- when(call.execute()).thenReturn(mockResponse);
- } else {
- when(call.execute()).thenThrow(ConnectException.class);
- }
- return call;
- });
- }
-
- @Override
- protected OkHttpClient getClient() {
- return client;
- }
- }
-
- @Test
- public void testPutElasticSearchBadHostInEL() {
- runner = TestRunners.newTestRunner(new PutElasticsearchTestProcessor(false)); // no failures
- runner.setProperty(AbstractElasticsearchHttpProcessor.ES_URL, "${es.url}");
-
- runner.setProperty(PutElasticsearchHttp.INDEX, "doc");
- runner.setProperty(PutElasticsearchHttp.TYPE, "status");
- runner.setProperty(PutElasticsearchHttp.BATCH_SIZE, "1");
- runner.setProperty(PutElasticsearchHttp.ID_ATTRIBUTE, "doc_id");
-
- runner.enqueue(docExample, new HashMap<String, String>() {{
- put("doc_id", "28039652140");
- }});
- assertThrows(AssertionError.class, () -> runner.run(1, true, true));
- }
-}
diff --git a/nifi-nar-bundles/nifi-elasticsearch-bundle/nifi-elasticsearch-processors/src/test/java/org/apache/nifi/processors/elasticsearch/TestPutElasticsearchHttpRecord.java b/nifi-nar-bundles/nifi-elasticsearch-bundle/nifi-elasticsearch-processors/src/test/java/org/apache/nifi/processors/elasticsearch/TestPutElasticsearchHttpRecord.java
deleted file mode 100644
index 047ae521db..0000000000
--- a/nifi-nar-bundles/nifi-elasticsearch-bundle/nifi-elasticsearch-processors/src/test/java/org/apache/nifi/processors/elasticsearch/TestPutElasticsearchHttpRecord.java
+++ /dev/null
@@ -1,898 +0,0 @@
-/*
- * 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 com.fasterxml.jackson.databind.ObjectMapper;
-import okhttp3.Call;
-import okhttp3.MediaType;
-import okhttp3.OkHttpClient;
-import okhttp3.Protocol;
-import okhttp3.Request;
-import okhttp3.Response;
-import okhttp3.ResponseBody;
-import okio.Buffer;
-import org.apache.nifi.processor.ProcessContext;
-import org.apache.nifi.processor.exception.ProcessException;
-import org.apache.nifi.provenance.ProvenanceEventRecord;
-import org.apache.nifi.provenance.ProvenanceEventType;
-import org.apache.nifi.reporting.InitializationException;
-import org.apache.nifi.serialization.record.MockRecordParser;
-import org.apache.nifi.serialization.record.MockRecordWriter;
-import org.apache.nifi.serialization.record.RecordFieldType;
-import org.apache.nifi.util.MockFlowFile;
-import org.apache.nifi.util.TestRunner;
-import org.apache.nifi.util.TestRunners;
-import org.junit.jupiter.api.AfterEach;
-import org.junit.jupiter.api.Test;
-
-import java.io.IOException;
-import java.math.BigDecimal;
-import java.net.ConnectException;
-import java.sql.Date;
-import java.sql.Time;
-import java.sql.Timestamp;
-import java.time.LocalDate;
-import java.time.LocalDateTime;
-import java.time.LocalTime;
-import java.time.format.DateTimeFormatter;
-import java.util.Collections;
-import java.util.HashMap;
-import java.util.List;
-import java.util.Map;
-import java.util.Objects;
-import java.util.function.Consumer;
-
-import static org.junit.jupiter.api.Assertions.assertEquals;
-import static org.junit.jupiter.api.Assertions.assertFalse;
-import static org.junit.jupiter.api.Assertions.assertNotNull;
-import static org.junit.jupiter.api.Assertions.assertThrows;
-import static org.junit.jupiter.api.Assertions.assertTrue;
-import static org.mockito.ArgumentMatchers.any;
-import static org.mockito.Mockito.mock;
-import static org.mockito.Mockito.when;
-
-public class TestPutElasticsearchHttpRecord {
- private static final int DATE_YEAR = 2018;
- private static final int DATE_MONTH = 12;
- private static final int DATE_DAY = 20;
- private static final int TIME_HOUR = 12;
- private static final int TIME_MINUTE = 55;
-
- private static final String ISO_DATE = String.format("%d-%d-%d", DATE_YEAR, DATE_MONTH, DATE_DAY);
- private static final String EXPECTED_DATE = String.format("%d/%d/%d", DATE_DAY, DATE_MONTH, DATE_YEAR);
- private static final LocalDateTime LOCAL_DATE_TIME = LocalDateTime.of(DATE_YEAR, DATE_MONTH, DATE_DAY, TIME_HOUR, TIME_MINUTE);
- private static final LocalDate LOCAL_DATE = LocalDate.of(DATE_YEAR, DATE_MONTH, DATE_DAY);
- private static final LocalTime LOCAL_TIME = LocalTime.of(TIME_HOUR, TIME_MINUTE);
-
- private TestRunner runner;
-
- @AfterEach
- public void teardown() {
- runner = null;
- }
-
- @Test
- public void testPutElasticSearchOnTriggerIndex() throws IOException {
- PutElasticsearchHttpRecordTestProcessor processor = new PutElasticsearchHttpRecordTestProcessor(false);
- DateTimeFormatter timeFormatter = DateTimeFormatter.ofPattern("h:m a");
- DateTimeFormatter dateTimeFormatter = DateTimeFormatter.ofPattern("d/M/yyyy h:m a");
- processor.setRecordChecks(record -> {
- assertEquals(1, record.get("id"));
- assertEquals("reç1", record.get("name"));
- assertEquals(101, record.get("code"));
- assertEquals(EXPECTED_DATE, record.get("date"));
- assertEquals(LOCAL_TIME.format(timeFormatter), record.get("time"));
- assertEquals(LOCAL_DATE_TIME.format(dateTimeFormatter), record.get("ts"));
- }, record -> {
- assertEquals(2, record.get("id"));
- assertEquals("reç2", record.get("name"));
- assertEquals(102, record.get("code"));
- assertEquals(EXPECTED_DATE, record.get("date"));
- assertEquals(LOCAL_TIME.format(timeFormatter), record.get("time"));
- assertEquals(LOCAL_DATE_TIME.format(dateTimeFormatter), record.get("ts"));
- }, record -> {
- assertEquals(3, record.get("id"));
- assertEquals("reç3", record.get("name"));
- assertEquals(103, record.get("code"));
- assertEquals(EXPECTED_DATE, record.get("date"));
- assertEquals(LOCAL_TIME.format(timeFormatter), record.get("time"));
- assertEquals(LOCAL_DATE_TIME.format(dateTimeFormatter), record.get("ts"));
- }, record -> {
- assertEquals(4, record.get("id"));
- assertEquals("reç4", record.get("name"));
- assertEquals(104, record.get("code"));
- assertEquals(EXPECTED_DATE, record.get("date"));
- assertEquals(LOCAL_TIME.format(timeFormatter), record.get("time"));
- assertEquals(LOCAL_DATE_TIME.format(dateTimeFormatter), record.get("ts"));
- });
- runner = TestRunners.newTestRunner(processor); // no failures
- generateTestData();
- runner.setProperty(AbstractElasticsearchHttpProcessor.ES_URL, "http://127.0.0.1:9200");
-
- runner.setProperty(PutElasticsearchHttpRecord.INDEX, "doc");
- runner.setProperty(PutElasticsearchHttpRecord.TYPE, "status");
- runner.setProperty(PutElasticsearchHttpRecord.ID_RECORD_PATH, "/id");
- runner.setProperty(PutElasticsearchHttpRecord.DATE_FORMAT, "d/M/yyyy");
- runner.setProperty(PutElasticsearchHttpRecord.TIME_FORMAT, "h:m a");
- runner.setProperty(PutElasticsearchHttpRecord.TIMESTAMP_FORMAT, "d/M/yyyy h:m a");
-
- runner.enqueue(new byte[0], Collections.singletonMap("doc_id", "28039652140"));
- runner.run(1, true, true);
-
- runner.assertAllFlowFilesTransferred(PutElasticsearchHttpRecord.REL_SUCCESS, 1);
- final MockFlowFile out = runner.getFlowFilesForRelationship(PutElasticsearchHttpRecord.REL_SUCCESS).get(0);
- assertNotNull(out);
- out.assertAttributeEquals("doc_id", "28039652140");
- out.assertAttributeEquals("record.count", "4");
- List<ProvenanceEventRecord> provEvents = runner.getProvenanceEvents();
- assertNotNull(provEvents);
- assertEquals(1, provEvents.size());
- assertEquals(ProvenanceEventType.SEND, provEvents.get(0).getEventType());
- }
-
- @Test
- public void testPutElasticSearchOnTriggerCreate() throws IOException {
- runner = TestRunners.newTestRunner(new PutElasticsearchHttpRecordTestProcessor(false)); // no failures
- generateTestData();
- runner.setProperty(AbstractElasticsearchHttpProcessor.ES_URL, "http://127.0.0.1:9200");
-
- runner.setProperty(PutElasticsearchHttpRecord.INDEX, "doc");
- runner.setProperty(PutElasticsearchHttpRecord.TYPE, "status");
- runner.setProperty(PutElasticsearchHttpRecord.INDEX_OP, "create");
-
- runner.enqueue(new byte[0], Collections.singletonMap("doc_id", "28039652140"));
- runner.run(1, true, true);
-
- runner.assertAllFlowFilesTransferred(PutElasticsearchHttpRecord.REL_SUCCESS, 1);
- final MockFlowFile out = runner.getFlowFilesForRelationship(PutElasticsearchHttpRecord.REL_SUCCESS).get(0);
- assertNotNull(out);
- out.assertAttributeEquals("doc_id", "28039652140");
- out.assertAttributeEquals("record.count", "4");
- List<ProvenanceEventRecord> provEvents = runner.getProvenanceEvents();
- assertNotNull(provEvents);
- assertEquals(1, provEvents.size());
- assertEquals(ProvenanceEventType.SEND, provEvents.get(0).getEventType());
- }
-
- @Test
- public void testPutElasticSearchOnTriggerIndex_withoutType() throws IOException {
- PutElasticsearchHttpRecordTestProcessor processor = new PutElasticsearchHttpRecordTestProcessor(false);
- DateTimeFormatter timeFormatter = DateTimeFormatter.ofPattern("h:m a");
- DateTimeFormatter dateTimeFormatter = DateTimeFormatter.ofPattern("d/M/yyyy h:m a");
- processor.setRecordChecks(record -> {
- assertEquals(1, record.get("id"));
- assertEquals("reç1", record.get("name"));
- assertEquals(101, record.get("code"));
- assertEquals("20/12/2018", record.get("date"));
- assertEquals(LOCAL_TIME.format(timeFormatter), record.get("time"));
- assertEquals(LOCAL_DATE_TIME.format(dateTimeFormatter), record.get("ts"));
- }, record -> {
- assertEquals(2, record.get("id"));
- assertEquals("reç2", record.get("name"));
- assertEquals(102, record.get("code"));
- assertEquals(EXPECTED_DATE, record.get("date"));
- assertEquals(LOCAL_TIME.format(timeFormatter), record.get("time"));
- assertEquals(LOCAL_DATE_TIME.format(dateTimeFormatter), record.get("ts"));
- }, record -> {
- assertEquals(3, record.get("id"));
- assertEquals("reç3", record.get("name"));
- assertEquals(103, record.get("code"));
- assertEquals(EXPECTED_DATE, record.get("date"));
- assertEquals(LOCAL_TIME.format(timeFormatter), record.get("time"));
- assertEquals(LOCAL_DATE_TIME.format(dateTimeFormatter), record.get("ts"));
- }, record -> {
- assertEquals(4, record.get("id"));
- assertEquals("reç4", record.get("name"));
- assertEquals(104, record.get("code"));
- assertEquals(EXPECTED_DATE, record.get("date"));
- assertEquals(LOCAL_TIME.format(timeFormatter), record.get("time"));
- assertEquals(LOCAL_DATE_TIME.format(dateTimeFormatter), record.get("ts"));
- });
- runner = TestRunners.newTestRunner(processor); // no failures
- generateTestData();
- runner.setProperty(AbstractElasticsearchHttpProcessor.ES_URL, "http://127.0.0.1:9200");
-
- runner.setProperty(PutElasticsearchHttpRecord.INDEX, "doc");
- runner.removeProperty(PutElasticsearchHttpRecord.TYPE);
- runner.setProperty(PutElasticsearchHttpRecord.ID_RECORD_PATH, "/id");
- runner.setProperty(PutElasticsearchHttpRecord.DATE_FORMAT, "d/M/yyyy");
- runner.setProperty(PutElasticsearchHttpRecord.TIME_FORMAT, "h:m a");
- runner.setProperty(PutElasticsearchHttpRecord.TIMESTAMP_FORMAT, "d/M/yyyy h:m a");
-
- runner.enqueue(new byte[0], Collections.singletonMap("doc_id", "28039652140"));
- runner.run(1, true, true);
-
- runner.assertAllFlowFilesTransferred(PutElasticsearchHttpRecord.REL_SUCCESS, 1);
- final MockFlowFile out = runner.getFlowFilesForRelationship(PutElasticsearchHttpRecord.REL_SUCCESS).get(0);
- assertNotNull(out);
- out.assertAttributeEquals("doc_id", "28039652140");
- out.assertAttributeEquals("record.count", "4");
- List<ProvenanceEventRecord> provEvents = runner.getProvenanceEvents();
- assertNotNull(provEvents);
- assertEquals(1, provEvents.size());
- assertEquals(ProvenanceEventType.SEND, provEvents.get(0).getEventType());
- }
-
- @Test
- public void testPutElasticSearchOnTriggerUpdate() throws IOException {
- runner = TestRunners.newTestRunner(new PutElasticsearchHttpRecordTestProcessor(false)); // no failures
- generateTestData();
- runner.setProperty(AbstractElasticsearchHttpProcessor.ES_URL, "http://127.0.0.1:9200");
-
- runner.setProperty(PutElasticsearchHttpRecord.INDEX, "doc");
- runner.setProperty(PutElasticsearchHttpRecord.TYPE, "status");
- runner.setProperty(PutElasticsearchHttpRecord.ID_RECORD_PATH, "/id");
- runner.setProperty(PutElasticsearchHttpRecord.INDEX_OP, "Update");
- runner.enqueue(new byte[0], Collections.singletonMap("doc_id", "28039652140"));
- runner.run(1, true, true);
-
- runner.assertAllFlowFilesTransferred(PutElasticsearchHttpRecord.REL_SUCCESS, 1);
- final MockFlowFile out = runner.getFlowFilesForRelationship(PutElasticsearchHttpRecord.REL_SUCCESS).get(0);
- assertNotNull(out);
- out.assertAttributeEquals("doc_id", "28039652140");
- }
-
- @Test
- public void testPutElasticSearchOnTriggerUpdate_withoutType() throws IOException {
- runner = TestRunners.newTestRunner(new PutElasticsearchHttpRecordTestProcessor(false)); // no failures
- generateTestData();
- runner.setProperty(AbstractElasticsearchHttpProcessor.ES_URL, "http://127.0.0.1:9200");
-
- runner.setProperty(PutElasticsearchHttpRecord.INDEX, "doc");
- runner.removeProperty(PutElasticsearchHttpRecord.TYPE);
- runner.setProperty(PutElasticsearchHttpRecord.ID_RECORD_PATH, "/id");
- runner.setProperty(PutElasticsearchHttpRecord.INDEX_OP, "Update");
- runner.enqueue(new byte[0], Collections.singletonMap("doc_id", "28039652140"));
- runner.run(1, true, true);
-
- runner.assertAllFlowFilesTransferred(PutElasticsearchHttpRecord.REL_SUCCESS, 1);
- final MockFlowFile out = runner.getFlowFilesForRelationship(PutElasticsearchHttpRecord.REL_SUCCESS).get(0);
- assertNotNull(out);
- out.assertAttributeEquals("doc_id", "28039652140");
- }
-
- @Test
- public void testPutElasticSearchOnTriggerDelete() throws IOException {
- runner = TestRunners.newTestRunner(new PutElasticsearchHttpRecordTestProcessor(false)); // no failures
- generateTestData();
- runner.setProperty(AbstractElasticsearchHttpProcessor.ES_URL, "http://127.0.0.1:9200");
-
- runner.setProperty(PutElasticsearchHttpRecord.INDEX, "doc");
- runner.setProperty(PutElasticsearchHttpRecord.TYPE, "status");
- runner.setProperty(PutElasticsearchHttpRecord.ID_RECORD_PATH, "/id");
- runner.setProperty(PutElasticsearchHttpRecord.INDEX_OP, "DELETE");
- runner.enqueue(new byte[0], Collections.singletonMap("doc_id", "28039652140"));
- runner.run(1, true, true);
-
- runner.assertAllFlowFilesTransferred(PutElasticsearchHttpRecord.REL_SUCCESS, 1);
- final MockFlowFile out = runner.getFlowFilesForRelationship(PutElasticsearchHttpRecord.REL_SUCCESS).get(0);
- assertNotNull(out);
- out.assertAttributeEquals("doc_id", "28039652140");
- }
-
- @Test
- public void testPutElasticSearchOnTriggerDelete_withoutType() throws IOException {
- runner = TestRunners.newTestRunner(new PutElasticsearchHttpRecordTestProcessor(false)); // no failures
- generateTestData();
- runner.setProperty(AbstractElasticsearchHttpProcessor.ES_URL, "http://127.0.0.1:9200");
-
- runner.setProperty(PutElasticsearchHttpRecord.INDEX, "doc");
- runner.removeProperty(PutElasticsearchHttpRecord.TYPE);
- runner.setProperty(PutElasticsearchHttpRecord.ID_RECORD_PATH, "/id");
- runner.setProperty(PutElasticsearchHttpRecord.INDEX_OP, "DELETE");
- runner.enqueue(new byte[0], Collections.singletonMap("doc_id", "28039652140"));
- runner.run(1, true, true);
-
- runner.assertAllFlowFilesTransferred(PutElasticsearchHttpRecord.REL_SUCCESS, 1);
- final MockFlowFile out = runner.getFlowFilesForRelationship(PutElasticsearchHttpRecord.REL_SUCCESS).get(0);
- assertNotNull(out);
- out.assertAttributeEquals("doc_id", "28039652140");
- }
-
- @Test
- public void testPutElasticSearchOnTriggerEL() throws IOException {
- runner = TestRunners.newTestRunner(new PutElasticsearchHttpRecordTestProcessor(false)); // no failures
- generateTestData();
- runner.setProperty(AbstractElasticsearchHttpProcessor.ES_URL, "${es.url}");
-
- runner.setProperty(PutElasticsearchHttpRecord.INDEX, "doc");
- runner.setProperty(PutElasticsearchHttpRecord.TYPE, "status");
- runner.setProperty(PutElasticsearchHttpRecord.ID_RECORD_PATH, "/id");
- runner.setProperty(AbstractElasticsearchHttpProcessor.CONNECT_TIMEOUT, "${connect.timeout}");
- runner.assertValid();
-
- runner.setVariable("es.url", "http://127.0.0.1:9200");
- runner.setVariable("connect.timeout", "5s");
-
- runner.enqueue(new byte[0], Collections.singletonMap("doc_id", "28039652140"));
- runner.run(1, true, true);
-
- runner.assertAllFlowFilesTransferred(PutElasticsearchHttpRecord.REL_SUCCESS, 1);
- final MockFlowFile out = runner.getFlowFilesForRelationship(PutElasticsearchHttpRecord.REL_SUCCESS).get(0);
- assertNotNull(out);
- out.assertAttributeEquals("doc_id", "28039652140");
- }
-
- @Test
- public void testPutElasticSearchOnTriggerBadIndexOp() throws IOException {
- runner = TestRunners.newTestRunner(new PutElasticsearchHttpRecordTestProcessor(false)); // no failures
- generateTestData();
- runner.setProperty(AbstractElasticsearchHttpProcessor.ES_URL, "http://127.0.0.1:9200");
-
- runner.setProperty(PutElasticsearchHttpRecord.INDEX, "doc");
- runner.setProperty(PutElasticsearchHttpRecord.TYPE, "status");
- runner.setProperty(PutElasticsearchHttpRecord.ID_RECORD_PATH, "/id");
- runner.setProperty(PutElasticsearchHttpRecord.INDEX_OP, "${no.attr}");
- runner.enqueue(new byte[0], Collections.singletonMap("doc_id", "28039652140"));
- runner.run(1, true, true);
-
- runner.assertAllFlowFilesTransferred(PutElasticsearchHttpRecord.REL_FAILURE, 1);
- final MockFlowFile out = runner.getFlowFilesForRelationship(PutElasticsearchHttpRecord.REL_FAILURE).get(0);
- assertNotNull(out);
- out.assertAttributeEquals("doc_id", "28039652140");
- }
-
- @Test
- public void testPutElasticSearchInvalidConfig() throws IOException {
- runner = TestRunners.newTestRunner(new PutElasticsearchHttpRecordTestProcessor(false)); // no failures
- generateTestData();
- runner.setProperty(AbstractElasticsearchHttpProcessor.ES_URL, "http://127.0.0.1:9200");
-
- runner.assertNotValid();
- runner.setProperty(PutElasticsearchHttpRecord.INDEX, "doc");
- runner.setProperty(PutElasticsearchHttpRecord.TYPE, "status");
- runner.assertValid();
- runner.setProperty(PutElasticsearchHttpRecord.INDEX_OP, "");
- runner.assertNotValid();
- runner.setProperty(PutElasticsearchHttpRecord.INDEX_OP, "index");
- runner.assertValid();
- runner.setProperty(PutElasticsearchHttpRecord.INDEX_OP, "create");
- runner.assertValid();
- runner.setProperty(PutElasticsearchHttpRecord.INDEX_OP, "upsert");
- runner.assertNotValid();
- }
-
- @Test
- public void testPutElasticSearchOnTriggerWithFailures() throws IOException {
- PutElasticsearchHttpRecordTestProcessor processor = new PutElasticsearchHttpRecordTestProcessor(true);
- processor.setStatus(100, "Should fail");
- runner = TestRunners.newTestRunner(processor); // simulate failures
- generateTestData();
-
- runner.setProperty(AbstractElasticsearchHttpProcessor.ES_URL, "http://127.0.0.1:9200");
- runner.setProperty(PutElasticsearchHttpRecord.INDEX, "doc");
- runner.setProperty(PutElasticsearchHttpRecord.TYPE, "status");
- runner.setProperty(PutElasticsearchHttpRecord.ID_RECORD_PATH, "/id");
-
- runner.enqueue(new byte[0], Collections.singletonMap("doc_id", "28039652140"));
- runner.run(1, true, true);
- runner.assertAllFlowFilesTransferred(PutElasticsearchHttpRecord.REL_FAILURE, 1);
- runner.clearTransferState();
-
- processor.setStatus(500, "Should retry");
- runner.enqueue(new byte[0], Collections.singletonMap("doc_id", "28039652140"));
- runner.run(1, true, true);
- runner.assertAllFlowFilesTransferred(PutElasticsearchHttpRecord.REL_RETRY, 1);
- }
-
- @Test
- public void testPutElasticSearchOnTriggerWithConnectException() throws IOException {
- PutElasticsearchHttpRecordTestProcessor processor = new PutElasticsearchHttpRecordTestProcessor(true);
- processor.setStatus(-1, "Connection Exception");
- runner = TestRunners.newTestRunner(processor); // simulate failures
- generateTestData();
- runner.setProperty(AbstractElasticsearchHttpProcessor.ES_URL, "http://127.0.0.1:9200");
- runner.setProperty(PutElasticsearchHttpRecord.INDEX, "doc");
- runner.setProperty(PutElasticsearchHttpRecord.TYPE, "status");
- runner.setProperty(PutElasticsearchHttpRecord.ID_RECORD_PATH, "/id");
-
- runner.enqueue(new byte[0], Collections.singletonMap("doc_id", "28039652140"));
- runner.run(1, true, true);
- runner.assertAllFlowFilesTransferred(PutElasticsearchHttpRecord.REL_FAILURE, 1);
- }
-
- @Test
- public void testPutElasticsearchOnTriggerWithNoIdPath() throws Exception {
- runner = TestRunners.newTestRunner(new PutElasticsearchHttpRecordTestProcessor(false));
- generateTestData();
- runner.setProperty(AbstractElasticsearchHttpProcessor.ES_URL, "http://127.0.0.1:9200");
- runner.setProperty(PutElasticsearchHttpRecord.INDEX, "doc");
- runner.setProperty(PutElasticsearchHttpRecord.TYPE, "status");
- runner.setProperty(PutElasticsearchHttpRecord.ID_RECORD_PATH, "/none"); // Field does not exist
-
- runner.enqueue(new byte[0]);
- runner.run(1, true, true);
-
- runner.assertTransferCount(PutElasticsearchHttpRecord.REL_FAILURE, 1);
- runner.assertTransferCount(PutElasticsearchHttpRecord.REL_SUCCESS, 0);
- }
-
- @Test
- public void testPutElasticsearchOnTriggerWithNoIdField() throws IOException {
- runner = TestRunners.newTestRunner(new PutElasticsearchHttpRecordTestProcessor(true)); // simulate failures
- generateTestData();
- runner.setProperty(AbstractElasticsearchHttpProcessor.ES_URL, "http://127.0.0.1:9200");
- runner.setProperty(PutElasticsearchHttpRecord.INDEX, "doc");
- runner.setProperty(PutElasticsearchHttpRecord.TYPE, "status");
-
- runner.enqueue(new byte[0]);
- runner.run(1, true, true);
-
- runner.assertTransferCount(PutElasticsearchHttpRecord.REL_SUCCESS, 0);
- runner.assertTransferCount(PutElasticsearchHttpRecord.REL_FAILURE, 1);
- MockFlowFile flowFile = runner.getFlowFilesForRelationship(PutElasticsearchHttpRecord.REL_FAILURE).get(0);
- flowFile.assertAttributeEquals("failure.count", "1");
- }
-
- @Test
- public void testPutElasticsearchOnTriggerWithIndexFromAttribute() throws IOException {
- runner = TestRunners.newTestRunner(new PutElasticsearchHttpRecordTestProcessor(false));
- generateTestData();
- runner.setProperty(AbstractElasticsearchHttpProcessor.ES_URL, "http://127.0.0.1:9200");
- runner.setProperty(PutElasticsearchHttpRecord.INDEX, "${i}");
- runner.setProperty(PutElasticsearchHttpRecord.TYPE, "${type}");
- runner.setProperty(PutElasticsearchHttpRecord.ID_RECORD_PATH, "/id");
-
- runner.enqueue(new byte[0], new HashMap<String, String>() {{
- put("doc_id", "28039652144");
- put("i", "doc");
- put("type", "status");
- }});
- runner.run(1, true, true);
-
- runner.assertAllFlowFilesTransferred(PutElasticsearchHttpRecord.REL_SUCCESS, 1);
- final MockFlowFile out = runner.getFlowFilesForRelationship(PutElasticsearchHttpRecord.REL_SUCCESS).get(0);
- assertNotNull(out);
- runner.clearTransferState();
-
- // Now try an empty attribute value, should fail
- runner.enqueue(new byte[0], new HashMap<String, String>() {{
- put("doc_id", "28039652144");
- put("type", "status");
- }});
- runner.run(1, true, true);
-
- runner.assertAllFlowFilesTransferred(PutElasticsearchHttpRecord.REL_FAILURE, 1);
- final MockFlowFile out2 = runner.getFlowFilesForRelationship(PutElasticsearchHttpRecord.REL_FAILURE).get(0);
- assertNotNull(out2);
- }
-
- @Test
- public void testPutElasticSearchOnTriggerWithInvalidIndexOp() throws IOException {
- runner = TestRunners.newTestRunner(new PutElasticsearchHttpRecordTestProcessor(false)); // no failures
- generateTestData();
- runner.setProperty(AbstractElasticsearchHttpProcessor.ES_URL, "http://127.0.0.1:9200");
-
- runner.assertNotValid();
- runner.setProperty(PutElasticsearchHttpRecord.INDEX, "doc");
- runner.setProperty(PutElasticsearchHttpRecord.TYPE, "status");
- runner.assertValid();
- runner.setProperty(PutElasticsearchHttpRecord.ID_RECORD_PATH, "/id");
- runner.assertValid();
- runner.setProperty(PutElasticsearchHttpRecord.INDEX_OP, "index");
- runner.assertValid();
- runner.setProperty(PutElasticsearchHttpRecord.INDEX_OP, "create");
- runner.assertValid();
-
- runner.setProperty(PutElasticsearchHttpRecord.INDEX_OP, "index_fail");
- runner.assertValid();
-
- runner.enqueue(new byte[0], Collections.singletonMap("doc_id", "28039652140"));
- runner.run(1, true, true);
-
- runner.assertAllFlowFilesTransferred(PutElasticsearchHttpRecord.REL_FAILURE, 1);
- final MockFlowFile out = runner.getFlowFilesForRelationship(PutElasticsearchHttpRecord.REL_FAILURE).get(0);
- assertNotNull(out);
- }
-
- @Test
- public void testPutElasticsearchOnTriggerWithNoAtTimestampPath() throws Exception {
- PutElasticsearchHttpRecordTestProcessor processor = new PutElasticsearchHttpRecordTestProcessor(false);
- runner = TestRunners.newTestRunner(processor);
- generateTestData(1);
- runner.setProperty(AbstractElasticsearchHttpProcessor.ES_URL, "http://127.0.0.1:9200");
- runner.setProperty(PutElasticsearchHttpRecord.INDEX, "doc");
-
- runner.removeProperty(PutElasticsearchHttpRecord.AT_TIMESTAMP); // no default
- runner.setProperty(PutElasticsearchHttpRecord.AT_TIMESTAMP_RECORD_PATH, "/none"); // Field does not exist
- processor.setRecordChecks(record -> assertTimestamp(record, null)); // no @timestamp
- runner.enqueue(new byte[0]);
- runner.run(1, true, true);
-
- runner.assertAllFlowFilesTransferred(PutElasticsearchHttpRecord.REL_SUCCESS, 1);
- final MockFlowFile out = runner.getFlowFilesForRelationship(PutElasticsearchHttpRecord.REL_SUCCESS).get(0);
- assertNotNull(out);
- runner.clearTransferState();
-
- // now add a default @timestamp
- final String timestamp = "2020-11-27T14:37:00.000Z";
- runner.setProperty(PutElasticsearchHttpRecord.AT_TIMESTAMP, timestamp);
- processor.setRecordChecks(record -> assertTimestamp(record, timestamp)); // @timestamp defaulted
- runner.enqueue(new byte[0]);
- runner.run(1, true, true);
-
- runner.assertAllFlowFilesTransferred(PutElasticsearchHttpRecord.REL_SUCCESS, 1);
- final MockFlowFile out2 = runner.getFlowFilesForRelationship(PutElasticsearchHttpRecord.REL_SUCCESS).get(0);
- assertNotNull(out2);
- }
-
- @Test
- public void testPutElasticsearchOnTriggerWithAtTimestampFromAttribute() throws IOException {
- PutElasticsearchHttpRecordTestProcessor processor = new PutElasticsearchHttpRecordTestProcessor(false);
- runner = TestRunners.newTestRunner(processor);
- generateTestData(1);
- runner.setProperty(AbstractElasticsearchHttpProcessor.ES_URL, "http://127.0.0.1:9200");
- runner.setProperty(PutElasticsearchHttpRecord.INDEX, "${i}");
- runner.setProperty(PutElasticsearchHttpRecord.AT_TIMESTAMP, "${timestamp}");
-
- final String timestamp = "2020-11-27T15:10:00.000Z";
- processor.setRecordChecks(record -> assertTimestamp(record, timestamp));
- runner.enqueue(new byte[0], new HashMap<String, String>() {{
- put("doc_id", "28039652144");
- put("i", "doc");
- put("timestamp", timestamp);
- }});
- runner.run(1, true, true);
-
- runner.assertAllFlowFilesTransferred(PutElasticsearchHttpRecord.REL_SUCCESS, 1);
- final MockFlowFile out = runner.getFlowFilesForRelationship(PutElasticsearchHttpRecord.REL_SUCCESS).get(0);
- assertNotNull(out);
- runner.clearTransferState();
-
- // Now try an empty attribute value, should be no timestamp
- processor.setRecordChecks(record -> assertTimestamp(record, null));
- runner.enqueue(new byte[0], new HashMap<String, String>() {{
- put("doc_id", "28039652144");
- put("i", "doc");
- }});
- runner.run(1, true, true);
-
- runner.assertAllFlowFilesTransferred(PutElasticsearchHttpRecord.REL_SUCCESS, 1);
- final MockFlowFile out2 = runner.getFlowFilesForRelationship(PutElasticsearchHttpRecord.REL_SUCCESS).get(0);
- assertNotNull(out2);
- }
-
- @Test
- public void testPutElasticsearchOnTriggerWithAtTimstampPath() throws Exception {
- PutElasticsearchHttpRecordTestProcessor processor = new PutElasticsearchHttpRecordTestProcessor(false);
- DateTimeFormatter timeFormatter = DateTimeFormatter.ofPattern(RecordFieldType.TIME.getDefaultFormat());
- DateTimeFormatter dateTimeFormatter = DateTimeFormatter.ofPattern(RecordFieldType.TIMESTAMP.getDefaultFormat());
- DateTimeFormatter dateFormatter = DateTimeFormatter.ofPattern(RecordFieldType.DATE.getDefaultFormat());
- runner = TestRunners.newTestRunner(processor);
- generateTestData(1);
- runner.setProperty(AbstractElasticsearchHttpProcessor.ES_URL, "http://127.0.0.1:9200");
- runner.setProperty(PutElasticsearchHttpRecord.INDEX, "doc");
-
- runner.setProperty(PutElasticsearchHttpRecord.AT_TIMESTAMP_RECORD_PATH, "/ts"); // TIMESTAMP
- processor.setRecordChecks(record -> assertTimestamp(record, LOCAL_DATE_TIME.format(dateTimeFormatter)));
- runner.enqueue(new byte[0]);
- runner.run(1, true, true);
-
- runner.assertAllFlowFilesTransferred(PutElasticsearchHttpRecord.REL_SUCCESS, 1);
- assertNotNull(runner.getFlowFilesForRelationship(PutElasticsearchHttpRecord.REL_SUCCESS).get(0));
- runner.clearTransferState();
-
- runner.setProperty(PutElasticsearchHttpRecord.AT_TIMESTAMP_RECORD_PATH, "/date"); // DATE;
- processor.setRecordChecks(record -> assertTimestamp(record, LOCAL_DATE.format(dateFormatter)));
- runner.enqueue(new byte[0]);
- runner.run(1, true, true);
-
- runner.assertAllFlowFilesTransferred(PutElasticsearchHttpRecord.REL_SUCCESS, 1);
- assertNotNull(runner.getFlowFilesForRelationship(PutElasticsearchHttpRecord.REL_SUCCESS).get(0));
- runner.clearTransferState();
-
- runner.setProperty(PutElasticsearchHttpRecord.AT_TIMESTAMP_RECORD_PATH, "/time"); // TIME
- processor.setRecordChecks(record -> assertTimestamp(record, LOCAL_TIME.format(timeFormatter)));
- runner.enqueue(new byte[0]);
- runner.run(1, true, true);
-
- runner.assertAllFlowFilesTransferred(PutElasticsearchHttpRecord.REL_SUCCESS, 1);
- assertNotNull(runner.getFlowFilesForRelationship(PutElasticsearchHttpRecord.REL_SUCCESS).get(0));
- runner.clearTransferState();
-
- // these INT/STRING values might not make sense from an Elasticsearch point of view,
- // but we want to prove we can handle them being selected from the Record
- runner.setProperty(PutElasticsearchHttpRecord.AT_TIMESTAMP_RECORD_PATH, "/code"); // INT
- processor.setRecordChecks(record -> assertTimestamp(record, 101));
- runner.enqueue(new byte[0]);
- runner.run(1, true, true);
-
- runner.assertAllFlowFilesTransferred(PutElasticsearchHttpRecord.REL_SUCCESS, 1);
- assertNotNull(runner.getFlowFilesForRelationship(PutElasticsearchHttpRecord.REL_SUCCESS).get(0));
- runner.clearTransferState();
-
- runner.setProperty(PutElasticsearchHttpRecord.AT_TIMESTAMP_RECORD_PATH, "/name"); // STRING
- processor.setRecordChecks(record -> assertTimestamp(record, "reç1"));
- runner.enqueue(new byte[0]);
- runner.run(1, true, true);
-
- runner.assertAllFlowFilesTransferred(PutElasticsearchHttpRecord.REL_SUCCESS, 1);
- assertNotNull(runner.getFlowFilesForRelationship(PutElasticsearchHttpRecord.REL_SUCCESS).get(0));
- runner.clearTransferState();
-
- runner.setProperty(PutElasticsearchHttpRecord.AT_TIMESTAMP_RECORD_PATH, "/coerce"); // STRING coerced to LONG
- processor.setRecordChecks(record -> assertTimestamp(record, 1000));
- runner.enqueue(new byte[0]);
- runner.run(1, true, true);
-
- runner.assertAllFlowFilesTransferred(PutElasticsearchHttpRecord.REL_SUCCESS, 1);
- assertNotNull(runner.getFlowFilesForRelationship(PutElasticsearchHttpRecord.REL_SUCCESS).get(0));
- runner.clearTransferState();
- }
-
- @Test
- public void testPutElasticSearchOnTriggerQueryParameter() throws IOException {
- PutElasticsearchHttpRecordTestProcessor p = new PutElasticsearchHttpRecordTestProcessor(false); // no failures
- p.setExpectedUrl("http://127.0.0.1:9200/_bulk?pipeline=my-pipeline");
- runner = TestRunners.newTestRunner(p);
- generateTestData();
- runner.setProperty(AbstractElasticsearchHttpProcessor.ES_URL, "http://127.0.0.1:9200");
-
- runner.setProperty(PutElasticsearchHttpRecord.INDEX, "doc");
- runner.setProperty(PutElasticsearchHttpRecord.TYPE, "status");
- runner.setProperty(PutElasticsearchHttpRecord.ID_RECORD_PATH, "/id");
-
- // Set dynamic property, to be added to the URL as a query parameter
- runner.setProperty("pipeline", "my-pipeline");
-
- runner.enqueue(new byte[0], Collections.singletonMap("doc_id", "28039652140"));
- runner.run(1, true, true);
-
- runner.assertAllFlowFilesTransferred(PutElasticsearchHttpRecord.REL_SUCCESS, 1);
- final MockFlowFile out = runner.getFlowFilesForRelationship(PutElasticsearchHttpRecord.REL_SUCCESS).get(0);
- assertNotNull(out);
- out.assertAttributeEquals("doc_id", "28039652140");
- List<ProvenanceEventRecord> provEvents = runner.getProvenanceEvents();
- assertNotNull(provEvents);
- assertEquals(1, provEvents.size());
- assertEquals(ProvenanceEventType.SEND, provEvents.get(0).getEventType());
- }
-
- @Test
- public void testPutElasticsearchOnTriggerFailureWithWriter() throws IOException {
- runner = TestRunners.newTestRunner(new PutElasticsearchHttpRecordTestProcessor(true)); // simulate failures
- generateTestData(1);
- generateWriter();
- runner.setProperty(AbstractElasticsearchHttpProcessor.ES_URL, "http://127.0.0.1:9200");
- runner.setProperty(PutElasticsearchHttpRecord.INDEX, "doc");
- runner.setProperty(PutElasticsearchHttpRecord.TYPE, "status");
-
- runner.enqueue(new byte[0]);
- runner.run(1, true, true);
-
- runner.assertTransferCount(PutElasticsearchHttpRecord.REL_SUCCESS, 0);
- runner.assertTransferCount(PutElasticsearchHttpRecord.REL_FAILURE, 1);
- MockFlowFile flowFileFailure = runner.getFlowFilesForRelationship(PutElasticsearchHttpRecord.REL_FAILURE).get(0);
- flowFileFailure.assertAttributeEquals("failure.count", "1");
- }
-
- @Test
- public void testPutElasticsearchOnTriggerFailureWithWriterMultipleRecords() throws IOException {
- runner = TestRunners.newTestRunner(new PutElasticsearchHttpRecordTestProcessor(2)); // simulate failures
- generateTestData();
- generateWriter();
- runner.setProperty(AbstractElasticsearchHttpProcessor.ES_URL, "http://127.0.0.1:9200");
- runner.setProperty(PutElasticsearchHttpRecord.INDEX, "doc");
- runner.setProperty(PutElasticsearchHttpRecord.TYPE, "status");
-
- runner.enqueue(new byte[0]);
- runner.run(1, true, true);
-
- runner.assertTransferCount(PutElasticsearchHttpRecord.REL_SUCCESS, 1);
- runner.assertTransferCount(PutElasticsearchHttpRecord.REL_FAILURE, 1);
- MockFlowFile flowFileSuccess = runner.getFlowFilesForRelationship(PutElasticsearchHttpRecord.REL_SUCCESS).get(0);
- flowFileSuccess.assertAttributeEquals("record.count", "2");
- MockFlowFile flowFileFailure = runner.getFlowFilesForRelationship(PutElasticsearchHttpRecord.REL_FAILURE).get(0);
- flowFileFailure.assertAttributeEquals("record.count", "2");
- flowFileFailure.assertAttributeEquals("failure.count", "2");
-
- assertEquals(1, runner.getLogger().getErrorMessages().size());
- }
-
- @Test
- public void testPutElasticsearchOnTriggerFailureWithWriterMultipleRecordsLogging() throws IOException {
- runner = TestRunners.newTestRunner(new PutElasticsearchHttpRecordTestProcessor(2)); // simulate failures
- generateTestData();
- generateWriter();
- runner.setProperty(AbstractElasticsearchHttpProcessor.ES_URL, "http://127.0.0.1:9200");
- runner.setProperty(PutElasticsearchHttpRecord.INDEX, "doc");
- runner.setProperty(PutElasticsearchHttpRecord.TYPE, "status");
- runner.setProperty(PutElasticsearchHttpRecord.LOG_ALL_ERRORS, "true");
-
- runner.enqueue(new byte[0]);
- runner.run(1, true, true);
-
- runner.assertTransferCount(PutElasticsearchHttpRecord.REL_SUCCESS, 1);
- runner.assertTransferCount(PutElasticsearchHttpRecord.REL_FAILURE, 1);
- MockFlowFile flowFileSuccess = runner.getFlowFilesForRelationship(PutElasticsearchHttpRecord.REL_SUCCESS).get(0);
- flowFileSuccess.assertAttributeEquals("record.count", "2");
- MockFlowFile flowFileFailure = runner.getFlowFilesForRelationship(PutElasticsearchHttpRecord.REL_FAILURE).get(0);
- flowFileFailure.assertAttributeEquals("record.count", "2");
- flowFileFailure.assertAttributeEquals("failure.count", "2");
-
- assertEquals(2, runner.getLogger().getErrorMessages().size());
- }
-
- /**
- * A Test class that extends the processor in order to inject/mock behavior
- */
- private static class PutElasticsearchHttpRecordTestProcessor extends PutElasticsearchHttpRecord {
- int numResponseFailures;
- OkHttpClient client;
- int statusCode = 200;
- String statusMessage = "OK";
- String expectedUrl = null;
- Consumer<Map<String, Object>>[] recordChecks;
-
- PutElasticsearchHttpRecordTestProcessor(boolean responseHasFailures) {
- this.numResponseFailures = responseHasFailures ? 1 : 0;
- }
-
- PutElasticsearchHttpRecordTestProcessor(int numResponseFailures) {
- this.numResponseFailures = numResponseFailures;
- }
-
- void setStatus(int code, String message) {
- statusCode = code;
- statusMessage = message;
- }
-
- void setExpectedUrl(String url) {
- expectedUrl = url;
- }
-
- @SafeVarargs
- final void setRecordChecks(Consumer<Map<String, Object>>... checks) {
- recordChecks = checks;
- }
-
- @SuppressWarnings("unchecked")
- @Override
- protected void createElasticsearchClient(ProcessContext context) throws ProcessException {
- client = mock(OkHttpClient.class);
-
- when(client.newCall(any(Request.class))).thenAnswer(invocationOnMock -> {
- final Call call = mock(Call.class);
- if (statusCode != -1) {
- Request realRequest = (Request) invocationOnMock.getArguments()[0];
- assertTrue((expectedUrl == null) || (expectedUrl.equals(realRequest.url().toString())));
- if (recordChecks != null) {
- final ObjectMapper mapper = new ObjectMapper();
- Buffer sink = new Buffer();
- Objects.requireNonNull(realRequest.body()).writeTo(sink);
- String line;
- int recordIndex = 0;
- boolean content = false;
- while ((line = sink.readUtf8Line()) != null) {
- if (content) {
- content = false;
- if (recordIndex < recordChecks.length) {
- recordChecks[recordIndex++].accept(mapper.readValue(line, Map.class));
- }
- } else {
- content = true;
- }
- }
- }
- StringBuilder sb = new StringBuilder("{\"took\": 1, \"errors\": \"");
- sb.append(numResponseFailures > 0);
- sb.append("\", \"items\": [");
- for (int i = 0; i < numResponseFailures; i ++) {
- // This case is for a status code of 200 for the bulk response itself, but with an error (of 400) inside
- sb.append("{\"index\":{\"_index\":\"doc\",\"_type\":\"status\",\"_id\":\"28039652140\",\"status\":\"400\",");
- sb.append("\"error\":{\"type\":\"mapper_parsing_exception\",\"reason\":\"failed to parse [gender]\",");
- sb.append("\"caused_by\":{\"type\":\"json_parse_exception\",\"reason\":\"Unexpected end-of-input in VALUE_STRING\\n at ");
- sb.append("[Source: org.elasticsearch.common.io.stream.InputStreamStreamInput@1a2e3ac4; line: 1, column: 39]\"}}}},");
- }
- sb.append("{\"index\":{\"_index\":\"doc\",\"_type\":\"status\",\"_id\":\"28039652140\",\"status\":");
- sb.append(statusCode);
- sb.append(",\"_source\":{\"text\": \"This is a test document\"}}}");
-
- sb.append("]}");
- Response mockResponse = new Response.Builder()
- .request(realRequest)
- .protocol(Protocol.HTTP_1_1)
- .code(statusCode)
- .message(statusMessage)
- .body(ResponseBody.create(sb.toString(), MediaType.parse("application/json")))
- .build();
-
- when(call.execute()).thenReturn(mockResponse);
- } else {
- when(call.execute()).thenThrow(ConnectException.class);
- }
- return call;
- });
- }
-
- @Override
- protected OkHttpClient getClient() {
- return client;
- }
- }
-
- @Test
- public void testPutElasticSearchBadHostInEL() {
- final TestRunner runner = TestRunners.newTestRunner(new PutElasticsearchHttpRecord());
-
- runner.setProperty(AbstractElasticsearchHttpProcessor.ES_URL, "${es.url}");
- runner.setProperty(PutElasticsearchHttpRecord.INDEX, "doc");
- runner.setProperty(PutElasticsearchHttpRecord.TYPE, "status");
- runner.setProperty(PutElasticsearchHttpRecord.ID_RECORD_PATH, "/id");
- assertThrows(AssertionError.class, () -> {
- runner.assertValid();
- runner.enqueue(new byte[0], new HashMap<String, String>() {{
- put("doc_id", "1");
- }});
-
- runner.enqueue(new byte[0], Collections.singletonMap("doc_id", "1"));
-
- runner.run();
- });
- }
-
- private void generateTestData() throws IOException {
- generateTestData(4);
- }
-
- private void generateTestData(int numRecords) throws IOException {
-
- final MockRecordParser parser = new MockRecordParser();
- try {
- runner.addControllerService("parser", parser);
- } catch (InitializationException e) {
- throw new IOException(e);
- }
- runner.enableControllerService(parser);
- runner.setProperty(PutElasticsearchHttpRecord.RECORD_READER, "parser");
-
- parser.addSchemaField("id", RecordFieldType.INT);
- parser.addSchemaField("name", RecordFieldType.STRING);
- parser.addSchemaField("code", RecordFieldType.INT);
- parser.addSchemaField("date", RecordFieldType.DATE);
- parser.addSchemaField("time", RecordFieldType.TIME);
- parser.addSchemaField("ts", RecordFieldType.TIMESTAMP);
- parser.addSchemaField("amount", RecordFieldType.DECIMAL);
- parser.addSchemaField("coerce", RecordFieldType.STRING);
-
- final Date date = Date.valueOf(ISO_DATE);
- final Timestamp timestamp = Timestamp.valueOf(LOCAL_DATE_TIME);
- final Time time = Time.valueOf(LOCAL_TIME);
- for(int i=1; i<=numRecords; i++) {
- parser.addRecord(i, "reç" + i, 100 + i, date, time, timestamp, new BigDecimal(Double.MAX_VALUE).multiply(BigDecimal.TEN), "1000");
- }
- }
-
- private void generateWriter() throws IOException {
- final MockRecordWriter writer = new MockRecordWriter();
- try {
- runner.addControllerService("writer", writer);
- } catch (InitializationException e) {
- throw new IOException(e);
- }
- runner.enableControllerService(writer);
- runner.setProperty(PutElasticsearchHttpRecord.RECORD_WRITER, "writer");
- }
-
- private void assertTimestamp(final Map<String, Object> record, final Object timestamp) {
- if (timestamp == null) {
- assertFalse(record.containsKey("@timestamp"));
- } else {
- assertEquals(timestamp, record.get("@timestamp"));
- }
- }
-}
diff --git a/nifi-nar-bundles/nifi-elasticsearch-bundle/nifi-elasticsearch-processors/src/test/java/org/apache/nifi/processors/elasticsearch/TestQueryElasticsearchHttp.java b/nifi-nar-bundles/nifi-elasticsearch-bundle/nifi-elasticsearch-processors/src/test/java/org/apache/nifi/processors/elasticsearch/TestQueryElasticsearchHttp.java
deleted file mode 100644
index c072f2ada9..0000000000
--- a/nifi-nar-bundles/nifi-elasticsearch-bundle/nifi-elasticsearch-processors/src/test/java/org/apache/nifi/processors/elasticsearch/TestQueryElasticsearchHttp.java
+++ /dev/null
@@ -1,522 +0,0 @@
-/*
- * 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 okhttp3.Call;
-import okhttp3.MediaType;
-import okhttp3.OkHttpClient;
-import okhttp3.Protocol;
-import okhttp3.Request;
-import okhttp3.Response;
-import okhttp3.ResponseBody;
-import org.apache.commons.io.IOUtils;
-import org.apache.nifi.processor.ProcessContext;
-import org.apache.nifi.processor.exception.ProcessException;
-import org.apache.nifi.ssl.SSLContextService;
-import org.apache.nifi.util.MockFlowFile;
-import org.apache.nifi.util.TestRunner;
-import org.apache.nifi.util.TestRunners;
-
-import org.junit.jupiter.api.AfterEach;
-import org.junit.jupiter.api.Test;
-import org.mockito.stubbing.Answer;
-import org.mockito.stubbing.OngoingStubbing;
-
-import java.io.IOException;
-import java.nio.charset.StandardCharsets;
-import java.util.Arrays;
-import java.util.HashMap;
-import java.util.List;
-
-import static org.junit.jupiter.api.Assertions.assertEquals;
-import static org.junit.jupiter.api.Assertions.assertNotNull;
-import static org.junit.jupiter.api.Assertions.assertTrue;
-import static org.mockito.ArgumentMatchers.any;
-import static org.mockito.Mockito.mock;
-import static org.mockito.Mockito.when;
-
-public class TestQueryElasticsearchHttp {
-
- private TestRunner runner;
-
- @AfterEach
- public void teardown() {
- runner = null;
- }
-
- @Test
- public void testQueryElasticsearchOnTrigger_withInput() {
- runner = TestRunners.newTestRunner(new QueryElasticsearchHttpTestProcessor());
- runner.setProperty(AbstractElasticsearchHttpProcessor.ES_URL, "http://127.0.0.1:9200");
-
- runner.setProperty(QueryElasticsearchHttp.INDEX, "doc");
- runner.assertNotValid();
- runner.setProperty(QueryElasticsearchHttp.TYPE, "status");
- runner.assertNotValid();
- runner.setProperty(QueryElasticsearchHttp.QUERY,
- "source:Twitter AND identifier:\"${identifier}\"");
- runner.assertValid();
- runner.setProperty(QueryElasticsearchHttp.PAGE_SIZE, "2");
- runner.assertValid();
-
- runAndVerifySuccess(true);
- }
-
- @Test
- public void testQueryElasticsearchOnTrigger_withInput_withQueryInAttrs() {
- runner = TestRunners.newTestRunner(new QueryElasticsearchHttpTestProcessor());
- runner.setValidateExpressionUsage(true);
- runner.setProperty(AbstractElasticsearchHttpProcessor.ES_URL, "http://127.0.0.1:9200");
-
- runner.setProperty(QueryElasticsearchHttp.INDEX, "doc");
- runner.assertNotValid();
- runner.setProperty(QueryElasticsearchHttp.TYPE, "status");
- runner.assertNotValid();
- runner.setProperty(QueryElasticsearchHttp.QUERY,
- "source:Twitter AND identifier:\"${identifier}\"");
- runner.assertValid();
- runner.setProperty(QueryElasticsearchHttp.PAGE_SIZE, "2");
- runner.assertValid();
-
- runAndVerifySuccess(true);
- }
-
- @Test
- public void testQueryElasticsearchOnTrigger_withInput_EL() {
- runner = TestRunners.newTestRunner(new QueryElasticsearchHttpTestProcessor());
- runner.setProperty(AbstractElasticsearchHttpProcessor.ES_URL, "${es.url}");
-
- runner.setProperty(QueryElasticsearchHttp.INDEX, "doc");
- runner.assertNotValid();
- runner.setProperty(QueryElasticsearchHttp.QUERY,
- "source:Twitter AND identifier:\"${identifier}\"");
- runner.removeProperty(QueryElasticsearchHttp.TYPE);
- runner.assertValid();
- runner.setProperty(QueryElasticsearchHttp.TYPE, "status");
- runner.assertValid();
- runner.setProperty(QueryElasticsearchHttp.TYPE, "");
- runner.assertNotValid();
- runner.setProperty(QueryElasticsearchHttp.TYPE, "${type}");
- runner.assertValid();
- runner.setProperty(QueryElasticsearchHttp.TYPE, "_doc");
- runner.assertValid();
- runner.setProperty(QueryElasticsearchHttp.PAGE_SIZE, "2");
- runner.assertValid();
- runner.setProperty(AbstractElasticsearchHttpProcessor.CONNECT_TIMEOUT, "${connect.timeout}");
- runner.assertValid();
-
- runner.setVariable("es.url", "http://127.0.0.1:9200");
-
- runAndVerifySuccess(true);
- }
-
- @Test
- public void testQueryElasticsearchOnTrigger_withInput_attributeTarget() {
- runner = TestRunners.newTestRunner(new QueryElasticsearchHttpTestProcessor());
- runner.setProperty(AbstractElasticsearchHttpProcessor.ES_URL, "http://127.0.0.1:9200");
-
- runner.setProperty(QueryElasticsearchHttp.INDEX, "doc");
- runner.assertNotValid();
- runner.setProperty(QueryElasticsearchHttp.TYPE, "status");
- runner.assertNotValid();
- runner.setProperty(QueryElasticsearchHttp.QUERY,
- "source:Twitter AND identifier:\"${identifier}\"");
- runner.assertValid();
- runner.setProperty(QueryElasticsearchHttp.PAGE_SIZE, "2");
- runner.assertValid();
- runner.setProperty(QueryElasticsearchHttp.TARGET,
- QueryElasticsearchHttp.TARGET_FLOW_FILE_ATTRIBUTES);
-
- runAndVerifySuccess(false);
- final MockFlowFile out = runner.getFlowFilesForRelationship(
- QueryElasticsearchHttp.REL_SUCCESS).get(0);
- assertNotNull(out);
- assertEquals("blah", new String(out.toByteArray()));
- assertEquals("arrays,are,supported,too", out.getAttribute("es.result.tags"));
- assertEquals("Twitter", out.getAttribute("es.result.source"));
- }
-
- @Test
- public void testQueryElasticsearchOnTrigger_withNoInput() {
- runner = TestRunners.newTestRunner(new QueryElasticsearchHttpTestProcessor());
- runner.setProperty(AbstractElasticsearchHttpProcessor.ES_URL, "http://127.0.0.1:9200");
-
- runner.setProperty(QueryElasticsearchHttp.INDEX, "doc");
- runner.assertNotValid();
- runner.setProperty(QueryElasticsearchHttp.TYPE, "status");
- runner.assertNotValid();
- runner.setProperty(QueryElasticsearchHttp.QUERY,
- "source:Twitter AND identifier:\"${identifier}\"");
- runner.assertValid();
- runner.setProperty(QueryElasticsearchHttp.PAGE_SIZE, "2");
- runner.assertValid();
-
- runner.setIncomingConnection(false);
- runAndVerifySuccess(true);
- }
-
- private void runAndVerifySuccess(int expectedResults, boolean targetIsContent) {
- runner.enqueue("blah".getBytes(), new HashMap<String, String>() {
- {
- put("identifier", "28039652140");
- }
- });
-
- // Running once should page through all 3 docs
- runner.run(1, true, true);
-
- runner.assertAllFlowFilesTransferred(QueryElasticsearchHttp.REL_SUCCESS, expectedResults);
- final MockFlowFile out = runner.getFlowFilesForRelationship(
- QueryElasticsearchHttp.REL_SUCCESS).get(0);
- assertNotNull(out);
- if (targetIsContent) {
- out.assertAttributeEquals("filename", "abc-97b-ASVsZu_"
- + "vShwtGCJpGOObmuSqUJRUC3L_-SEND-S3");
- }
- out.assertAttributeExists("es.query.url");
- }
-
- // By default, 3 files should go to Success
- private void runAndVerifySuccess(boolean targetIsContent) {
- runAndVerifySuccess(3, targetIsContent);
- }
-
- @Test
- public void testQueryElasticsearchOnTriggerWithFields() {
- runner = TestRunners.newTestRunner(new QueryElasticsearchHttpTestProcessor());
- runner.setProperty(AbstractElasticsearchHttpProcessor.ES_URL, "http://127.0.0.1:9200");
-
- runner.setProperty(QueryElasticsearchHttp.INDEX, "doc");
- runner.assertNotValid();
- runner.setProperty(QueryElasticsearchHttp.TYPE, "status");
- runner.assertNotValid();
- runner.setProperty(QueryElasticsearchHttp.QUERY, "${doc_id}");
- runner.assertValid();
- runner.setProperty(QueryElasticsearchHttp.FIELDS, "id,, userinfo.location");
- runner.assertValid();
- runner.setProperty(QueryElasticsearchHttp.SORT, "timestamp:asc,identifier:desc");
- runner.assertValid();
-
- runAndVerifySuccess(true);
- }
-
- @Test
- public void testQueryElasticsearchOnTriggerWithLimit() {
- runner = TestRunners.newTestRunner(new QueryElasticsearchHttpTestProcessor());
- runner.setProperty(AbstractElasticsearchHttpProcessor.ES_URL, "http://127.0.0.1:9200");
-
- runner.setProperty(QueryElasticsearchHttp.INDEX, "doc");
- runner.assertNotValid();
- runner.setProperty(QueryElasticsearchHttp.TYPE, "status");
- runner.assertNotValid();
- runner.setProperty(QueryElasticsearchHttp.QUERY, "${doc_id}");
- runner.assertValid();
- runner.setProperty(QueryElasticsearchHttp.FIELDS, "id,, userinfo.location");
- runner.assertValid();
- runner.setProperty(QueryElasticsearchHttp.SORT, "timestamp:asc,identifier:desc");
- runner.assertValid();
- runner.setProperty(QueryElasticsearchHttp.LIMIT, "2");
-
- runAndVerifySuccess(2, true);
- }
-
- @Test
- public void testQueryElasticsearchOnTriggerWithServerErrorRetry() {
- QueryElasticsearchHttpTestProcessor processor = new QueryElasticsearchHttpTestProcessor();
- processor.setStatus(500, "Server error");
- runner = TestRunners.newTestRunner(processor); // simulate doc not found
- runner.setProperty(AbstractElasticsearchHttpProcessor.ES_URL, "http://127.0.0.1:9200");
- runner.setProperty(QueryElasticsearchHttp.INDEX, "doc");
- runner.setProperty(QueryElasticsearchHttp.TYPE, "status");
- runner.setProperty(QueryElasticsearchHttp.QUERY, "${doc_id}");
-
- runner.enqueue("".getBytes(), new HashMap<String, String>() {
- {
- put("identifier", "28039652140");
- }
- });
-
- runner.run(1, true, true);
-
- // This test generates a HTTP 500 "Server error"
- runner.assertAllFlowFilesTransferred(QueryElasticsearchHttp.REL_RETRY, 1);
- final MockFlowFile out = runner.getFlowFilesForRelationship(
- QueryElasticsearchHttp.REL_RETRY).get(0);
- assertNotNull(out);
- }
-
- @Test
- public void testQueryElasticsearchOnTriggerWithServerFail() {
- QueryElasticsearchHttpTestProcessor processor = new QueryElasticsearchHttpTestProcessor();
- processor.setStatus(100, "Should fail");
- runner = TestRunners.newTestRunner(processor); // simulate doc not found
- runner.setProperty(AbstractElasticsearchHttpProcessor.ES_URL, "http://127.0.0.1:9200");
- runner.setProperty(QueryElasticsearchHttp.INDEX, "doc");
- runner.setProperty(QueryElasticsearchHttp.TYPE, "status");
- runner.setProperty(QueryElasticsearchHttp.QUERY, "${doc_id}");
-
- runner.enqueue("".getBytes(), new HashMap<String, String>() {
- {
- put("identifier", "28039652140");
- }
- });
-
- runner.run(1, true, true);
-
- // This test generates a HTTP 100 "Should fail"
- runner.assertAllFlowFilesTransferred(QueryElasticsearchHttp.REL_FAILURE, 1);
- final MockFlowFile out = runner.getFlowFilesForRelationship(
- QueryElasticsearchHttp.REL_FAILURE).get(0);
- assertNotNull(out);
- }
-
- @Test
- public void testQueryElasticsearchOnTriggerWithIOException() {
- QueryElasticsearchHttpTestProcessor processor = new QueryElasticsearchHttpTestProcessor();
- processor.setExceptionToThrow(new IOException("Error reading from disk"));
- runner = TestRunners.newTestRunner(processor); // simulate doc not found
- runner.setProperty(AbstractElasticsearchHttpProcessor.ES_URL, "http://127.0.0.1:9200");
- runner.setProperty(QueryElasticsearchHttp.INDEX, "doc");
- runner.setProperty(QueryElasticsearchHttp.TYPE, "status");
- runner.setProperty(QueryElasticsearchHttp.QUERY, "${doc_id}");
-
- runner.enqueue("".getBytes(), new HashMap<String, String>() {
- {
- put("identifier", "28039652140");
- }
- });
-
- runner.run(1, true, true);
-
- // This test generates a HTTP 100 "Should fail"
- runner.assertAllFlowFilesTransferred(QueryElasticsearchHttp.REL_RETRY, 1);
- final MockFlowFile out = runner.getFlowFilesForRelationship(
- QueryElasticsearchHttp.REL_RETRY).get(0);
- assertNotNull(out);
- }
-
- @Test
- public void testQueryElasticsearchOnTriggerWithServerFailAfterSuccess() {
- QueryElasticsearchHttpTestProcessor processor = new QueryElasticsearchHttpTestProcessor();
- processor.setStatus(100, "Should fail", 2);
- runner = TestRunners.newTestRunner(processor); // simulate doc not found
- runner.setProperty(AbstractElasticsearchHttpProcessor.ES_URL, "http://127.0.0.1:9200");
- runner.setProperty(QueryElasticsearchHttp.INDEX, "doc");
- runner.setProperty(QueryElasticsearchHttp.TYPE, "status");
- runner.setProperty(QueryElasticsearchHttp.QUERY, "${doc_id}");
-
- runner.enqueue("".getBytes(), new HashMap<String, String>() {
- {
- put("identifier", "28039652140");
- }
- });
-
- runner.run(1, true, true);
-
- // This test generates a HTTP 100 "Should fail"
- runner.assertTransferCount(QueryElasticsearchHttp.REL_SUCCESS, 2);
- runner.assertTransferCount(QueryElasticsearchHttp.REL_FAILURE, 1);
- final MockFlowFile out = runner.getFlowFilesForRelationship(
- QueryElasticsearchHttp.REL_FAILURE).get(0);
- assertNotNull(out);
- }
-
- @Test
- public void testQueryElasticsearchOnTriggerWithServerFailNoIncomingFlowFile() {
- QueryElasticsearchHttpTestProcessor processor = new QueryElasticsearchHttpTestProcessor();
- processor.setStatus(100, "Should fail", 1);
- runner = TestRunners.newTestRunner(processor); // simulate doc not found
- runner.setProperty(AbstractElasticsearchHttpProcessor.ES_URL, "http://127.0.0.1:9200");
- runner.setProperty(QueryElasticsearchHttp.INDEX, "doc");
- runner.setProperty(QueryElasticsearchHttp.TYPE, "status");
- runner.setProperty(QueryElasticsearchHttp.QUERY, "${doc_id}");
-
- runner.setIncomingConnection(false);
- runner.run(1, true, true);
-
- // This test generates a HTTP 100 with no incoming flow file, so nothing should be transferred
- processor.getRelationships().forEach(relationship -> runner.assertTransferCount(relationship, 0));
- runner.assertTransferCount(QueryElasticsearchHttp.REL_FAILURE, 0);
- }
-
- @Test
- public void testSetupSecureClient() throws Exception {
- QueryElasticsearchHttpTestProcessor processor = new QueryElasticsearchHttpTestProcessor();
- runner = TestRunners.newTestRunner(processor);
- SSLContextService sslService = mock(SSLContextService.class);
- when(sslService.getIdentifier()).thenReturn("ssl-context");
- runner.addControllerService("ssl-context", sslService);
- runner.enableControllerService(sslService);
- runner.setProperty(QueryElasticsearchHttp.PROP_SSL_CONTEXT_SERVICE, "ssl-context");
- runner.setProperty(AbstractElasticsearchHttpProcessor.ES_URL, "http://127.0.0.1:9200");
- runner.setProperty(QueryElasticsearchHttp.INDEX, "doc");
- runner.removeProperty(QueryElasticsearchHttp.TYPE);
- runner.setProperty(QueryElasticsearchHttp.QUERY, "${doc_id}");
-
- // Allow time for the controller service to fully initialize
- Thread.sleep(500);
-
- runner.enqueue("".getBytes(), new HashMap<String, String>() {
- {
- put("doc_id", "28039652140");
- }
- });
- runner.run(1, true, true);
- }
-
- @Test
- public void testQueryElasticsearchOnTrigger_withQueryParameters() throws IOException {
- QueryElasticsearchHttpTestProcessor p = new QueryElasticsearchHttpTestProcessor();
- p.setExpectedParam("myparam=myvalue");
- runner = TestRunners.newTestRunner(p);
- runner.setProperty(AbstractElasticsearchHttpProcessor.ES_URL, "http://127.0.0.1:9200");
-
- runner.setProperty(QueryElasticsearchHttp.INDEX, "doc");
- runner.setProperty(QueryElasticsearchHttp.TYPE, "status");
- runner.setProperty(QueryElasticsearchHttp.QUERY, "source:Twitter");
- // Set dynamic property, to be added to the URL as a query parameter
- runner.setProperty("myparam", "myvalue");
- runAndVerifySuccess(true);
- }
-
- @Test
- public void testQueryElasticsearchOnTrigger_sourceIncludes() {
- QueryElasticsearchHttpTestProcessor p = new QueryElasticsearchHttpTestProcessor();
- p.setExpectedParam("_source=test");
- runner = TestRunners.newTestRunner(p);
- runner.setProperty(AbstractElasticsearchHttpProcessor.ES_URL, "http://127.0.0.1:9200");
-
- runner.setProperty(QueryElasticsearchHttp.INDEX, "doc");
- runner.setProperty(QueryElasticsearchHttp.TYPE, "status");
- runner.setProperty(QueryElasticsearchHttp.QUERY, "source:Twitter");
- runner.setProperty(QueryElasticsearchHttp.FIELDS, "test");
- runAndVerifySuccess(true);
- }
-
- /**
- * A Test class that extends the processor in order to inject/mock behavior
- */
- private static class QueryElasticsearchHttpTestProcessor extends QueryElasticsearchHttp {
- Exception exceptionToThrow = null;
- OkHttpClient client;
- int goodStatusCode = 200;
- String goodStatusMessage = "OK";
-
- int badStatusCode;
- String badStatusMessage;
- int runNumber;
-
- List<String> pages = Arrays.asList(getDoc("query-page1.json"), getDoc("query-page2.json"),
- getDoc("query-page3.json"));
-
- String expectedParam = null;
-
- public void setExceptionToThrow(Exception exceptionToThrow) {
- this.exceptionToThrow = exceptionToThrow;
- }
-
- /**
- * Sets the status code and message for the 1st query
- *
- * @param code
- * The status code to return
- * @param message
- * The status message
- */
- void setStatus(int code, String message) {
- this.setStatus(code, message, 1);
- }
-
- /**
- * Sets an query parameter (name=value) expected to be at the end of the URL for the query operation
- *
- * @param param
- * The parameter to expect
- */
- void setExpectedParam(String param) {
- expectedParam = param;
- }
-
- /**
- * Sets the status code and message for the runNumber-th query
- *
- * @param code
- * The status code to return
- * @param message
- * The status message
- * @param runNumber
- * The run number for which to set this status
- */
- void setStatus(int code, String message, int runNumber) {
- badStatusCode = code;
- badStatusMessage = message;
- this.runNumber = runNumber;
- }
-
- @Override
- protected void createElasticsearchClient(ProcessContext context) throws ProcessException {
- client = mock(OkHttpClient.class);
-
- OngoingStubbing<Call> stub = when(client.newCall(any(Request.class)));
-
- for (int i = 0; i < pages.size(); i++) {
- String page = pages.get(i);
- if (runNumber == i + 1) {
- stub = mockReturnDocument(stub, page, badStatusCode, badStatusMessage);
- } else {
- stub = mockReturnDocument(stub, page, goodStatusCode, goodStatusMessage);
- }
- }
- }
-
- private OngoingStubbing<Call> mockReturnDocument(OngoingStubbing<Call> stub,
- final String document, int statusCode, String statusMessage) {
- return stub.thenAnswer((Answer<Call>) invocationOnMock -> {
- Request realRequest = (Request) invocationOnMock.getArguments()[0];
- assertTrue((expectedParam == null) || (realRequest.url().toString().contains(expectedParam)));
- Response mockResponse = new Response.Builder()
- .request(realRequest)
- .protocol(Protocol.HTTP_1_1)
- .code(statusCode)
- .message(statusMessage)
- .body(ResponseBody.create(MediaType.parse("application/json"), document))
- .build();
- final Call call = mock(Call.class);
- if (exceptionToThrow != null) {
- when(call.execute()).thenThrow(exceptionToThrow);
- } else {
- when(call.execute()).thenReturn(mockResponse);
- }
- return call;
- });
- }
-
- @Override
- protected OkHttpClient getClient() {
- return client;
- }
- }
-
- private static String getDoc(String filename) {
- try {
- return IOUtils.toString(QueryElasticsearchHttp.class.getClassLoader().getResourceAsStream(filename), StandardCharsets.UTF_8);
- } catch (IOException e) {
- System.out.println("Error reading document " + filename);
- return "";
- }
- }
-}
diff --git a/nifi-nar-bundles/nifi-elasticsearch-bundle/nifi-elasticsearch-processors/src/test/java/org/apache/nifi/processors/elasticsearch/TestQueryElasticsearchHttpNoHits.java b/nifi-nar-bundles/nifi-elasticsearch-bundle/nifi-elasticsearch-processors/src/test/java/org/apache/nifi/processors/elasticsearch/TestQueryElasticsearchHttpNoHits.java
deleted file mode 100644
index 4f8d4d0a36..0000000000
--- a/nifi-nar-bundles/nifi-elasticsearch-bundle/nifi-elasticsearch-processors/src/test/java/org/apache/nifi/processors/elasticsearch/TestQueryElasticsearchHttpNoHits.java
+++ /dev/null
@@ -1,412 +0,0 @@
-/*
- * 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 okhttp3.Call;
-import okhttp3.MediaType;
-import okhttp3.OkHttpClient;
-import okhttp3.Protocol;
-import okhttp3.Request;
-import okhttp3.Response;
-import okhttp3.ResponseBody;
-import org.apache.commons.io.IOUtils;
-import org.apache.nifi.processor.ProcessContext;
-import org.apache.nifi.processor.exception.ProcessException;
-import org.apache.nifi.util.MockFlowFile;
-import org.apache.nifi.util.TestRunner;
-import org.apache.nifi.util.TestRunners;
-import org.junit.jupiter.api.AfterEach;
-import org.junit.jupiter.api.Test;
-import org.mockito.stubbing.Answer;
-import org.mockito.stubbing.OngoingStubbing;
-
-import java.io.IOException;
-import java.nio.charset.StandardCharsets;
-import java.util.Arrays;
-import java.util.HashMap;
-import java.util.List;
-
-import static org.junit.jupiter.api.Assertions.assertNotNull;
-import static org.junit.jupiter.api.Assertions.assertTrue;
-import static org.mockito.ArgumentMatchers.any;
-import static org.mockito.Mockito.mock;
-import static org.mockito.Mockito.when;
-
-public class TestQueryElasticsearchHttpNoHits {
-
- private TestRunner runner;
-
- @AfterEach
- public void teardown() {
- runner = null;
- }
-
-
- @Test
- public void testQueryElasticsearchOnTrigger_NoHits_NoHits() throws IOException {
- runner = TestRunners.newTestRunner(new QueryElasticsearchHttpTestProcessor());
- runner.setValidateExpressionUsage(true);
- runner.setProperty(AbstractElasticsearchHttpProcessor.ES_URL, "http://127.0.0.1:9200");
-
- runner.setProperty(QueryElasticsearchHttp.INDEX, "doc");
- runner.assertNotValid();
- runner.setProperty(QueryElasticsearchHttp.TYPE, "status");
- runner.assertNotValid();
- runner.setProperty(QueryElasticsearchHttp.QUERY,
- "source:Twitter AND identifier:\"${identifier}\"");
- runner.assertValid();
- runner.setProperty(QueryElasticsearchHttp.PAGE_SIZE, "2");
- runner.assertValid();
- runner.setProperty(QueryElasticsearchHttp.ROUTING_QUERY_INFO_STRATEGY, QueryElasticsearchHttp.QueryInfoRouteStrategy.NOHIT.name());
- runner.assertValid();
-
- runner.setIncomingConnection(false);
- runAndVerify(0,1,0,true);
- }
-
- @Test
- public void testQueryElasticsearchOnTrigger_NoHits_Never() throws IOException {
- runner = TestRunners.newTestRunner(new QueryElasticsearchHttpTestProcessor());
- runner.setValidateExpressionUsage(true);
- runner.setProperty(AbstractElasticsearchHttpProcessor.ES_URL, "http://127.0.0.1:9200");
-
- runner.setProperty(QueryElasticsearchHttp.INDEX, "doc");
- runner.assertNotValid();
- runner.setProperty(QueryElasticsearchHttp.TYPE, "status");
- runner.assertNotValid();
- runner.setProperty(QueryElasticsearchHttp.QUERY,
- "source:Twitter AND identifier:\"${identifier}\"");
- runner.assertValid();
- runner.setProperty(QueryElasticsearchHttp.PAGE_SIZE, "2");
- runner.assertValid();
- runner.setProperty(QueryElasticsearchHttp.ROUTING_QUERY_INFO_STRATEGY, QueryElasticsearchHttp.QueryInfoRouteStrategy.NEVER.name());
- runner.assertValid();
-
- runner.setIncomingConnection(false);
- runAndVerify(0,0,0,true);
- }
-
- @Test
- public void testQueryElasticsearchOnTrigger_NoHits_Always() throws IOException {
- runner = TestRunners.newTestRunner(new QueryElasticsearchHttpTestProcessor());
- runner.setValidateExpressionUsage(true);
- runner.setProperty(AbstractElasticsearchHttpProcessor.ES_URL, "http://127.0.0.1:9200");
-
- runner.setProperty(QueryElasticsearchHttp.INDEX, "doc");
- runner.assertNotValid();
- runner.setProperty(QueryElasticsearchHttp.TYPE, "status");
- runner.assertNotValid();
- runner.setProperty(QueryElasticsearchHttp.QUERY,
- "source:Twitter AND identifier:\"${identifier}\"");
- runner.assertValid();
- runner.setProperty(QueryElasticsearchHttp.PAGE_SIZE, "2");
- runner.assertValid();
- runner.setProperty(QueryElasticsearchHttp.ROUTING_QUERY_INFO_STRATEGY, QueryElasticsearchHttp.QueryInfoRouteStrategy.ALWAYS.name());
- runner.assertValid();
-
- runner.setIncomingConnection(false);
- runAndVerify(0,1,0,true);
- }
-
- @Test
- public void testQueryElasticsearchOnTrigger_Hits_NoHits() throws IOException {
- runner = TestRunners.newTestRunner(new QueryElasticsearchHttpTestProcessor(true));
- runner.setValidateExpressionUsage(true);
- runner.setProperty(AbstractElasticsearchHttpProcessor.ES_URL, "http://127.0.0.1:9200");
-
- runner.setProperty(QueryElasticsearchHttp.INDEX, "doc");
- runner.assertNotValid();
- runner.setProperty(QueryElasticsearchHttp.TYPE, "status");
- runner.assertNotValid();
- runner.setProperty(QueryElasticsearchHttp.QUERY,
- "source:Twitter AND identifier:\"${identifier}\"");
- runner.assertValid();
- runner.setProperty(QueryElasticsearchHttp.PAGE_SIZE, "2");
- runner.assertValid();
- runner.setProperty(QueryElasticsearchHttp.ROUTING_QUERY_INFO_STRATEGY, QueryElasticsearchHttp.QueryInfoRouteStrategy.NOHIT.name());
- runner.assertValid();
-
- runner.setIncomingConnection(false);
- runAndVerify(3,0,0,true);
- }
-
- @Test
- public void testQueryElasticsearchOnTrigger_Hits_Never() throws IOException {
- runner = TestRunners.newTestRunner(new QueryElasticsearchHttpTestProcessor(true));
- runner.setValidateExpressionUsage(true);
- runner.setProperty(AbstractElasticsearchHttpProcessor.ES_URL, "http://127.0.0.1:9200");
-
- runner.setProperty(QueryElasticsearchHttp.INDEX, "doc");
- runner.assertNotValid();
- runner.setProperty(QueryElasticsearchHttp.TYPE, "status");
- runner.assertNotValid();
- runner.setProperty(QueryElasticsearchHttp.QUERY,
- "source:Twitter AND identifier:\"${identifier}\"");
- runner.assertValid();
- runner.setProperty(QueryElasticsearchHttp.PAGE_SIZE, "2");
- runner.assertValid();
- runner.setProperty(QueryElasticsearchHttp.ROUTING_QUERY_INFO_STRATEGY, QueryElasticsearchHttp.QueryInfoRouteStrategy.NEVER.name());
- runner.assertValid();
-
- runner.setIncomingConnection(false);
- runAndVerify(3,0,0,true);
- }
-
- @Test
- public void testQueryElasticsearchOnTrigger_Hits_Always() throws IOException {
- runner = TestRunners.newTestRunner(new QueryElasticsearchHttpTestProcessor(true));
- runner.setValidateExpressionUsage(true);
- runner.setProperty(AbstractElasticsearchHttpProcessor.ES_URL, "http://127.0.0.1:9200");
-
- runner.setProperty(QueryElasticsearchHttp.INDEX, "doc");
- runner.assertNotValid();
- runner.setProperty(QueryElasticsearchHttp.TYPE, "status");
- runner.assertNotValid();
- runner.setProperty(QueryElasticsearchHttp.QUERY,
- "source:Twitter AND identifier:\"${identifier}\"");
- runner.assertValid();
- runner.setProperty(QueryElasticsearchHttp.PAGE_SIZE, "2");
- runner.assertValid();
- runner.setProperty(QueryElasticsearchHttp.ROUTING_QUERY_INFO_STRATEGY, QueryElasticsearchHttp.QueryInfoRouteStrategy.ALWAYS.name());
- runner.assertValid();
-
- runner.setIncomingConnection(false);
- runAndVerify(3,3,2,true);
- }
-
- @Test
- public void testQueryElasticsearchOnTrigger_Hits_AppendAsAttributes() throws IOException {
- runner = TestRunners.newTestRunner(new QueryElasticsearchHttpTestProcessor(false));
- runner.setValidateExpressionUsage(true);
- runner.setProperty(AbstractElasticsearchHttpProcessor.ES_URL, "http://127.0.0.1:9200");
-
- runner.setProperty(QueryElasticsearchHttp.INDEX, "doc");
- runner.assertNotValid();
- runner.setProperty(QueryElasticsearchHttp.TYPE, "status");
- runner.assertNotValid();
- runner.setProperty(QueryElasticsearchHttp.QUERY,
- "source:Twitter AND identifier:\"${identifier}\"");
- runner.assertValid();
- runner.setProperty(QueryElasticsearchHttp.PAGE_SIZE, "2");
- runner.assertValid();
- runner.setProperty(QueryElasticsearchHttp.ROUTING_QUERY_INFO_STRATEGY, QueryElasticsearchHttp.QueryInfoRouteStrategy.APPEND_AS_ATTRIBUTES.name());
- runner.assertValid();
-
- runner.setIncomingConnection(true);
- runAndVerify(1,0,0,false, false);
- }
-
- @Test
- public void testQueryElasticsearchOnTrigger_Hits_AppendAsAttributes_noHits() throws IOException {
- runner = TestRunners.newTestRunner(new QueryElasticsearchHttpTestProcessor(true));
- runner.setValidateExpressionUsage(true);
- runner.setProperty(AbstractElasticsearchHttpProcessor.ES_URL, "http://127.0.0.1:9200");
-
- runner.setProperty(QueryElasticsearchHttp.INDEX, "doc");
- runner.assertNotValid();
- runner.setProperty(QueryElasticsearchHttp.TYPE, "status");
- runner.assertNotValid();
- runner.setProperty(QueryElasticsearchHttp.QUERY,
- "source:Twitter AND identifier:\"${identifier}\"");
- runner.assertValid();
- runner.setProperty(QueryElasticsearchHttp.PAGE_SIZE, "2");
- runner.assertValid();
- runner.setProperty(QueryElasticsearchHttp.ROUTING_QUERY_INFO_STRATEGY, QueryElasticsearchHttp.QueryInfoRouteStrategy.APPEND_AS_ATTRIBUTES.name());
- runner.assertValid();
-
- runner.setIncomingConnection(false);
- runAndVerify(3,3,2,true, false);
- }
-
- private void runAndVerify(int expectedResults,int expectedQueryInfoResults,int expectedHits, boolean targetIsContent) {
- runAndVerify(expectedResults, expectedQueryInfoResults, expectedHits, targetIsContent, true);
- }
-
- private void runAndVerify(int expectedResults,int expectedQueryInfoResults,int expectedHits, boolean targetIsContent,
- boolean expectHitCountOnQueryInfo) {
- runner.enqueue("blah".getBytes(), new HashMap<String, String>() {
- {
- put("identifier", "28039652140");
- }
- });
-
- // Running once should page through the no hit doc
- runner.run(1, true, true);
-
- if (expectHitCountOnQueryInfo) {
- runner.assertTransferCount(QueryElasticsearchHttp.REL_QUERY_INFO, expectedQueryInfoResults);
- if (expectedQueryInfoResults > 0) {
- final MockFlowFile out = runner.getFlowFilesForRelationship(QueryElasticsearchHttp.REL_QUERY_INFO).get(0);
- assertNotNull(out);
- if (targetIsContent) {
- if (expectHitCountOnQueryInfo) {
- out.assertAttributeEquals("es.query.hitcount", String.valueOf(expectedHits));
- }
- assertTrue(out.getAttribute("es.query.url").startsWith("http://127.0.0.1:9200/doc/status/_search?q=source%3ATwitter%20AND%20identifier%3A%22%22&size=2"));
- }
- }
- }
-
- runner.assertTransferCount(QueryElasticsearchHttp.REL_SUCCESS, expectedResults);
- if (expectedResults > 0) {
- final MockFlowFile out = runner.getFlowFilesForRelationship(QueryElasticsearchHttp.REL_SUCCESS).get(0);
- assertNotNull(out);
- if (!expectHitCountOnQueryInfo) {
- out.assertAttributeEquals("es.query.hitcount", String.valueOf(expectedHits));
- }
- if (targetIsContent) {
- out.assertAttributeEquals("filename", "abc-97b-ASVsZu_" + "vShwtGCJpGOObmuSqUJRUC3L_-SEND-S3");
- }
- }
- }
-
- // By default, 3 files should go to Success
- private void runAndVerify(boolean targetIsContent) {
- runAndVerify(0,1,0, targetIsContent);
- }
-
-
-
- /**
- * A Test class that extends the processor in order to inject/mock behavior
- */
- private static class QueryElasticsearchHttpTestProcessor extends QueryElasticsearchHttp {
- Exception exceptionToThrow = null;
- OkHttpClient client;
- int goodStatusCode = 200;
- String goodStatusMessage = "OK";
-
- int badStatusCode;
- String badStatusMessage;
- int runNumber;
-
- boolean useHitPages;
-
- // query-page3 has no hits
- List<String> noHitPages = Arrays.asList(getDoc("query-page3.json"));
- List<String> hitPages = Arrays.asList(getDoc("query-page1.json"), getDoc("query-page2.json"),
- getDoc("query-page3.json"));
-
- String expectedParam = null;
-
- public QueryElasticsearchHttpTestProcessor() {
- this(false);
- }
- public QueryElasticsearchHttpTestProcessor(boolean useHitPages) {
- this.useHitPages = useHitPages;
- }
-
- public void setExceptionToThrow(Exception exceptionToThrow) {
- this.exceptionToThrow = exceptionToThrow;
- }
-
- /**
- * Sets the status code and message for the 1st query
- *
- * @param code
- * The status code to return
- * @param message
- * The status message
- */
- void setStatus(int code, String message) {
- this.setStatus(code, message, 1);
- }
-
- /**
- * Sets an query parameter (name=value) expected to be at the end of the URL for the query operation
- *
- * @param param
- * The parameter to expect
- */
- void setExpectedParam(String param) {
- expectedParam = param;
- }
-
- /**
- * Sets the status code and message for the runNumber-th query
- *
- * @param code
- * The status code to return
- * @param message
- * The status message
- * @param runNumber
- * The run number for which to set this status
- */
- void setStatus(int code, String message, int runNumber) {
- badStatusCode = code;
- badStatusMessage = message;
- this.runNumber = runNumber;
- }
-
- @Override
- protected void createElasticsearchClient(ProcessContext context) throws ProcessException {
- client = mock(OkHttpClient.class);
-
- OngoingStubbing<Call> stub = when(client.newCall(any(Request.class)));
- List<String> pages;
- if(useHitPages) {
- pages = hitPages;
- } else {
- pages = noHitPages;
- }
-
- for (int i = 0; i < pages.size(); i++) {
- String page = pages.get(i);
- if (runNumber == i + 1) {
- stub = mockReturnDocument(stub, page, badStatusCode, badStatusMessage);
- } else {
- stub = mockReturnDocument(stub, page, goodStatusCode, goodStatusMessage);
- }
- }
- }
-
- private OngoingStubbing<Call> mockReturnDocument(OngoingStubbing<Call> stub,
- final String document, int statusCode, String statusMessage) {
- return stub.thenAnswer((Answer<Call>) invocationOnMock -> {
- Request realRequest = (Request) invocationOnMock.getArguments()[0];
- assertTrue((expectedParam == null) || (realRequest.url().toString().endsWith(expectedParam)));
- Response mockResponse = new Response.Builder()
- .request(realRequest)
- .protocol(Protocol.HTTP_1_1)
- .code(statusCode)
- .message(statusMessage)
- .body(ResponseBody.create(MediaType.parse("application/json"), document))
- .build();
- final Call call = mock(Call.class);
- if (exceptionToThrow != null) {
- when(call.execute()).thenThrow(exceptionToThrow);
- } else {
- when(call.execute()).thenReturn(mockResponse);
- }
- return call;
- });
- }
-
- protected OkHttpClient getClient() {
- return client;
- }
- }
-
- private static String getDoc(String filename) {
- try {
- return IOUtils.toString(QueryElasticsearchHttp.class.getClassLoader().getResourceAsStream(filename), StandardCharsets.UTF_8);
- } catch (IOException e) {
- System.out.println("Error reading document " + filename);
- return "";
- }
- }
-}
diff --git a/nifi-nar-bundles/nifi-elasticsearch-bundle/nifi-elasticsearch-processors/src/test/java/org/apache/nifi/processors/elasticsearch/TestScrollElasticsearchHttp.java b/nifi-nar-bundles/nifi-elasticsearch-bundle/nifi-elasticsearch-processors/src/test/java/org/apache/nifi/processors/elasticsearch/TestScrollElasticsearchHttp.java
deleted file mode 100644
index c2ab44679f..0000000000
--- a/nifi-nar-bundles/nifi-elasticsearch-bundle/nifi-elasticsearch-processors/src/test/java/org/apache/nifi/processors/elasticsearch/TestScrollElasticsearchHttp.java
+++ /dev/null
@@ -1,463 +0,0 @@
-/*
- * 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 okhttp3.Call;
-import okhttp3.MediaType;
-import okhttp3.OkHttpClient;
-import okhttp3.Protocol;
-import okhttp3.Request;
-import okhttp3.Response;
-import okhttp3.ResponseBody;
-import org.apache.commons.io.IOUtils;
-import org.apache.commons.lang3.StringUtils;
-import org.apache.nifi.processor.ProcessContext;
-import org.apache.nifi.processor.exception.ProcessException;
-import org.apache.nifi.ssl.SSLContextService;
-import org.apache.nifi.util.MockFlowFile;
-import org.apache.nifi.util.TestRunner;
-import org.apache.nifi.util.TestRunners;
-import org.junit.jupiter.api.AfterEach;
-import org.junit.jupiter.api.Test;
-import org.mockito.stubbing.Answer;
-import org.mockito.stubbing.OngoingStubbing;
-
-import java.io.IOException;
-import java.nio.charset.StandardCharsets;
-import java.util.Arrays;
-import java.util.HashMap;
-import java.util.List;
-
-import static org.junit.jupiter.api.Assertions.assertEquals;
-import static org.junit.jupiter.api.Assertions.assertNotNull;
-import static org.junit.jupiter.api.Assertions.assertTrue;
-import static org.mockito.ArgumentMatchers.any;
-import static org.mockito.Mockito.mock;
-import static org.mockito.Mockito.when;
-
-public class TestScrollElasticsearchHttp {
-
- private TestRunner runner;
-
- @AfterEach
- public void teardown() {
- runner = null;
- }
-
- @Test
- public void testScrollElasticsearchOnTrigger_withNoInput() {
- runner = TestRunners.newTestRunner(new ScrollElasticsearchHttpTestProcessor());
- runner.setProperty(AbstractElasticsearchHttpProcessor.ES_URL, "http://127.0.0.1:9200");
-
- runner.setProperty(ScrollElasticsearchHttp.INDEX, "doc");
- runner.assertNotValid();
- runner.setProperty(ScrollElasticsearchHttp.TYPE, "status");
- runner.assertNotValid();
- runner.setProperty(ScrollElasticsearchHttp.QUERY,
- "source:WZ AND identifier:\"${identifier}\"");
- runner.assertValid();
- runner.setProperty(ScrollElasticsearchHttp.PAGE_SIZE, "2");
- runner.assertValid();
-
- runner.setIncomingConnection(false);
- runAndVerifySuccess();
- }
-
- @Test
- public void testScrollElasticsearchOnTrigger_sourceIncludes() throws IOException {
- ScrollElasticsearchHttpTestProcessor p = new ScrollElasticsearchHttpTestProcessor();
- p.setExpectedParam("_source=test");
- runner = TestRunners.newTestRunner(p);
- runner.setProperty(AbstractElasticsearchHttpProcessor.ES_URL, "http://127.0.0.1:9200");
-
- runner.setProperty(ScrollElasticsearchHttp.INDEX, "doc");
- runner.setProperty(ScrollElasticsearchHttp.TYPE, "status");
- runner.setProperty(ScrollElasticsearchHttp.QUERY, "source:Twitter");
- runner.setProperty(ScrollElasticsearchHttp.FIELDS, "test");
- runAndVerifySuccess();
- }
-
- @Test
- public void testScrollElasticsearchOnTrigger_withNoInput_EL() throws IOException {
- runner = TestRunners.newTestRunner(new ScrollElasticsearchHttpTestProcessor());
- runner.setProperty(AbstractElasticsearchHttpProcessor.ES_URL, "${es.url}");
-
- runner.setProperty(ScrollElasticsearchHttp.INDEX, "doc");
- runner.assertNotValid();
- runner.setProperty(ScrollElasticsearchHttp.TYPE, "status");
- runner.assertNotValid();
- runner.setProperty(ScrollElasticsearchHttp.QUERY,
- "source:WZ AND identifier:\"${identifier}\"");
- runner.assertValid();
- runner.setProperty(ScrollElasticsearchHttp.PAGE_SIZE, "2");
- runner.assertValid();
- runner.setProperty(AbstractElasticsearchHttpProcessor.CONNECT_TIMEOUT, "${connect.timeout}");
- runner.assertValid();
-
- runner.setVariable("es.url", "http://127.0.0.1:9200");
-
- runner.setIncomingConnection(false);
- runAndVerifySuccess();
- }
-
- private void runAndVerifySuccess() {
- runner.enqueue("".getBytes(), new HashMap<String, String>() {
- {
- put("identifier", "28039652140");
- }
- });
-
- // Must run once for each of the 3 pages
- runner.run(3, true, true);
-
- runner.assertAllFlowFilesTransferred(ScrollElasticsearchHttp.REL_SUCCESS, 2);
- final MockFlowFile out = runner.getFlowFilesForRelationship(
- ScrollElasticsearchHttp.REL_SUCCESS).get(0);
- assertNotNull(out);
-
- int numHits = runner.getFlowFilesForRelationship(
- ScrollElasticsearchHttp.REL_SUCCESS).stream().map(ff -> {
- String page = new String(ff.toByteArray());
- return StringUtils.countMatches(page, "{\"timestamp\"");
- })
- .reduce((a, b) -> a + b).get();
- assertEquals(3, numHits);
- }
-
- @Test
- public void testScrollElasticsearchOnTriggerWithFields() throws IOException {
- runner = TestRunners.newTestRunner(new ScrollElasticsearchHttpTestProcessor());
- runner.setProperty(AbstractElasticsearchHttpProcessor.ES_URL, "http://127.0.0.1:9200");
-
- runner.setProperty(ScrollElasticsearchHttp.INDEX, "doc");
- runner.assertNotValid();
- runner.setProperty(ScrollElasticsearchHttp.QUERY, "${doc_id}");
- runner.assertValid();
- runner.removeProperty(ScrollElasticsearchHttp.TYPE);
- runner.assertValid();
- runner.setProperty(ScrollElasticsearchHttp.TYPE, "status");
- runner.assertValid();
- runner.setProperty(ScrollElasticsearchHttp.TYPE, "${type}");
- runner.assertValid();
- runner.setProperty(ScrollElasticsearchHttp.TYPE, "");
- runner.assertNotValid();
- runner.setProperty(ScrollElasticsearchHttp.TYPE, "_doc");
- runner.assertValid();
- runner.setProperty(ScrollElasticsearchHttp.FIELDS, "id,, userinfo.location");
- runner.assertValid();
- runner.setProperty(ScrollElasticsearchHttp.SORT, "timestamp:asc,identifier:desc");
- runner.assertValid();
- runner.setIncomingConnection(false);
-
- runAndVerifySuccess();
- }
-
- @Test
- public void testScrollElasticsearchOnTriggerWithServerFail() throws IOException {
- ScrollElasticsearchHttpTestProcessor processor = new ScrollElasticsearchHttpTestProcessor();
- processor.setStatus(100, "Should fail");
- runner = TestRunners.newTestRunner(processor); // simulate doc not found
- runner.setProperty(AbstractElasticsearchHttpProcessor.ES_URL, "http://127.0.0.1:9200");
- runner.setProperty(ScrollElasticsearchHttp.INDEX, "doc");
- runner.setProperty(ScrollElasticsearchHttp.TYPE, "status");
- runner.setProperty(ScrollElasticsearchHttp.QUERY, "${doc_id}");
- runner.setIncomingConnection(false);
-
- runner.enqueue("".getBytes(), new HashMap<String, String>() {
- {
- put("identifier", "28039652140");
- }
- });
-
- runner.run(1, true, true);
-
- // This test generates a HTTP 100 "Should fail"
- runner.assertTransferCount(ScrollElasticsearchHttp.REL_FAILURE, 0);
- runner.assertTransferCount(ScrollElasticsearchHttp.REL_SUCCESS, 0);
- }
-
- @Test
- public void testScrollElasticsearchOnTriggerWithServerRetry() throws IOException {
- ScrollElasticsearchHttpTestProcessor processor = new ScrollElasticsearchHttpTestProcessor();
- processor.setStatus(500, "Internal error");
- runner = TestRunners.newTestRunner(processor); // simulate doc not found
- runner.setProperty(AbstractElasticsearchHttpProcessor.ES_URL, "http://127.0.0.1:9200");
- runner.setProperty(ScrollElasticsearchHttp.INDEX, "doc");
- runner.setProperty(ScrollElasticsearchHttp.TYPE, "status");
- runner.setProperty(ScrollElasticsearchHttp.QUERY, "${doc_id}");
- runner.setIncomingConnection(false);
-
- runner.enqueue("".getBytes(), new HashMap<String, String>() {
- {
- put("identifier", "28039652140");
- }
- });
-
- runner.run(1, true, true);
-
- // This test generates a HTTP 500 "Internal error"
- runner.assertTransferCount(ScrollElasticsearchHttp.REL_FAILURE, 0);
- runner.assertTransferCount(ScrollElasticsearchHttp.REL_SUCCESS, 0);
- }
-
- @Test
- public void testScrollElasticsearchOnTriggerWithServerFailAfterSuccess() throws IOException {
- ScrollElasticsearchHttpTestProcessor processor = new ScrollElasticsearchHttpTestProcessor();
- processor.setStatus(100, "Should fail", 2);
- runner = TestRunners.newTestRunner(processor); // simulate doc not found
- runner.setProperty(AbstractElasticsearchHttpProcessor.ES_URL, "http://127.0.0.1:9200");
- runner.setProperty(ScrollElasticsearchHttp.INDEX, "doc");
- runner.setProperty(ScrollElasticsearchHttp.TYPE, "status");
- runner.setProperty(ScrollElasticsearchHttp.QUERY, "${doc_id}");
-
- runner.enqueue("".getBytes(), new HashMap<String, String>() {
- {
- put("identifier", "28039652140");
- }
- });
- runner.setIncomingConnection(false);
-
- runner.run(1, true, true);
-
- // This test generates a HTTP 100 "Should fail"
- runner.assertTransferCount(ScrollElasticsearchHttp.REL_SUCCESS, 1);
- runner.assertTransferCount(ScrollElasticsearchHttp.REL_FAILURE, 0);
- }
-
- @Test
- public void testScrollElasticsearchOnTriggerWithServerFailNoIncomingFlowFile() throws IOException {
- ScrollElasticsearchHttpTestProcessor processor = new ScrollElasticsearchHttpTestProcessor();
- processor.setStatus(100, "Should fail", 1);
- runner = TestRunners.newTestRunner(processor); // simulate doc not found
- runner.setProperty(AbstractElasticsearchHttpProcessor.ES_URL, "http://127.0.0.1:9200");
- runner.setProperty(ScrollElasticsearchHttp.INDEX, "doc");
- runner.setProperty(ScrollElasticsearchHttp.TYPE, "status");
- runner.setProperty(ScrollElasticsearchHttp.QUERY, "${doc_id}");
-
- runner.setIncomingConnection(false);
- runner.run(1, true, true);
-
- // This test generates a HTTP 100 with no incoming flow file, so nothing should be transferred
- processor.getRelationships().forEach(relationship -> runner.assertTransferCount(relationship, 0));
- runner.assertTransferCount(ScrollElasticsearchHttp.REL_FAILURE, 0);
- }
-
- @Test
- public void testSetupSecureClient() throws Exception {
- ScrollElasticsearchHttpTestProcessor processor = new ScrollElasticsearchHttpTestProcessor();
- runner = TestRunners.newTestRunner(processor);
- SSLContextService sslService = mock(SSLContextService.class);
- when(sslService.getIdentifier()).thenReturn("ssl-context");
- runner.addControllerService("ssl-context", sslService);
- runner.enableControllerService(sslService);
- runner.setProperty(ScrollElasticsearchHttp.PROP_SSL_CONTEXT_SERVICE, "ssl-context");
- runner.setProperty(AbstractElasticsearchHttpProcessor.ES_URL, "http://127.0.0.1:9200");
- runner.setProperty(ScrollElasticsearchHttp.INDEX, "doc");
- runner.removeProperty(ScrollElasticsearchHttp.TYPE);
- runner.setProperty(ScrollElasticsearchHttp.QUERY, "${doc_id}");
- runner.setIncomingConnection(false);
-
- // Allow time for the controller service to fully initialize
- Thread.sleep(500);
-
- runner.enqueue("".getBytes(), new HashMap<String, String>() {
- {
- put("doc_id", "28039652140");
- }
- });
- runner.run(1, true, true);
-
- }
-
- @Test
- public void testScrollElasticsearchOnTriggerWithIOException() throws IOException {
- ScrollElasticsearchHttpTestProcessor processor = new ScrollElasticsearchHttpTestProcessor();
- processor.setExceptionToThrow(new IOException("Error reading from disk"));
- runner = TestRunners.newTestRunner(processor); // simulate doc not found
- runner.setProperty(AbstractElasticsearchHttpProcessor.ES_URL, "http://127.0.0.1:9200");
- runner.setProperty(ScrollElasticsearchHttp.INDEX, "doc");
- runner.setProperty(ScrollElasticsearchHttp.TYPE, "status");
- runner.setProperty(ScrollElasticsearchHttp.QUERY, "${doc_id}");
-
- runner.enqueue("".getBytes(), new HashMap<String, String>() {
- {
- put("identifier", "28039652140");
- }
- });
-
- runner.run(1, true, true);
-
- // This test generates a HTTP 100 "Should fail"
- runner.assertTransferCount(ScrollElasticsearchHttp.REL_SUCCESS, 0);
- runner.assertTransferCount(ScrollElasticsearchHttp.REL_FAILURE, 0);
- }
-
- @Test
- public void testScrollElasticsearchOnTriggerWithOtherException() throws IOException {
- ScrollElasticsearchHttpTestProcessor processor = new ScrollElasticsearchHttpTestProcessor();
- processor.setExceptionToThrow(new IllegalArgumentException("Error reading from disk"));
- runner = TestRunners.newTestRunner(processor); // simulate doc not found
- runner.setProperty(AbstractElasticsearchHttpProcessor.ES_URL, "http://127.0.0.1:9200");
- runner.setProperty(ScrollElasticsearchHttp.INDEX, "doc");
- runner.setProperty(ScrollElasticsearchHttp.TYPE, "status");
- runner.setProperty(ScrollElasticsearchHttp.QUERY, "${doc_id}");
-
- runner.enqueue("".getBytes(), new HashMap<String, String>() {
- {
- put("identifier", "28039652140");
- }
- });
-
- runner.run(1, true, true);
-
- // This test generates a HTTP 100 "Should fail"
- runner.assertTransferCount(ScrollElasticsearchHttp.REL_SUCCESS, 0);
- runner.assertTransferCount(ScrollElasticsearchHttp.REL_FAILURE, 1);
- }
-
- @Test
- public void testScrollElasticsearchOnTrigger_withQueryParameter() throws IOException {
- ScrollElasticsearchHttpTestProcessor p = new ScrollElasticsearchHttpTestProcessor();
- p.setExpectedParam("myparam=myvalue");
- runner = TestRunners.newTestRunner(p);
- runner.setProperty(AbstractElasticsearchHttpProcessor.ES_URL, "http://127.0.0.1:9200");
-
- runner.setProperty(ScrollElasticsearchHttp.INDEX, "doc");
- runner.setProperty(ScrollElasticsearchHttp.TYPE, "status");
- runner.setProperty(ScrollElasticsearchHttp.QUERY, "source:WZ");
- runner.setProperty(ScrollElasticsearchHttp.PAGE_SIZE, "2");
- // Set dynamic property, to be added to the URL as a query parameter
- runner.setProperty("myparam", "myvalue");
- runner.setIncomingConnection(false);
- runAndVerifySuccess();
- }
-
- /**
- * A Test class that extends the processor in order to inject/mock behavior
- */
- private static class ScrollElasticsearchHttpTestProcessor extends ScrollElasticsearchHttp {
- Exception exceptionToThrow = null;
- OkHttpClient client;
- int goodStatusCode = 200;
- String goodStatusMessage = "OK";
-
- int badStatusCode;
- String badStatusMessage;
- int runNumber;
-
- List<String> pages = Arrays.asList(getDoc("scroll-page1.json"),
- getDoc("scroll-page2.json"), getDoc("scroll-page3.json"));
-
- String expectedParam = null;
-
- public void setExceptionToThrow(Exception exceptionToThrow) {
- this.exceptionToThrow = exceptionToThrow;
- }
-
- /**
- * Sets the status code and message for the 1st query
- *
- * @param code
- * The status code to return
- * @param message
- * The status message
- */
- void setStatus(int code, String message) {
- this.setStatus(code, message, 1);
- }
-
- /**
- * Sets the status code and message for the runNumber-th query
- *
- * @param code
- * The status code to return
- * @param message
- * The status message
- * @param runNumber
- * The run number for which to set this status
- */
- void setStatus(int code, String message, int runNumber) {
- badStatusCode = code;
- badStatusMessage = message;
- this.runNumber = runNumber;
- }
-
- /**
- * Sets an query parameter (name=value) expected to be at the end of the URL for the query operation
- *
- * @param param
- * The parameter to expect
- */
- void setExpectedParam(String param) {
- expectedParam = param;
- }
-
- @Override
- protected void createElasticsearchClient(ProcessContext context) throws ProcessException {
- client = mock(OkHttpClient.class);
-
- OngoingStubbing<Call> stub = when(client.newCall(any(Request.class)));
-
- for (int i = 0; i < pages.size(); i++) {
- String page = pages.get(i);
- if (runNumber == i + 1) {
- stub = mockReturnDocument(stub, page, badStatusCode, badStatusMessage);
- } else {
- stub = mockReturnDocument(stub, page, goodStatusCode, goodStatusMessage);
- }
- }
- }
-
- private OngoingStubbing<Call> mockReturnDocument(OngoingStubbing<Call> stub,
- final String document, int statusCode, String statusMessage) {
- return stub.thenAnswer((Answer<Call>) invocationOnMock -> {
- Request realRequest = (Request) invocationOnMock.getArguments()[0];
- if (realRequest.method().equals("GET")) {
- assertTrue((expectedParam == null) || (realRequest.url().toString().contains(expectedParam)));
- }
- Response mockResponse = new Response.Builder()
- .request(realRequest)
- .protocol(Protocol.HTTP_1_1)
- .code(statusCode)
- .message(statusMessage)
- .body(ResponseBody.create(MediaType.parse("application/json"), document))
- .build();
- final Call call = mock(Call.class);
- if (exceptionToThrow != null) {
- when(call.execute()).thenThrow(exceptionToThrow);
- } else {
- when(call.execute()).thenReturn(mockResponse);
- }
- return call;
- });
- }
-
- @Override
- protected OkHttpClient getClient() {
- return client;
- }
- }
-
- private static String getDoc(String filename) {
- try {
- return IOUtils.toString(ScrollElasticsearchHttp.class.getClassLoader().getResourceAsStream(filename), StandardCharsets.UTF_8);
- } catch (IOException e) {
- System.out.println("Error reading document " + filename);
- return "";
- }
- }
-}
diff --git a/nifi-nar-bundles/nifi-elasticsearch-bundle/nifi-elasticsearch-processors/src/test/resources/DocumentExample.json b/nifi-nar-bundles/nifi-elasticsearch-bundle/nifi-elasticsearch-processors/src/test/resources/DocumentExample.json
deleted file mode 100644
index 66449cf1e1..0000000000
--- a/nifi-nar-bundles/nifi-elasticsearch-bundle/nifi-elasticsearch-processors/src/test/resources/DocumentExample.json
+++ /dev/null
@@ -1,21 +0,0 @@
-{
- "created_at": "Thu Jan 21 16:02:46 +0000 2016",
- "text": "This is a test document from a mock social media service",
- "contributors": null,
- "id": 28039652140,
- "shares": null,
- "geographic_location": null,
- "userinfo": {
- "name": "Not A. Person",
- "location": "Orlando, FL",
- "created_at": "Fri Oct 24 23:22:09 +0000 2008",
- "follow_count": 1,
- "url": "http://not.a.real.site",
- "id": 16958875,
- "lang": "en",
- "time_zone": "Mountain Time (US & Canada)",
- "description": "I'm a test person.",
- "following_count": 71,
- "screen_name": "Nobody"
- }
-}
\ No newline at end of file
diff --git a/nifi-nar-bundles/nifi-elasticsearch-bundle/nifi-elasticsearch-processors/src/test/resources/query-page1.json b/nifi-nar-bundles/nifi-elasticsearch-bundle/nifi-elasticsearch-processors/src/test/resources/query-page1.json
deleted file mode 100644
index 649b480ffc..0000000000
--- a/nifi-nar-bundles/nifi-elasticsearch-bundle/nifi-elasticsearch-processors/src/test/resources/query-page1.json
+++ /dev/null
@@ -1,59 +0,0 @@
-{
-
- "took": 3,
- "timed_out": false,
- "_shards": {
- "total": 5,
- "successful": 5,
- "failed": 0
- },
- "hits": {
- "total": 3,
- "max_score": null,
- "hits": [
- {
- "_index": "myindex",
- "_type": "provenance",
- "_id": "abc-97b-ASVsZu_vShwtGCJpGOObmuSqUJRUC3L_-SEND-S3",
- "_score": null,
- "_source": {
- "timestamp": "2016-07-22T14:47:08.102Z",
- "event_type": "SEND",
- "source": "Twitter",
- "identifier": "abc-97b",
- "transit_type": "S3",
- "transit_uri": "file://cluster2/data/outgoing/S3/abc-97b.zip",
- "object_type": "Provenance Record",
- "version": "ASVsZu_vShwtGCJpGOObmuSqUJRUC3L_",
- "file_size": "3645525",
- "tags": ["arrays", "are", "supported", "too"]
- },
- "sort": [
- 1469198828102
- ]
- },
- {
- "_index": "myindex",
- "_type": "provenance",
- "_id": "abc-a78-SjJkrwnv6edIRqJChEYzrE7PeT1hzioz-SEND-S3",
- "_score": null,
- "_source": {
- "timestamp": "2016-07-22T14:47:08.101Z",
- "event_type": "SEND",
- "source": "Twitter",
- "identifier": "abc-a78",
- "transit_type": "S3",
- "transit_uri": "file://cluster2/data/outgoing/S3/abc-a78.zip",
- "object_type": "Provenance Record",
- "version": "SjJkrwnv6edIRqJChEYzrE7PeT1hzioz",
- "file_size": "4480294",
- "tags": ["arrays", "are", "supported", "too"]
- },
- "sort": [
- 1469198828101
- ]
-
- }
- ]
- }
- }
diff --git a/nifi-nar-bundles/nifi-elasticsearch-bundle/nifi-elasticsearch-processors/src/test/resources/query-page2.json b/nifi-nar-bundles/nifi-elasticsearch-bundle/nifi-elasticsearch-processors/src/test/resources/query-page2.json
deleted file mode 100644
index 01357032be..0000000000
--- a/nifi-nar-bundles/nifi-elasticsearch-bundle/nifi-elasticsearch-processors/src/test/resources/query-page2.json
+++ /dev/null
@@ -1,37 +0,0 @@
-{
-
- "took": 3,
- "timed_out": false,
- "_shards": {
- "total": 5,
- "successful": 5,
- "failed": 0
- },
- "hits": {
- "total": 3,
- "max_score": null,
- "hits": [
- {
- "_index": "myindex",
- "_type": "provenance",
- "_id": "abc-42a-ArPsIlGBKqDvfL6qQZOVpmDwUEB.nynh-SEND-S3",
- "_score": null,
- "_source": {
- "timestamp": "2016-07-22T14:47:08.101Z",
- "event_type": "SEND",
- "source": "Twitter",
- "identifier": "abc-42a",
- "transit_type": "S3",
- "transit_uri": "file://cluster2/data/outgoing/S3/abc-42a.zip",
- "object_type": "Provenance Record",
- "version": "ArPsIlGBKqDvfL6qQZOVpmDwUEB.nynh",
- "file_size": "18206872",
- "tags": ["arrays", "are", "supported", "too"]
- },
- "sort": [
- 1469198828101
- ]
- }
- ]
- }
- }
diff --git a/nifi-nar-bundles/nifi-elasticsearch-bundle/nifi-elasticsearch-processors/src/test/resources/query-page3.json b/nifi-nar-bundles/nifi-elasticsearch-bundle/nifi-elasticsearch-processors/src/test/resources/query-page3.json
deleted file mode 100644
index a495930b01..0000000000
--- a/nifi-nar-bundles/nifi-elasticsearch-bundle/nifi-elasticsearch-processors/src/test/resources/query-page3.json
+++ /dev/null
@@ -1,14 +0,0 @@
-{
- "took": 6,
- "timed_out": false,
- "_shards": {
- "total": 5,
- "successful": 5,
- "failed": 0
- },
- "hits": {
- "total": 3,
- "max_score": null,
- "hits": [ ]
- }
-}
\ No newline at end of file
diff --git a/nifi-nar-bundles/nifi-elasticsearch-bundle/nifi-elasticsearch-processors/src/test/resources/scroll-page1.json b/nifi-nar-bundles/nifi-elasticsearch-bundle/nifi-elasticsearch-processors/src/test/resources/scroll-page1.json
deleted file mode 100644
index 8d8c94063a..0000000000
--- a/nifi-nar-bundles/nifi-elasticsearch-bundle/nifi-elasticsearch-processors/src/test/resources/scroll-page1.json
+++ /dev/null
@@ -1,56 +0,0 @@
-{
- "_scroll_id": "cXVlcnlUaGVuRmV0Y2g7NTsyMDU3NjU6WUlIQVpmWTlRZWl4aURSWUVVR0lXdzsyMDU3NjY6WUlIQVpmWTlRZWl4aURSWUVVR0lXdzsyMDU3Njg6WUlIQVpmWTlRZWl4aURSWUVVR0lXdzsyMDU3Njk6WUlIQVpmWTlRZWl4aURSWUVVR0lXdzsyMDU3Njc6WUlIQVpmWTlRZWl4aURSWUVVR0lXdzswOw==",
- "took": 4,
- "timed_out": false,
- "_shards": {
- "total": 5,
- "successful": 5,
- "failed": 0
- },
- "hits": {
- "total": 3,
- "max_score": null,
- "hits": [
- {
- "_index": "myindex",
- "_type": "provenance",
- "_id": "abc-97b-ASVsZu_vShwtGCJpGOObmuSqUJRUC3L_-SEND-S3",
- "_score": null,
- "_source": {
- "timestamp": "2016-07-22T14:47:08.102Z",
- "event_type": "SEND",
- "source": "Twitter",
- "identifier": "abc-97b",
- "transit_type": "S3",
- "transit_uri": "file://cluster2/data/outgoing/S3/abc-97b.zip",
- "object_type": "Provenance Record",
- "version": "ASVsZu_vShwtGCJpGOObmuSqUJRUC3L_",
- "file_size": "3645525"
- },
- "sort": [
- 1469198828102
- ]
- },
- {
- "_index": "myindex",
- "_type": "provenance",
- "_id": "abc-a78-SjJkrwnv6edIRqJChEYzrE7PeT1hzioz-SEND-S3",
- "_score": null,
- "_source": {
- "timestamp": "2016-07-22T14:47:08.101Z",
- "event_type": "SEND",
- "source": "Twitter",
- "identifier": "abc-a78",
- "transit_type": "S3",
- "transit_uri": "file://cluster2/data/outgoing/S3/abc-a78.zip",
- "object_type": "Provenance Record",
- "version": "SjJkrwnv6edIRqJChEYzrE7PeT1hzioz",
- "file_size": "4480294"
- },
- "sort": [
- 1469198828101
- ]
- }
- ]
- }
-}
\ No newline at end of file
diff --git a/nifi-nar-bundles/nifi-elasticsearch-bundle/nifi-elasticsearch-processors/src/test/resources/scroll-page2.json b/nifi-nar-bundles/nifi-elasticsearch-bundle/nifi-elasticsearch-processors/src/test/resources/scroll-page2.json
deleted file mode 100644
index c02dfdc991..0000000000
--- a/nifi-nar-bundles/nifi-elasticsearch-bundle/nifi-elasticsearch-processors/src/test/resources/scroll-page2.json
+++ /dev/null
@@ -1,36 +0,0 @@
-{
- "_scroll_id": "dXVlcnlUaGVuRmV0Y2g7NTsyMDU3NjU6WUlIQVpmWTlRZWl4aURSWUVVR0lXdzsyMDU3NjY6WUlIQVpmWTlRZWl4aURSWUVVR0lXdzsyMDU3Njg6WUlIQVpmWTlRZWl4aURSWUVVR0lXdzsyMDU3Njk6WUlIQVpmWTlRZWl4aURSWUVVR0lXdzsyMDU3Njc6WUlIQVpmWTlRZWl4aURSWUVVR0lXdzswOw==",
- "took": 4,
- "timed_out": false,
- "_shards": {
- "total": 5,
- "successful": 5,
- "failed": 0
- },
- "hits": {
- "total": 3,
- "max_score": null,
- "hits": [
- {
- "_index": "myindex",
- "_type": "provenance",
- "_id": "abc-97b-ASVsZu_vShwtGCJpGOObmuSqUJRUC3L_-SEND-S3",
- "_score": null,
- "_source": {
- "timestamp": "2016-07-22T14:47:08.102Z",
- "event_type": "SEND",
- "source": "Twitter",
- "identifier": "abc-97b",
- "transit_type": "S3",
- "transit_uri": "file://cluster2/data/outgoing/S3/abc-97b.zip",
- "object_type": "Provenance Record",
- "version": "ASVsZu_vShwtGCJpGOObmuSqUJRUC3L_",
- "file_size": "3645525"
- },
- "sort": [
- 1469198828102
- ]
- }
- ]
- }
-}
\ No newline at end of file
diff --git a/nifi-nar-bundles/nifi-elasticsearch-bundle/nifi-elasticsearch-processors/src/test/resources/scroll-page3.json b/nifi-nar-bundles/nifi-elasticsearch-bundle/nifi-elasticsearch-processors/src/test/resources/scroll-page3.json
deleted file mode 100644
index 90bd3ba20f..0000000000
--- a/nifi-nar-bundles/nifi-elasticsearch-bundle/nifi-elasticsearch-processors/src/test/resources/scroll-page3.json
+++ /dev/null
@@ -1,15 +0,0 @@
-{
- "_scroll_id": "eXVlcnlUaGVuRmV0Y2g7NTsyMDU3NjU6WUlIQVpmWTlRZWl4aURSWUVVR0lXdzsyMDU3NjY6WUlIQVpmWTlRZWl4aURSWUVVR0lXdzsyMDU3Njg6WUlIQVpmWTlRZWl4aURSWUVVR0lXdzsyMDU3Njk6WUlIQVpmWTlRZWl4aURSWUVVR0lXdzsyMDU3Njc6WUlIQVpmWTlRZWl4aURSWUVVR0lXdzswOw==",
- "took": 4,
- "timed_out": false,
- "_shards": {
- "total": 5,
- "successful": 5,
- "failed": 0
- },
- "hits": {
- "total": 3,
- "max_score": null,
- "hits": [ ]
- }
-}
\ No newline at end of file
diff --git a/nifi-nar-bundles/nifi-elasticsearch-bundle/pom.xml b/nifi-nar-bundles/nifi-elasticsearch-bundle/pom.xml
index 02f71b4078..128260533f 100644
--- a/nifi-nar-bundles/nifi-elasticsearch-bundle/pom.xml
+++ b/nifi-nar-bundles/nifi-elasticsearch-bundle/pom.xml
@@ -27,8 +27,6 @@ language governing permissions and limitations under the License. -->
<module>nifi-elasticsearch-client-service-api-nar</module>
<module>nifi-elasticsearch-client-service</module>
<module>nifi-elasticsearch-client-service-nar</module>
- <module>nifi-elasticsearch-nar</module>
- <module>nifi-elasticsearch-processors</module>
<module>nifi-elasticsearch-restapi-nar</module>
<module>nifi-elasticsearch-restapi-processors</module>
</modules>