You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@nifi.apache.org by mt...@apache.org on 2022/11/05 22:18:09 UTC
[nifi] branch main updated: NIFI-9398 add verification to ElasticSearchClientService (with integration tests) and Elasticsearch REST API processors NIFI-9398 processor verification integration-tests for nifi-elasticsearch-restapi-processors using common nifi-elasticsearch-test-utils refactored from nifi-elasticsearch-client-service NIFI-9398 fail index exists verification for Elasticsearch processors whose operations would not create the index if it doesn't already exist
This is an automated email from the ASF dual-hosted git repository.
mthomsen 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 2acfd1e496 NIFI-9398 add verification to ElasticSearchClientService (with integration tests) and Elasticsearch REST API processors NIFI-9398 processor verification integration-tests for nifi-elasticsearch-restapi-processors using common nifi-elasticsearch-test-utils refactored from nifi-elasticsearch-client-service NIFI-9398 fail index exists verification for Elasticsearch processors whose operations would not create the index if it doesn't already exist
2acfd1e496 is described below
commit 2acfd1e4967e9989fcfe797729c98321e7fa150f
Author: Chris Sampson <ch...@gmail.com>
AuthorDate: Tue Mar 15 20:59:10 2022 +0000
NIFI-9398 add verification to ElasticSearchClientService (with integration tests) and Elasticsearch REST API processors
NIFI-9398 processor verification integration-tests for nifi-elasticsearch-restapi-processors using common nifi-elasticsearch-test-utils refactored from nifi-elasticsearch-client-service
NIFI-9398 fail index exists verification for Elasticsearch processors whose operations would not create the index if it doesn't already exist
This closes #6544
Signed-off-by: Mike Thomsen <mt...@apache.org>
---
.../README.md | 29 +-
.../elasticsearch/ElasticSearchClientService.java | 11 +-
.../nifi-elasticsearch-client-service/pom.xml | 79 +---
.../ElasticSearchClientServiceImpl.java | 154 ++++++-
.../TestElasticSearchClientService.java | 13 +
.../integration/AbstractElasticsearch_IT.java | 169 ++------
.../integration/ElasticSearchClientService_IT.java | 480 +++++++++++----------
.../integration/ElasticSearchLookupService_IT.java | 134 +++---
.../src/test/resources/conf-5/jvm.options | 127 ------
.../src/test/resources/setup-5.script | 48 ---
.../nifi-elasticsearch-restapi-processors/pom.xml | 65 +--
.../AbstractByQueryElasticsearch.java | 18 +-
.../AbstractJsonQueryElasticsearch.java | 13 +-
.../elasticsearch/AbstractPutElasticsearch.java | 18 +-
.../elasticsearch/DeleteByQueryElasticsearch.java | 2 +-
.../elasticsearch/ElasticsearchRestProcessor.java | 70 ++-
.../processors/elasticsearch/GetElasticsearch.java | 77 +++-
.../elasticsearch/JsonQueryElasticsearch.java | 2 +-
.../PaginatedJsonQueryElasticsearch.java | 2 +-
.../elasticsearch/PutElasticsearchJson.java | 8 +-
.../elasticsearch/PutElasticsearchRecord.java | 6 +-
.../elasticsearch/SearchElasticsearch.java | 2 +-
.../elasticsearch/UpdateByQueryElasticsearch.java | 2 +-
.../TestElasticsearchClientService.groovy | 13 +
.../mock/AbstractMockElasticsearchClient.groovy | 13 +
.../src/test/java/.gitignore | 1 -
.../integration/AbstractElasticsearch_IT.java | 127 ++++++
.../integration/GetElasticsearch_IT.java | 106 +++++
.../integration/PutElasticsearchJson_IT.java | 26 ++
.../src/test/resources/setup-6.script | 21 +
.../src/test/resources/setup-7.script | 20 +
.../src/test/resources/setup-8.script | 20 +
.../nifi-elasticsearch-test-utils/pom.xml | 73 ++++
.../org/apache/nifi/elasticsearch/MapBuilder.java | 70 +++
.../integration/AbstractElasticsearchITBase.java | 192 +++++++++
nifi-nar-bundles/nifi-elasticsearch-bundle/pom.xml | 71 ++-
.../nifi-framework/nifi-file-authorizer/pom.xml | 1 -
.../nifi-framework/nifi-shell-authorizer/pom.xml | 1 -
nifi-nar-bundles/nifi-smb-bundle/pom.xml | 10 -
.../nifi-registry-core/nifi-registry-test/pom.xml | 1 +
nifi-registry/pom.xml | 16 -
pom.xml | 8 +-
42 files changed, 1495 insertions(+), 824 deletions(-)
diff --git a/nifi-nar-bundles/nifi-elasticsearch-bundle/nifi-elasticsearch-client-service/README.md b/nifi-nar-bundles/nifi-elasticsearch-bundle/README.md
similarity index 73%
rename from nifi-nar-bundles/nifi-elasticsearch-bundle/nifi-elasticsearch-client-service/README.md
rename to nifi-nar-bundles/nifi-elasticsearch-bundle/README.md
index 82d7b194c9..b08546342c 100644
--- a/nifi-nar-bundles/nifi-elasticsearch-bundle/nifi-elasticsearch-client-service/README.md
+++ b/nifi-nar-bundles/nifi-elasticsearch-bundle/README.md
@@ -13,36 +13,43 @@
limitations under the License.
-->
-# Elasticsearch Client Service
+# Integration Tests
-## Integration Tests
-
-### Overview
+## Overview
The integration tests use [Testcontainers](https://www.testcontainers.org/) to provide a sane default for developers who have installed Docker. Testcontainers support can be disabled by setting the system property `elasticsearch.testcontainers.enabled` to something other than `true`. If Testcontainers are disabled, the endpoint will need to be configured. It can be set manually with the system property `elasticsearch.endpoint`. The default value is `http://localhost:9200`.
-### Maven Profiles
+## Maven Profiles
* `integration-tests`
* `elasticsearch6`
* `elasticsearch7`
-### Configurable System Properties
+## Configurable System Properties
* `elasticsearch.endpoint` - Manually configure the endpoint root for a non-Docker version of Elasticsearch,
* `elasticsearch.testcontainers.enabled` - Set to anything other than `true` to disable Testcontainers and use a non-Docker version of Elasticsearch.
* `elasticsearch.elastic_user.password` - Set the Elasticsearch `elastic` user's password. When Testcontainers are enabled, this sets up the Docker container and the rest clients for accessing it within the tests. When Testcontainers are disabled, it needs to be set to whatever password is used on the external Elasticsearch node or cluster.
-### Maven Run Examples
+## Maven Run Examples
Elasticsearch 8.X is the current default version of Elasticsearch when Testcontainers are used. An example run of the integration tests with Elasticsearch 7 support would be like this:
-`mvn clean install -Pintegration-tests,elasticsearch7`
+`mvn -Pintegration-tests,elasticsearch7 --fail-at-end clean install`
An example using a non-Docker version of Elasticsearch:
-`mvn clean install -Pintegration-tests -Delasticsearch.testcontainers.enabled=false -Delasticsearch.elastic_user.password=s3cret1234`
+`mvn -Pintegration-tests --fail-at-end -Delasticsearch.testcontainers.enabled=false -Delasticsearch.elastic_user.password=s3cret1234 clean install`
+
+## Modules with Integration Tests (using Testcontainers)
+
+- [Elasticsearch Client Service](nifi-elasticsearch-client-service)
+- [Elasticsearch REST API Processors](nifi-elasticsearch-restapi-processors)
+
+## Misc
+
+Integration Tests with Testcontainers currently only uses the `amd64` Docker Images.
-### Misc
+`elasticsearch6` is known to **not** work with `arm64` machines (e.g. Mac M1/M2), but other Elasticsearch images (e.g. 7.x and 8.x) appear to work.
-The Testcontainers support currently only supports the x64 release of Dockerized Elasticsearch. ARM64 support may be added later.
\ No newline at end of file
+Explicit `arm64` architecture support may be added in future where the Elasticsearch images exist.
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 d1dc33a37c..15a45b4b7f 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
@@ -22,6 +22,7 @@ import org.apache.nifi.components.AllowableValue;
import org.apache.nifi.components.PropertyDescriptor;
import org.apache.nifi.components.Validator;
import org.apache.nifi.controller.ControllerService;
+import org.apache.nifi.controller.VerifiableControllerService;
import org.apache.nifi.expression.ExpressionLanguageScope;
import org.apache.nifi.processor.util.StandardValidators;
import org.apache.nifi.proxy.ProxyConfiguration;
@@ -33,7 +34,7 @@ import java.util.Map;
@Tags({"elasticsearch", "client"})
@CapabilityDescription("A controller service for accessing an Elasticsearch client.")
-public interface ElasticSearchClientService extends ControllerService {
+public interface ElasticSearchClientService extends ControllerService, VerifiableControllerService {
PropertyDescriptor HTTP_HOSTS = new PropertyDescriptor.Builder()
.name("el-cs-http-hosts")
.displayName("HTTP Hosts")
@@ -203,6 +204,14 @@ public interface ElasticSearchClientService extends ControllerService {
*/
void refresh(final String index, final Map<String, String> requestParameters);
+ /**
+ * Check whether an index exists.
+ *
+ * @param index The index to target, if omitted then all indices will be updated.
+ * @param requestParameters A collection of URL request parameters. Optional.
+ */
+ boolean exists(final String index, final Map<String, String> requestParameters);
+
/**
* Get a document by ID.
*
diff --git a/nifi-nar-bundles/nifi-elasticsearch-bundle/nifi-elasticsearch-client-service/pom.xml b/nifi-nar-bundles/nifi-elasticsearch-bundle/nifi-elasticsearch-client-service/pom.xml
index 6fee72e581..f812fffbf5 100644
--- a/nifi-nar-bundles/nifi-elasticsearch-bundle/nifi-elasticsearch-client-service/pom.xml
+++ b/nifi-nar-bundles/nifi-elasticsearch-bundle/nifi-elasticsearch-client-service/pom.xml
@@ -131,25 +131,6 @@
<dependency>
<groupId>org.elasticsearch.client</groupId>
<artifactId>elasticsearch-rest-client</artifactId>
- <!-- stopped at 7.13.4 as it is the last version prior to Elastic forcing the client to check it is connecting
- to an Elastic-provided Elasticsearch instead of an instance provided by someone else (e.g. AWS OpenSearch)
- see: https://opensearch.org/blog/community/2021/08/community-clients/ for more info.
-
- Note: the low-level elasticsearch-rest-client remains licensed with Apache 2.0
- (https://www.elastic.co/guide/en/elasticsearch/client/java-rest/current/_license.html) even after the move
- of the main Elasticsearch product and elasticsearch-rest-high-level-client to Elastic 2.0/SSPL 1.0 in v7.11.0+ -->
- <version>7.13.4</version>
- <scope>compile</scope>
- <exclusions>
- <exclusion>
- <groupId>com.vividsolutions</groupId>
- <artifactId>jts</artifactId>
- </exclusion>
- <exclusion>
- <groupId>commons-logging</groupId>
- <artifactId>commons-logging</artifactId>
- </exclusion>
- </exclusions>
</dependency>
<!-- test dependencies -->
@@ -170,6 +151,12 @@
<artifactId>nifi-ssl-context-service</artifactId>
<scope>test</scope>
</dependency>
+ <dependency>
+ <groupId>org.apache.nifi</groupId>
+ <artifactId>nifi-elasticsearch-test-utils</artifactId>
+ <version>1.19.0-SNAPSHOT</version>
+ <scope>test</scope>
+ </dependency>
<dependency>
<groupId>org.apache.maven</groupId>
<artifactId>maven-artifact</artifactId>
@@ -182,59 +169,5 @@
<version>${nifi.groovy.version}</version>
<scope>test</scope>
</dependency>
-
- <dependency>
- <groupId>org.testcontainers</groupId>
- <artifactId>elasticsearch</artifactId>
- <version>${testcontainers.version}</version>
- </dependency>
- <dependency>
- <groupId>org.testcontainers</groupId>
- <artifactId>junit-jupiter</artifactId>
- <version>${testcontainers.version}</version>
- </dependency>
</dependencies>
-
- <profiles>
- <profile>
- <id>integration-tests</id>
- <activation>
- <activeByDefault>false</activeByDefault>
- </activation>
- <properties>
- <elasticsearch_docker_image>8.4.3</elasticsearch_docker_image>
- <elasticsearch.elastic.password>s3cret</elasticsearch.elastic.password>
- </properties>
- <build>
- <pluginManagement>
- <plugins>
- <plugin>
- <groupId>org.apache.maven.plugins</groupId>
- <artifactId>maven-failsafe-plugin</artifactId>
- <configuration>
- <systemPropertyVariables>
- <elasticsearch.endpoint>http://localhost:9200</elasticsearch.endpoint>
- <elasticsearch.testcontainers.enabled>true</elasticsearch.testcontainers.enabled>
- <elasticsearch.docker.image>docker.elastic.co/elasticsearch/elasticsearch:${elasticsearch_docker_image}</elasticsearch.docker.image>
- <elasticsearch.elastic_user.password>${elasticsearch.elastic.password}</elasticsearch.elastic_user.password>
- </systemPropertyVariables>
- </configuration>
- </plugin>
- </plugins>
- </pluginManagement>
- </build>
- </profile>
- <profile>
- <id>elasticsearch6</id>
- <properties>
- <elasticsearch_docker_image>6.8.23</elasticsearch_docker_image>
- </properties>
- </profile>
- <profile>
- <id>elasticsearch7</id>
- <properties>
- <elasticsearch_docker_image>7.17.6</elasticsearch_docker_image>
- </properties>
- </profile>
- </profiles>
</project>
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 2f2ae2a1a2..29ecad5609 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
@@ -32,9 +32,11 @@ import org.apache.http.impl.client.BasicCredentialsProvider;
import org.apache.http.nio.entity.NStringEntity;
import org.apache.nifi.annotation.lifecycle.OnDisabled;
import org.apache.nifi.annotation.lifecycle.OnEnabled;
+import org.apache.nifi.components.ConfigVerificationResult;
import org.apache.nifi.components.PropertyDescriptor;
import org.apache.nifi.controller.AbstractControllerService;
import org.apache.nifi.controller.ConfigurationContext;
+import org.apache.nifi.logging.ComponentLog;
import org.apache.nifi.processor.exception.ProcessException;
import org.apache.nifi.proxy.ProxyConfiguration;
import org.apache.nifi.proxy.ProxyConfigurationService;
@@ -67,6 +69,10 @@ import java.util.concurrent.TimeUnit;
import java.util.stream.Collectors;
public class ElasticSearchClientServiceImpl extends AbstractControllerService implements ElasticSearchClientService {
+ public static final String VERIFICATION_STEP_CONNECTION = "Elasticsearch Connection";
+ public static final String VERIFICATION_STEP_CLIENT_SETUP = "Elasticsearch Rest Client Setup";
+ public static final String VERIFICATION_STEP_WARNINGS = "Elasticsearch Warnings";
+
private ObjectMapper mapper;
private static final List<PropertyDescriptor> properties;
@@ -100,7 +106,7 @@ public class ElasticSearchClientServiceImpl extends AbstractControllerService im
@OnEnabled
public void onEnabled(final ConfigurationContext context) throws InitializationException {
try {
- setupClient(context);
+ this.client = setupClient(context);
responseCharset = Charset.forName(context.getProperty(CHARSET).getValue());
// re-create the ObjectMapper in case the SUPPRESS_NULLS property has changed - the JsonInclude settings aren't dynamic
@@ -117,13 +123,83 @@ public class ElasticSearchClientServiceImpl extends AbstractControllerService im
@OnDisabled
public void onDisabled() throws IOException {
- this.client.close();
+ if (this.client != null) {
+ this.client.close();
+ this.client = null;
+ }
this.url = null;
}
- private void setupClient(final ConfigurationContext context) throws MalformedURLException, InitializationException {
+ @Override
+ public List<ConfigVerificationResult> verify(final ConfigurationContext context, final ComponentLog verificationLogger, final Map<String, String> variables) {
+ final List<ConfigVerificationResult> results = new ArrayList<>();
+ final ConfigVerificationResult.Builder clientSetupResult = new ConfigVerificationResult.Builder()
+ .verificationStepName(VERIFICATION_STEP_CLIENT_SETUP);
+ final ConfigVerificationResult.Builder connectionResult = new ConfigVerificationResult.Builder()
+ .verificationStepName(VERIFICATION_STEP_CONNECTION);
+ final ConfigVerificationResult.Builder warningsResult = new ConfigVerificationResult.Builder()
+ .verificationStepName(VERIFICATION_STEP_WARNINGS);
+
+ // configure the Rest Client
+ try (final RestClient verifyClient = setupClient(context)) {
+ clientSetupResult.outcome(ConfigVerificationResult.Outcome.SUCCESSFUL);
+
+ // try to fetch the Elasticsearch root endpoint (system summary)
+ verifyRootConnection(verifyClient, connectionResult, warningsResult);
+ }catch (final MalformedURLException mue) {
+ clientSetupResult.outcome(ConfigVerificationResult.Outcome.FAILED)
+ .explanation("Incorrect/invalid " + ElasticSearchClientService.HTTP_HOSTS.getDisplayName());
+ } catch (final InitializationException ie) {
+ clientSetupResult.outcome(ConfigVerificationResult.Outcome.FAILED)
+ .explanation("Incorrect/invalid " + ElasticSearchClientService.PROP_SSL_CONTEXT_SERVICE.getDisplayName());
+ } catch (final Exception ex) {
+ getLogger().warn("Unable to setup Elasticsearch Rest Client", ex);
+
+ clientSetupResult.outcome(ConfigVerificationResult.Outcome.FAILED)
+ .explanation("Unable to configure, are all mandatory properties set (see logs for details)?");
+ } finally {
+ final ConfigVerificationResult clientSetup = clientSetupResult.build();
+ if (ConfigVerificationResult.Outcome.SUCCESSFUL != clientSetup.getOutcome()) {
+ connectionResult.outcome(ConfigVerificationResult.Outcome.SKIPPED)
+ .explanation("Elasticsearch Rest Client not configured");
+ warningsResult.outcome(ConfigVerificationResult.Outcome.SKIPPED)
+ .explanation("Elasticsearch Rest Client not configured");
+ }
+
+ results.add(clientSetup);
+ results.add(connectionResult.build());
+ results.add(warningsResult.build());
+ }
+
+ return results;
+ }
+
+ private void verifyRootConnection(final RestClient verifyClient, final ConfigVerificationResult.Builder connectionResult, final ConfigVerificationResult.Builder warningsResult) {
+ try {
+ final Response response = verifyClient.performRequest(new Request("GET", "/"));
+ final List<String> warnings = parseResponseWarningHeaders(response);
+ parseResponse(response);
+
+ connectionResult.outcome(ConfigVerificationResult.Outcome.SUCCESSFUL);
+ if (warnings.isEmpty()) {
+ warningsResult.outcome(ConfigVerificationResult.Outcome.SUCCESSFUL);
+ } else {
+ warningsResult.outcome(ConfigVerificationResult.Outcome.FAILED)
+ .explanation("Elasticsearch Warnings received during request (see logs for details)");
+ }
+ } catch (ElasticsearchException | IOException ex) {
+ getLogger().warn("Unable to connect to Elasticsearch", ex);
+
+ connectionResult.outcome(ConfigVerificationResult.Outcome.FAILED)
+ .explanation("Unable to retrieve system summary from Elasticsearch root endpoint");
+ warningsResult.outcome(ConfigVerificationResult.Outcome.SKIPPED)
+ .explanation("Connection to Elasticsearch failed");
+ }
+ }
+
+ private RestClient setupClient(final ConfigurationContext context) throws MalformedURLException, InitializationException {
final String hosts = context.getProperty(HTTP_HOSTS).evaluateAttributeExpressions().getValue();
- final String[] hostsSplit = hosts.split(",[\\s]*");
+ final String[] hostsSplit = hosts.split(",\\s*");
this.url = hostsSplit[0];
final SSLContextService sslService =
context.getProperty(PROP_SSL_CONTEXT_SERVICE).asControllerService(SSLContextService.class);
@@ -183,7 +259,7 @@ public class ElasticSearchClientServiceImpl extends AbstractControllerService im
return requestConfigBuilder;
});
- this.client = builder.build();
+ return builder.build();
}
private CredentialsProvider addCredentials(final CredentialsProvider credentialsProvider, final AuthScope authScope, final String username, final String password) {
@@ -199,16 +275,22 @@ public class ElasticSearchClientServiceImpl extends AbstractControllerService im
return cp;
}
+ private void appendIndex(final StringBuilder sb, final String index) {
+ if (StringUtils.isNotBlank(index) && !"/".equals(index)) {
+ if (!index.startsWith("/")) {
+ sb.append("/");
+ }
+ sb.append(index);
+ }
+ }
+
private Response runQuery(final String endpoint, final String query, final String index, final String type, final Map<String, String> requestParameters) {
final StringBuilder sb = new StringBuilder();
- if (StringUtils.isNotBlank(index)) {
- sb.append("/").append(index);
- }
+ appendIndex(sb, index);
if (StringUtils.isNotBlank(type)) {
sb.append("/").append(type);
}
-
- sb.append(String.format("/%s", endpoint));
+ sb.append("/").append(endpoint);
final HttpEntity queryEntity = new NStringEntity(query, ContentType.APPLICATION_JSON);
try {
@@ -239,11 +321,14 @@ public class ElasticSearchClientServiceImpl extends AbstractControllerService im
}
private List<String> parseResponseWarningHeaders(final Response response) {
- return Arrays.stream(response.getHeaders())
+ final List<String> warnings = Arrays.stream(response.getHeaders())
.filter(h -> "Warning".equalsIgnoreCase(h.getName()))
.map(Header::getValue)
- .peek(h -> getLogger().warn("Elasticsearch Warning: {}", h))
.collect(Collectors.toList());
+
+ warnings.forEach(w -> getLogger().warn("Elasticsearch Warning: {}", w));
+
+ return warnings;
}
@Override
@@ -405,9 +490,7 @@ public class ElasticSearchClientServiceImpl extends AbstractControllerService im
public void refresh(final String index, final Map<String, String> requestParameters) {
try {
final StringBuilder endpoint = new StringBuilder();
- if (StringUtils.isNotBlank(index) && !"/".equals(index)) {
- endpoint.append("/").append(index);
- }
+ appendIndex(endpoint, index);
endpoint.append("/_refresh");
final Response response = performRequest("POST", endpoint.toString(), requestParameters, null);
parseResponseWarningHeaders(response);
@@ -416,12 +499,34 @@ public class ElasticSearchClientServiceImpl extends AbstractControllerService im
}
}
+ @Override
+ public boolean exists(final String index, final Map<String, String> requestParameters) {
+ try {
+ final StringBuilder endpoint = new StringBuilder();
+ appendIndex(endpoint, index);
+ final Response response = performRequest("HEAD", endpoint.toString(), requestParameters, null);
+ parseResponseWarningHeaders(response);
+
+ if (response.getStatusLine().getStatusCode() == 200) {
+ return true;
+ } else if (response.getStatusLine().getStatusCode() == 404) {
+ return false;
+ } else {
+ throw new ProcessException(String.format("Error checking for index existence: %d; %s",
+ response.getStatusLine().getStatusCode(), response.getStatusLine().getReasonPhrase())
+ );
+ }
+ } catch (final Exception ex) {
+ throw new ElasticsearchException(ex);
+ }
+ }
+
@SuppressWarnings("unchecked")
@Override
public Map<String, Object> get(final String index, final String type, final String id, final Map<String, String> requestParameters) {
try {
final StringBuilder endpoint = new StringBuilder();
- endpoint.append("/").append(index);
+ appendIndex(endpoint, index);
if (StringUtils.isNotBlank(type)) {
endpoint.append("/").append(type);
} else {
@@ -482,7 +587,10 @@ public class ElasticSearchClientServiceImpl extends AbstractControllerService im
put("keep_alive", keepAlive);
}
}};
- final Response response = performRequest("POST", "/" + index + "/_pit", params, null);
+ final StringBuilder endpoint = new StringBuilder();
+ appendIndex(endpoint, index);
+ endpoint.append("/_pit");
+ final Response response = performRequest("POST", endpoint.toString(), params, null);
final String body = IOUtils.toString(response.getEntity().getContent(), StandardCharsets.UTF_8);
parseResponseWarningHeaders(response);
@@ -601,11 +709,13 @@ public class ElasticSearchClientServiceImpl extends AbstractControllerService im
@Override
public String getTransitUrl(final String index, final String type) {
- return this.url +
- (StringUtils.isNotBlank(index) ? "/" : "") +
- (StringUtils.isNotBlank(index) ? index : "") +
- (StringUtils.isNotBlank(type) ? "/" : "") +
- (StringUtils.isNotBlank(type) ? type : "");
+ final StringBuilder transitUrl = new StringBuilder();
+ transitUrl.append(this.url);
+ appendIndex(transitUrl, index);
+ if (StringUtils.isNotBlank(type)) {
+ transitUrl.append("/").append(type);
+ }
+ return transitUrl.toString();
}
private Response performRequest(final String method, final String endpoint, final Map<String, String> parameters, final HttpEntity entity) throws IOException {
diff --git a/nifi-nar-bundles/nifi-elasticsearch-bundle/nifi-elasticsearch-client-service/src/test/java/org/apache/nifi/elasticsearch/TestElasticSearchClientService.java b/nifi-nar-bundles/nifi-elasticsearch-bundle/nifi-elasticsearch-client-service/src/test/java/org/apache/nifi/elasticsearch/TestElasticSearchClientService.java
index 01dc128ca3..4ee196a9dc 100644
--- a/nifi-nar-bundles/nifi-elasticsearch-bundle/nifi-elasticsearch-client-service/src/test/java/org/apache/nifi/elasticsearch/TestElasticSearchClientService.java
+++ b/nifi-nar-bundles/nifi-elasticsearch-bundle/nifi-elasticsearch-client-service/src/test/java/org/apache/nifi/elasticsearch/TestElasticSearchClientService.java
@@ -17,7 +17,10 @@
package org.apache.nifi.elasticsearch;
+import org.apache.nifi.components.ConfigVerificationResult;
import org.apache.nifi.controller.AbstractControllerService;
+import org.apache.nifi.controller.ConfigurationContext;
+import org.apache.nifi.logging.ComponentLog;
import java.util.ArrayList;
import java.util.HashMap;
@@ -35,6 +38,11 @@ public class TestElasticSearchClientService extends AbstractControllerService im
data.put("position", "Software Engineer");
}
+ @Override
+ public List<ConfigVerificationResult> verify(ConfigurationContext context, ComponentLog verificationLogger, Map<String, String> variables) {
+ return null;
+ }
+
@Override
public IndexOperationResponse add(IndexOperationRequest operation, Map<String, String> requestParameters) {
return null;
@@ -74,6 +82,11 @@ public class TestElasticSearchClientService extends AbstractControllerService im
public void refresh(final String index, final Map<String, String> requestParameters) {
}
+ @Override
+ public boolean exists(final String index, final Map<String, String> requestParameters) {
+ return true;
+ }
+
@Override
public Map<String, Object> get(String index, String type, String id, Map<String, String> requestParameters) {
return data;
diff --git a/nifi-nar-bundles/nifi-elasticsearch-bundle/nifi-elasticsearch-client-service/src/test/java/org/apache/nifi/elasticsearch/integration/AbstractElasticsearch_IT.java b/nifi-nar-bundles/nifi-elasticsearch-bundle/nifi-elasticsearch-client-service/src/test/java/org/apache/nifi/elasticsearch/integration/AbstractElasticsearch_IT.java
index 476f944177..40df462a1a 100644
--- a/nifi-nar-bundles/nifi-elasticsearch-bundle/nifi-elasticsearch-client-service/src/test/java/org/apache/nifi/elasticsearch/integration/AbstractElasticsearch_IT.java
+++ b/nifi-nar-bundles/nifi-elasticsearch-bundle/nifi-elasticsearch-client-service/src/test/java/org/apache/nifi/elasticsearch/integration/AbstractElasticsearch_IT.java
@@ -16,153 +16,42 @@
*/
package org.apache.nifi.elasticsearch.integration;
-import com.fasterxml.jackson.databind.ObjectMapper;
-import org.apache.http.HttpEntity;
-import org.apache.http.HttpHost;
-import org.apache.http.auth.UsernamePasswordCredentials;
-import org.apache.http.entity.ContentType;
-import org.apache.http.impl.client.BasicCredentialsProvider;
-import org.apache.http.nio.entity.NStringEntity;
-import org.elasticsearch.client.Request;
-import org.elasticsearch.client.ResponseException;
-import org.elasticsearch.client.RestClient;
-import org.testcontainers.elasticsearch.ElasticsearchContainer;
-import org.testcontainers.utility.DockerImageName;
-
-import java.io.BufferedReader;
-import java.io.FileInputStream;
-import java.io.IOException;
-import java.io.InputStreamReader;
-import java.net.URL;
-import java.util.ArrayList;
+import org.apache.nifi.elasticsearch.ElasticSearchClientService;
+import org.apache.nifi.elasticsearch.ElasticSearchClientServiceImpl;
+import org.apache.nifi.elasticsearch.TestControllerServiceProcessor;
+import org.apache.nifi.util.TestRunners;
+import org.junit.jupiter.api.AfterAll;
+import org.junit.jupiter.api.BeforeEach;
+
+import java.util.Arrays;
import java.util.List;
-import static org.apache.http.auth.AuthScope.ANY;
-
-public abstract class AbstractElasticsearch_IT {
- protected static final DockerImageName IMAGE = DockerImageName
- .parse(System.getProperty("elasticsearch.docker.image"));
- protected static final ElasticsearchContainer ELASTICSEARCH_CONTAINER = new ElasticsearchContainer(IMAGE)
- .withPassword(System.getProperty("elasticsearch.elastic_user.password"))
- .withEnv("xpack.security.enabled", "true");
+abstract class AbstractElasticsearch_IT extends AbstractElasticsearchITBase {
+ static final List<String> TEST_INDICES =
+ Arrays.asList("user_details", "complex", "nested", "bulk_a", "bulk_b", "bulk_c", "error_handler", "messages");
- protected static final String ELASTIC_USER_PASSWORD = System.getProperty("elasticsearch.elastic_user.password");
- protected static final boolean ENABLE_TEST_CONTAINERS = System.getProperty("elasticsearch.testcontainers.enabled")
- != null && System.getProperty("elasticsearch.testcontainers.enabled").equalsIgnoreCase("true");
- protected static String ELASTIC_HOST;
+ ElasticSearchClientServiceImpl service;
- protected static final ObjectMapper MAPPER = new ObjectMapper();
+ @BeforeEach
+ void before() throws Exception {
+ runner = TestRunners.newTestRunner(TestControllerServiceProcessor.class);
+ service = new ElasticSearchClientServiceImpl();
+ runner.addControllerService(CLIENT_SERVICE_NAME, service);
+ runner.setProperty(service, ElasticSearchClientService.HTTP_HOSTS, elasticsearchHost);
+ runner.setProperty(service, ElasticSearchClientService.CONNECT_TIMEOUT, "10000");
+ runner.setProperty(service, ElasticSearchClientService.SOCKET_TIMEOUT, "60000");
+ runner.setProperty(service, ElasticSearchClientService.SUPPRESS_NULLS, ElasticSearchClientService.ALWAYS_SUPPRESS.getValue());
+ runner.setProperty(service, ElasticSearchClientService.USERNAME, "elastic");
+ runner.setProperty(service, ElasticSearchClientService.PASSWORD, ELASTIC_USER_PASSWORD);
- public static void startTestcontainer() {
- if (ENABLE_TEST_CONTAINERS) {
- ELASTICSEARCH_CONTAINER.start();
- ELASTIC_HOST = String.format("http://%s", ELASTICSEARCH_CONTAINER.getHttpHostAddress());
- } else {
- ELASTIC_HOST = System.getProperty("elasticsearch.endpoint");
- }
- }
+ runner.enableControllerService(service);
- public static void stopTestcontainer() {
- if (ENABLE_TEST_CONTAINERS) {
- ELASTICSEARCH_CONTAINER.stop();
- }
+ service.refresh(null, null);
}
- private static String[] getElasticVersion() {
- String fullVersion = IMAGE.getVersionPart();
- String[] parts = fullVersion.split("\\.");
- if (parts.length == 1) {
- throw new RuntimeException("The elasticsearch version should have at least a major and minor version ex. 7.17");
- }
-
- return parts;
- }
-
- protected static int getElasticMajorVersion() {
- return Integer.valueOf(getElasticVersion()[0]);
- }
-
- protected static int getElasticMinorVersion() {
- return Integer.valueOf(getElasticVersion()[1]);
- }
-
- private static RestClient testDataManagementClient;
-
- protected static void setupTestData() throws IOException {
- int majorVersion = getElasticMajorVersion();
- URL url = new URL(ELASTIC_HOST);
- testDataManagementClient = RestClient
- .builder(new HttpHost(url.getHost(), url.getPort(), url.getProtocol()))
- .setHttpClientConfigCallback(httpClientBuilder -> {
- UsernamePasswordCredentials credentials = new UsernamePasswordCredentials("elastic", ELASTIC_USER_PASSWORD);
- BasicCredentialsProvider provider = new BasicCredentialsProvider();
- provider.setCredentials(ANY, credentials);
- httpClientBuilder.setDefaultCredentialsProvider(provider);
- return httpClientBuilder;
- })
- .build();
- String script = String.format("src/test/resources/setup-%s.script", majorVersion);
-
- List<SetupAction> actions = readSetupActions(script);
-
- for (SetupAction action : actions) {
- String endpoint = String.format("%s/%s", ELASTIC_HOST, action.path);
- Request request = new Request(action.verb, endpoint);
- HttpEntity jsonBody = new NStringEntity(action.json, ContentType.APPLICATION_JSON);
- request.setEntity(jsonBody);
-
- try {
- testDataManagementClient.performRequest(request);
- } catch (ResponseException re) {
- throw new RuntimeException(re);
- }
- }
- }
-
- protected static void tearDownTestData() throws IOException {
- deleteIndex("user_details");
- deleteIndex("complex");
- deleteIndex("nested");
- deleteIndex("bulk_a");
- deleteIndex("bulk_b");
- deleteIndex("bulk_c");
- deleteIndex("error_handler");
- deleteIndex("messages");
- }
-
- private static void deleteIndex(String name) throws IOException {
- Request request = new Request("DELETE", String.format("%s/%s", ELASTIC_HOST, name));
- testDataManagementClient.performRequest(request);
- }
-
- private static List<SetupAction> readSetupActions(String scriptPath) throws IOException {
- List<SetupAction> actions = new ArrayList<>();
- BufferedReader reader = new BufferedReader(new InputStreamReader(new FileInputStream(scriptPath)));
- String line = reader.readLine();
- while (line != null) {
- if (!line.trim().isEmpty() && !line.trim().startsWith("#")) {
- String verb = line.substring(0, line.indexOf(":"));
- String path = line.substring(verb.length() + 1, line.indexOf(":", verb.length() + 1));
- int loc = verb.length() + path.length() + 2;
- String json = line.substring(loc);
-
- actions.add(new SetupAction(verb, path, json));
- }
- line = reader.readLine();
- }
-
- return actions;
- }
-
- private static final class SetupAction {
- private String verb;
- private String path;
- private String json;
-
- public SetupAction(String verb, String path, String json) {
- this.verb = verb;
- this.path = path;
- this.json = json;
- }
+ @AfterAll
+ static void afterAll() {
+ tearDownTestData(TEST_INDICES);
+ stopTestcontainer();
}
}
diff --git a/nifi-nar-bundles/nifi-elasticsearch-bundle/nifi-elasticsearch-client-service/src/test/java/org/apache/nifi/elasticsearch/integration/ElasticSearchClientService_IT.java b/nifi-nar-bundles/nifi-elasticsearch-bundle/nifi-elasticsearch-client-service/src/test/java/org/apache/nifi/elasticsearch/integration/ElasticSearchClientService_IT.java
index b136c46c5a..035b3b9bc4 100644
--- a/nifi-nar-bundles/nifi-elasticsearch-bundle/nifi-elasticsearch-client-service/src/test/java/org/apache/nifi/elasticsearch/integration/ElasticSearchClientService_IT.java
+++ b/nifi-nar-bundles/nifi-elasticsearch-bundle/nifi-elasticsearch-client-service/src/test/java/org/apache/nifi/elasticsearch/integration/ElasticSearchClientService_IT.java
@@ -18,33 +18,37 @@
package org.apache.nifi.elasticsearch.integration;
import com.fasterxml.jackson.core.JsonProcessingException;
+import org.apache.nifi.components.ConfigVerificationResult;
+import org.apache.nifi.components.PropertyDescriptor;
+import org.apache.nifi.controller.VerifiableControllerService;
import org.apache.nifi.elasticsearch.DeleteOperationResponse;
import org.apache.nifi.elasticsearch.ElasticSearchClientService;
import org.apache.nifi.elasticsearch.ElasticSearchClientServiceImpl;
import org.apache.nifi.elasticsearch.ElasticsearchException;
import org.apache.nifi.elasticsearch.IndexOperationRequest;
import org.apache.nifi.elasticsearch.IndexOperationResponse;
+import org.apache.nifi.elasticsearch.MapBuilder;
import org.apache.nifi.elasticsearch.SearchResponse;
import org.apache.nifi.elasticsearch.TestControllerServiceProcessor;
import org.apache.nifi.elasticsearch.UpdateOperationResponse;
-import org.apache.nifi.security.util.StandardTlsConfiguration;
-import org.apache.nifi.security.util.TemporaryKeyStoreBuilder;
-import org.apache.nifi.security.util.TlsConfiguration;
-import org.apache.nifi.util.TestRunner;
-import org.apache.nifi.util.TestRunners;
-import org.junit.jupiter.api.AfterAll;
+import org.apache.nifi.ssl.SSLContextService;
+import org.apache.nifi.ssl.StandardRestrictedSSLContextService;
+import org.apache.nifi.ssl.StandardSSLContextService;
+import org.apache.nifi.util.MockConfigurationContext;
+import org.apache.nifi.util.MockControllerServiceLookup;
+import org.apache.nifi.util.MockVariableRegistry;
import org.junit.jupiter.api.AfterEach;
-import org.junit.jupiter.api.BeforeAll;
-import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Disabled;
import org.junit.jupiter.api.Test;
-import java.io.IOException;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.Collections;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
+import java.util.Objects;
+import java.util.Optional;
import static org.junit.jupiter.api.Assertions.assertEquals;
import static org.junit.jupiter.api.Assertions.assertFalse;
@@ -53,129 +57,109 @@ import static org.junit.jupiter.api.Assertions.assertNotNull;
import static org.junit.jupiter.api.Assertions.assertNull;
import static org.junit.jupiter.api.Assertions.assertThrows;
import static org.junit.jupiter.api.Assertions.assertTrue;
+import static org.junit.jupiter.api.Assertions.fail;
import static org.junit.jupiter.api.Assumptions.assumeTrue;
-public class ElasticSearchClientService_IT extends AbstractElasticsearch_IT {
- private TestRunner runner;
- private ElasticSearchClientServiceImpl service;
-
- static final String INDEX = "messages";
- public static String TYPE;
-
- private static TlsConfiguration generatedTlsConfiguration;
- private static TlsConfiguration truststoreTlsConfiguration;
-
- @BeforeAll
- static void beforeAll() throws IOException {
- startTestcontainer();
- TYPE = getElasticMajorVersion() == 6 ? "_doc" : "";
- System.out.println(
- String.format("%n%n%n%n%n%n%n%n%n%n%n%n%n%n%nTYPE: %s%n%s%s%n%n%n%n%n%n%n%n%n%n%n%n%n%n%n",
- TYPE, IMAGE.getRepository(), IMAGE.getVersionPart())
- );
-
- generatedTlsConfiguration = new TemporaryKeyStoreBuilder().build();
- truststoreTlsConfiguration = new StandardTlsConfiguration(
- null,
- null,
- null,
- generatedTlsConfiguration.getTruststorePath(),
- generatedTlsConfiguration.getTruststorePassword(),
- generatedTlsConfiguration.getTruststoreType()
- );
-
- setupTestData();
- }
-
- @AfterAll
- public static void afterAll() throws IOException {
- tearDownTestData();
- stopTestcontainer();
- }
-
- @BeforeEach
- void before() throws Exception {
- runner = TestRunners.newTestRunner(TestControllerServiceProcessor.class);
- service = new ElasticSearchClientServiceImpl();
- runner.addControllerService("Client Service", service);
- runner.setProperty(service, ElasticSearchClientService.HTTP_HOSTS, ELASTIC_HOST);
- runner.setProperty(service, ElasticSearchClientService.CONNECT_TIMEOUT, "10000");
- runner.setProperty(service, ElasticSearchClientService.SOCKET_TIMEOUT, "60000");
- runner.setProperty(service, ElasticSearchClientService.RETRY_TIMEOUT, "60000");
- runner.setProperty(service, ElasticSearchClientService.SUPPRESS_NULLS, ElasticSearchClientService.ALWAYS_SUPPRESS.getValue());
- runner.setProperty(service, ElasticSearchClientService.USERNAME, "elastic");
- runner.setProperty(service, ElasticSearchClientService.PASSWORD, ELASTIC_USER_PASSWORD);
-
- try {
- runner.enableControllerService(service);
- } catch (Exception ex) {
- throw ex;
- }
-
- service.refresh(null, null);
- }
-
+class ElasticSearchClientService_IT extends AbstractElasticsearch_IT {
@AfterEach
void after() throws Exception {
service.onDisabled();
}
- private String prettyJson(Object o) throws JsonProcessingException {
+ private String prettyJson(final Object o) throws JsonProcessingException {
return MAPPER.writerWithDefaultPrettyPrinter().writeValueAsString(o);
}
- private class MapBuilder {
- private Map<String, Object> toBuild;
+ private Map<PropertyDescriptor, String> getClientServiceProperties() {
+ return ((MockControllerServiceLookup) runner.getProcessContext().getControllerServiceLookup())
+ .getControllerServices().get(CLIENT_SERVICE_NAME).getProperties();
+ }
- public MapBuilder() {
- toBuild = new HashMap<>();
- }
+ @Test
+ void testVerifySuccess() {
+ final List<ConfigVerificationResult> results = ((VerifiableControllerService) service).verify(
+ new MockConfigurationContext(service, getClientServiceProperties(), runner.getProcessContext().getControllerServiceLookup(), new MockVariableRegistry()),
+ runner.getLogger(),
+ Collections.emptyMap()
+ );
+ assertEquals(3, results.size());
+ assertEquals(3, results.stream().filter(result -> result.getOutcome() == ConfigVerificationResult.Outcome.SUCCESSFUL).count(), results.toString());
+ }
- public MapBuilder of(String key, Object value) {
- toBuild.put(key, value);
- return this;
- }
+ @Test
+ void testVerifyFailedURL() {
+ runner.disableControllerService(service);
+ runner.setProperty(service, ElasticSearchClientService.HTTP_HOSTS, "invalid");
- public MapBuilder of(String key, Object value, String key2, Object value2) {
- toBuild.put(key, value);
- toBuild.put(key2, value2);
- return this;
- }
+ final List<ConfigVerificationResult> results = ((VerifiableControllerService) service).verify(
+ new MockConfigurationContext(service, getClientServiceProperties(), runner.getProcessContext().getControllerServiceLookup(), new MockVariableRegistry()),
+ runner.getLogger(),
+ Collections.emptyMap()
+ );
+ assertEquals(3, results.size());
+ assertEquals(2, results.stream().filter(result -> result.getOutcome() == ConfigVerificationResult.Outcome.SKIPPED).count(), results.toString());
+ assertEquals(1, results.stream().filter(
+ result -> Objects.equals(result.getVerificationStepName(), ElasticSearchClientServiceImpl.VERIFICATION_STEP_CLIENT_SETUP)
+ && Objects.equals(result.getExplanation(), "Incorrect/invalid " + ElasticSearchClientService.HTTP_HOSTS.getDisplayName())
+ && result.getOutcome() == ConfigVerificationResult.Outcome.FAILED).count(),
+ results.toString()
+ );
+ }
- public MapBuilder of(String key, Object value, String key2, Object value2, String key3, Object value3) {
- toBuild.put(key, value);
- toBuild.put(key2, value2);
- toBuild.put(key3, value3);
- return this;
+ @Test
+ void testVerifyFailedSSL() throws Exception {
+ runner.disableControllerService(service);
+ final SSLContextService sslContextService = new StandardRestrictedSSLContextService();
+ runner.addControllerService("SSL Context", sslContextService);
+ runner.setProperty(service, ElasticSearchClientService.PROP_SSL_CONTEXT_SERVICE, "SSL Context");
+ runner.setProperty(sslContextService, StandardSSLContextService.TRUSTSTORE, "not/a/file");
+ runner.setProperty(sslContextService, StandardSSLContextService.TRUSTSTORE_PASSWORD, "ignored");
+ try {
+ runner.enableControllerService(sslContextService);
+ } catch (final Exception ignored) {
+ // expected, ignore
}
- public MapBuilder of(String key, Object value, String key2, Object value2, String key3, Object value3,
- String key4, Object value4) {
- toBuild.put(key, value);
- toBuild.put(key2, value2);
- toBuild.put(key3, value3);
- toBuild.put(key4, value4);
- return this;
- }
+ final List<ConfigVerificationResult> results = ((VerifiableControllerService) service).verify(
+ new MockConfigurationContext(service, getClientServiceProperties(), runner.getProcessContext().getControllerServiceLookup(), new MockVariableRegistry()),
+ runner.getLogger(),
+ Collections.emptyMap()
+ );
+ assertEquals(3, results.size());
+ assertEquals(2, results.stream().filter(result -> result.getOutcome() == ConfigVerificationResult.Outcome.SKIPPED).count(), results.toString());
+ assertEquals(1, results.stream().filter(
+ result -> Objects.equals(result.getVerificationStepName(), ElasticSearchClientServiceImpl.VERIFICATION_STEP_CLIENT_SETUP)
+ && Objects.equals(result.getExplanation(), "Incorrect/invalid " + ElasticSearchClientService.PROP_SSL_CONTEXT_SERVICE.getDisplayName())
+ && result.getOutcome() == ConfigVerificationResult.Outcome.FAILED).count(),
+ results.toString()
+ );
+ }
- public MapBuilder of(String key, Object value, String key2, Object value2, String key3, Object value3,
- String key4, Object value4, String key5, Object value5) {
- toBuild.put(key, value);
- toBuild.put(key2, value2);
- toBuild.put(key3, value3);
- toBuild.put(key4, value4);
- toBuild.put(key5, value5);
- return this;
- }
+ @Test
+ void testVerifyFailedAuth() {
+ runner.disableControllerService(service);
+ runner.setProperty(service, ElasticSearchClientService.USERNAME, "invalid");
+ runner.setProperty(service, ElasticSearchClientService.PASSWORD, "not-real");
- public Map<String, Object> build() {
- return toBuild;
- }
+ final List<ConfigVerificationResult> results = ((VerifiableControllerService) service).verify(
+ new MockConfigurationContext(service, getClientServiceProperties(), runner.getProcessContext().getControllerServiceLookup(), new MockVariableRegistry()),
+ runner.getLogger(),
+ Collections.emptyMap()
+ );
+ assertEquals(3, results.size());
+ assertEquals(1, results.stream().filter(result -> result.getOutcome() == ConfigVerificationResult.Outcome.SUCCESSFUL).count(), results.toString());
+ assertEquals(1, results.stream().filter(result -> result.getOutcome() == ConfigVerificationResult.Outcome.SKIPPED).count(), results.toString());
+ assertEquals(1, results.stream().filter(
+ result -> Objects.equals(result.getVerificationStepName(), ElasticSearchClientServiceImpl.VERIFICATION_STEP_CONNECTION)
+ && Objects.equals(result.getExplanation(), "Unable to retrieve system summary from Elasticsearch root endpoint")
+ && result.getOutcome() == ConfigVerificationResult.Outcome.FAILED).count(),
+ results.toString()
+ );
}
@Test
void testBasicSearch() throws Exception {
- Map<String, Object> temp = new MapBuilder()
+ final Map<String, Object> temp = new MapBuilder()
.of("size", 10, "query", new MapBuilder().of("match_all", new HashMap<>()).build(),
"aggs", new MapBuilder()
.of("term_counts", new MapBuilder()
@@ -185,9 +169,9 @@ public class ElasticSearchClientService_IT extends AbstractElasticsearch_IT {
.build())
.build())
.build();
- String query = prettyJson(temp);
+ final String query = prettyJson(temp);
- SearchResponse response = service.search(query, INDEX, TYPE, null);
+ final SearchResponse response = service.search(query, INDEX, type, null);
assertNotNull(response, "Response was null");
assertEquals(15, response.getNumberOfHits(), "Wrong count");
@@ -200,25 +184,25 @@ public class ElasticSearchClientService_IT extends AbstractElasticsearch_IT {
assertNull(response.getSearchAfter(), "Unexpected Search_After");
assertNull(response.getPitId(), "Unexpected pitId");
- Map termCounts = (Map) response.getAggregations().get("term_counts");
+ @SuppressWarnings("unchecked") final Map<String, Object> termCounts = (Map<String, Object>) response.getAggregations().get("term_counts");
assertNotNull(termCounts, "Term counts was missing");
- List<Map<String, Object>> buckets = (List<java.util.Map<String, Object>>) termCounts.get("buckets");
+ @SuppressWarnings("unchecked") final List<Map<String, Object>> buckets = (List<Map<String, Object>>) termCounts.get("buckets");
assertNotNull(buckets, "Buckets branch was empty");
- Map<String, Object> expected = new MapBuilder()
+ final Map<String, Object> expected = new MapBuilder()
.of("one", 1, "two", 2, "three", 3,
"four", 4, "five", 5)
.build();
buckets.forEach( aggRes -> {
- String key = (String) aggRes.get("key");
- Integer docCount = (Integer) aggRes.get("doc_count");
+ final String key = (String) aggRes.get("key");
+ final Integer docCount = (Integer) aggRes.get("doc_count");
assertEquals(expected.get(key), docCount, "${key} did not match.");
});
}
@Test
void testBasicSearchRequestParameters() throws Exception {
- Map<String, Object> temp = new MapBuilder()
+ final Map<String, Object> temp = new MapBuilder()
.of("size", 10, "query", new MapBuilder().of("match_all", new HashMap<>()).build(),
"aggs", new MapBuilder()
.of("term_counts", new MapBuilder()
@@ -228,10 +212,10 @@ public class ElasticSearchClientService_IT extends AbstractElasticsearch_IT {
.build())
.build())
.build();
- String query = prettyJson(temp);
+ final String query = prettyJson(temp);
- SearchResponse response = service.search(query, "messages", TYPE, createParameters("preference", "_local"));
+ final SearchResponse response = service.search(query, "messages", type, createParameters("preference", "_local"));
assertNotNull(response, "Response was null");
assertEquals(15, response.getNumberOfHits(), "Wrong count");
@@ -241,45 +225,52 @@ public class ElasticSearchClientService_IT extends AbstractElasticsearch_IT {
assertNotNull(response.getAggregations(), "Aggregations are missing");
assertEquals(1, response.getAggregations().size(), "Aggregation count is wrong");
- Map termCounts = (Map) response.getAggregations().get("term_counts");
+ @SuppressWarnings("unchecked") final Map<String, Object> termCounts = (Map<String, Object>) response.getAggregations().get("term_counts");
assertNotNull(termCounts, "Term counts was missing");
- List<Map<String, Object>> buckets = (List<Map<String, Object>>) termCounts.get("buckets");
+ @SuppressWarnings("unchecked") final List<Map<String, Object>> buckets = (List<Map<String, Object>>) termCounts.get("buckets");
assertNotNull(buckets, "Buckets branch was empty");
- Map<String, Object> expected = new MapBuilder()
+ final Map<String, Object> expected = new MapBuilder()
.of("one", 1, "two", 2, "three", 3,
"four", 4, "five", 5)
.build();
buckets.forEach( (aggRes) -> {
- String key = (String) aggRes.get("key");
- Integer docCount = (Integer) aggRes.get("doc_count");
+ final String key = (String) aggRes.get("key");
+ final Integer docCount = (Integer) aggRes.get("doc_count");
assertEquals(expected.get(key), docCount, String.format("%s did not match.", key));
});
}
@Test
- void testV5SearchWarnings() throws JsonProcessingException {
- int version = getElasticMajorVersion();
- assumeTrue(version == 5, "Requires version 5 (no search API deprecations yet for 8.x)");
-
- String query = prettyJson(new MapBuilder()
- .of("size", 1, "query", new MapBuilder().of("query_string", new MapBuilder()
- .of("query", 1, "all_fields", true).build()).build())
+ void testV6SearchWarnings() throws JsonProcessingException {
+ assumeTrue(getElasticMajorVersion() == 6, "Requires Elasticsearch 6");
+ final String query = prettyJson(new MapBuilder()
+ .of("size", 1,
+ "query", new MapBuilder().of("query_string",
+ new MapBuilder().of("query", 1, "all_fields", true).build()
+ ).build())
.build());
- final SearchResponse response = service.search(query, INDEX, TYPE, null);
- assertTrue(!response.getWarnings().isEmpty(), "Missing warnings");
+ final String type = "a-type";
+ final SearchResponse response = service.search(query, INDEX, type, null);
+ assertFalse(response.getWarnings().isEmpty(), "Missing warnings");
}
@Test
- public void testV7SearchWarnings() throws JsonProcessingException {
- assumeTrue(getElasticMajorVersion() == 7,
- "Requires Elasticsearch 7");
- String query = prettyJson(new MapBuilder()
+ void testV7SearchWarnings() throws JsonProcessingException {
+ assumeTrue(getElasticMajorVersion() == 7, "Requires Elasticsearch 7");
+ final String query = prettyJson(new MapBuilder()
.of("size", 1, "query", new MapBuilder().of("match_all", new HashMap<>()).build())
.build());
- String type = "a-type";
+ final String type = "a-type";
final SearchResponse response = service.search(query, INDEX, type, null);
- assertTrue(!response.getWarnings().isEmpty(), "Missing warnings");
+ assertFalse(response.getWarnings().isEmpty(), "Missing warnings");
+ }
+
+ @Disabled("Skip until Elasticsearch 8.x has a _search API deprecation")
+ @Test
+ void testV8SearchWarnings() {
+ assumeTrue(getElasticMajorVersion() == 8, "Requires Elasticsearch 8");
+ fail("Elasticsearch 8 search API deprecations not currently available for test");
}
@Test
@@ -296,7 +287,7 @@ public class ElasticSearchClientService_IT extends AbstractElasticsearch_IT {
.build());
// initiate the scroll
- final SearchResponse response = service.search(query, INDEX, TYPE, Collections.singletonMap("scroll", "10s"));
+ final SearchResponse response = service.search(query, INDEX, type, Collections.singletonMap("scroll", "10s"));
assertNotNull(response, "Response was null");
assertEquals(15, response.getNumberOfHits(), "Wrong count");
@@ -309,12 +300,13 @@ public class ElasticSearchClientService_IT extends AbstractElasticsearch_IT {
assertNull(response.getSearchAfter(), "Unexpected Search_After");
assertNull(response.getPitId(), "Unexpected pitId");
- final Map termCounts = (Map) response.getAggregations().get("term_counts");
+ @SuppressWarnings("unchecked") final Map<String, Object> termCounts = (Map<String, Object>) response.getAggregations().get("term_counts");
assertNotNull(termCounts, "Term counts was missing");
- assertEquals(5, ((List)termCounts.get("buckets")).size(), "Buckets count is wrong");
+ @SuppressWarnings("unchecked") final List<Map<String, Object>> buckets = (List<Map<String, Object>>) termCounts.get("buckets");
+ assertEquals(5, buckets.size(), "Buckets count is wrong");
// scroll the next page
- Map<String, String> parameters = createParameters("scroll_id", response.getScrollId(), "scroll", "10s");
+ final Map<String, String> parameters = createParameters("scroll_id", response.getScrollId(), "scroll", "10s");
final SearchResponse scrollResponse = service.scroll(prettyJson(parameters));
assertNotNull(scrollResponse, "Scroll Response was null");
@@ -328,7 +320,7 @@ public class ElasticSearchClientService_IT extends AbstractElasticsearch_IT {
assertNull(scrollResponse.getSearchAfter(), "Unexpected Search_After");
assertNull(scrollResponse.getPitId(), "Unexpected pitId");
- assertNotEquals(scrollResponse.getHits(), response.getHits(), () -> "Same results");
+ assertNotEquals(scrollResponse.getHits(), response.getHits(), "Same results");
// delete the scroll
DeleteOperationResponse deleteResponse = service.deleteScroll(scrollResponse.getScrollId());
@@ -351,14 +343,14 @@ public class ElasticSearchClientService_IT extends AbstractElasticsearch_IT {
.of("field", "msg", "size", 5)
.build())
.build()).build())
- .of("sort", Arrays.asList(
- new MapBuilder().of("msg", "desc").build()
+ .of("sort", Collections.singletonList(
+ new MapBuilder().of("msg", "desc").build()
))
.build();
final String query = prettyJson(queryMap);
// search first page
- final SearchResponse response = service.search(query, INDEX, TYPE, null);
+ final SearchResponse response = service.search(query, INDEX, type, null);
assertNotNull(response, "Response was null");
assertEquals(15, response.getNumberOfHits(), "Wrong count");
@@ -371,15 +363,17 @@ public class ElasticSearchClientService_IT extends AbstractElasticsearch_IT {
assertNotNull(response.getSearchAfter(), "Search_After missing");
assertNull(response.getPitId(), "Unexpected pitId");
- final Map termCounts = (Map)response.getAggregations().get("term_counts");
+ @SuppressWarnings("unchecked") final Map<String, Object> termCounts = (Map<String, Object>) response.getAggregations().get("term_counts");
assertNotNull(termCounts, "Term counts was missing");
- assertEquals(5, ((List)termCounts.get("buckets")).size(), "Buckets count is wrong");
+ @SuppressWarnings("unchecked") final List<Map<String, Object>> buckets = (List<Map<String, Object>>) termCounts.get("buckets");
+ assertEquals(5, buckets.size(), "Buckets count is wrong");
// search the next page
- queryMap.put("search_after", MAPPER.readValue(response.getSearchAfter(), List.class));
- queryMap.remove("aggs");
- final String secondPage = prettyJson(queryMap);
- final SearchResponse secondResponse = service.search(secondPage, INDEX, TYPE, null);
+ final Map<String, Object> page2QueryMap = new HashMap<>(queryMap);
+ page2QueryMap.put("search_after", MAPPER.readValue(response.getSearchAfter(), List.class));
+ page2QueryMap.remove("aggs");
+ final String secondPage = prettyJson(page2QueryMap);
+ final SearchResponse secondResponse = service.search(secondPage, INDEX, type, null);
assertNotNull(secondResponse, "Second Response was null");
assertEquals(15, secondResponse.getNumberOfHits(), "Wrong count");
@@ -398,8 +392,8 @@ public class ElasticSearchClientService_IT extends AbstractElasticsearch_IT {
@Test
void testPointInTime() throws JsonProcessingException {
// Point in Time only available in 7.10+ with XPack enabled
- double majorVersion = getElasticMajorVersion();
- double minorVersion = getElasticMinorVersion();
+ final double majorVersion = getElasticMajorVersion();
+ final double minorVersion = getElasticMinorVersion();
assumeTrue(majorVersion >= 8 || (majorVersion == 7 && minorVersion >= 10), "Requires version 7.10+");
// initialise
@@ -412,7 +406,7 @@ public class ElasticSearchClientService_IT extends AbstractElasticsearch_IT {
.of("field", "msg", "size", 5)
.build())
.build()).build())
- .of("sort", Arrays.asList(
+ .of("sort", Collections.singletonList(
new MapBuilder().of("msg", "desc").build()
))
.of("pit", new MapBuilder()
@@ -422,7 +416,7 @@ public class ElasticSearchClientService_IT extends AbstractElasticsearch_IT {
final String query = prettyJson(queryMap);
// search first page
- final SearchResponse response = service.search(query, null, TYPE, null);
+ final SearchResponse response = service.search(query, null, type, null);
assertNotNull(response, "Response was null");
assertEquals(15, response.getNumberOfHits(), "Wrong count");
@@ -435,15 +429,17 @@ public class ElasticSearchClientService_IT extends AbstractElasticsearch_IT {
assertNotNull(response.getSearchAfter(), "Unexpected Search_After");
assertNotNull(response.getPitId(), "pitId missing");
- final Map termCounts = (Map) response.getAggregations().get("term_counts");
+ @SuppressWarnings("unchecked") final Map<String, Object> termCounts = (Map<String, Object>) response.getAggregations().get("term_counts");
assertNotNull(termCounts, "Term counts was missing");
- assertEquals(5, ((List)termCounts.get("buckets")).size(), "Buckets count is wrong");
+ @SuppressWarnings("unchecked") final List<Map<String, Object>> buckets = (List<Map<String, Object>>) termCounts.get("buckets");
+ assertEquals(5, buckets.size(), "Buckets count is wrong");
// search the next page
- queryMap.put("search_after", MAPPER.readValue(response.getSearchAfter(), List.class));
- queryMap.remove("aggs");
- final String secondPage = prettyJson(queryMap);
- final SearchResponse secondResponse = service.search(secondPage, null, TYPE, null);
+ final Map<String, Object> page2QueryMap = new HashMap<>(queryMap);
+ page2QueryMap.put("search_after", MAPPER.readValue(response.getSearchAfter(), List.class));
+ page2QueryMap.remove("aggs");
+ final String secondPage = prettyJson(page2QueryMap);
+ final SearchResponse secondResponse = service.search(secondPage, null, type, null);
assertNotNull(secondResponse, "Second Response was null");
assertEquals(15, secondResponse.getNumberOfHits(), "Wrong count");
@@ -471,49 +467,49 @@ public class ElasticSearchClientService_IT extends AbstractElasticsearch_IT {
@Test
void testDeleteByQuery() throws Exception {
- String query = prettyJson(new MapBuilder()
+ final String query = prettyJson(new MapBuilder()
.of("query", new MapBuilder()
.of("match", new MapBuilder().of("msg", "five").build())
.build()).build());
- DeleteOperationResponse response = service.deleteByQuery(query, INDEX, TYPE, null);
+ final DeleteOperationResponse response = service.deleteByQuery(query, INDEX, type, null);
assertNotNull(response);
assertTrue(response.getTook() > 0);
}
@Test
void testDeleteByQueryRequestParameters() throws Exception {
- String query = prettyJson(new MapBuilder()
+ final String query = prettyJson(new MapBuilder()
.of("query", new MapBuilder()
.of("match", new MapBuilder().of("msg", "six").build())
.build()).build());
- Map<String, String> parameters = new HashMap<>();
+ final Map<String, String> parameters = new HashMap<>();
parameters.put("refresh", "true");
- DeleteOperationResponse response = service.deleteByQuery(query, INDEX, TYPE, parameters);
+ final DeleteOperationResponse response = service.deleteByQuery(query, INDEX, type, parameters);
assertNotNull(response);
assertTrue(response.getTook() > 0);
}
@Test
void testUpdateByQuery() throws Exception {
- String query = prettyJson(new MapBuilder()
+ final String query = prettyJson(new MapBuilder()
.of("query", new MapBuilder()
.of("match", new MapBuilder().of("msg", "four").build())
.build()).build());
- UpdateOperationResponse response = service.updateByQuery(query, INDEX, TYPE, null);
+ final UpdateOperationResponse response = service.updateByQuery(query, INDEX, type, null);
assertNotNull(response);
assertTrue(response.getTook() > 0);
}
@Test
void testUpdateByQueryRequestParameters() throws Exception {
- String query = prettyJson(new MapBuilder()
+ final String query = prettyJson(new MapBuilder()
.of("query", new MapBuilder()
.of("match", new MapBuilder().of("msg", "four").build())
.build()).build());
- Map<String, String> parameters = new HashMap<>();
+ final Map<String, String> parameters = new HashMap<>();
parameters.put("refresh", "true");
parameters.put("slices", "1");
- UpdateOperationResponse response = service.updateByQuery(query, INDEX, TYPE, parameters);
+ final UpdateOperationResponse response = service.updateByQuery(query, INDEX, type, parameters);
assertNotNull(response);
assertTrue(response.getTook() > 0);
}
@@ -521,19 +517,19 @@ public class ElasticSearchClientService_IT extends AbstractElasticsearch_IT {
@Test
void testDeleteById() throws Exception {
final String ID = "1";
- final Map<String, Object> originalDoc = service.get(INDEX, TYPE, ID, null);
+ final Map<String, Object> originalDoc = service.get(INDEX, type, ID, null);
try {
- DeleteOperationResponse response = service.deleteById(INDEX, TYPE, ID, null);
+ final DeleteOperationResponse response = service.deleteById(INDEX, type, ID, null);
assertNotNull(response);
assertTrue(response.getTook() > 0);
final ElasticsearchException ee = assertThrows(ElasticsearchException.class, () ->
- service.get(INDEX, TYPE, ID, null));
+ service.get(INDEX, type, ID, null));
assertTrue(ee.isNotFound());
- final Map<String, Object> doc = service.get(INDEX, TYPE, "2", null);
+ final Map<String, Object> doc = service.get(INDEX, type, "2", null);
assertNotNull(doc);
} finally {
// replace the deleted doc
- service.add(new IndexOperationRequest(INDEX, TYPE, "1", originalDoc, IndexOperationRequest.Operation.Index), null);
+ service.add(new IndexOperationRequest(INDEX, type, "1", originalDoc, IndexOperationRequest.Operation.Index), null);
waitForIndexRefresh(); // (affects later tests using _search or _bulk)
}
}
@@ -541,8 +537,8 @@ public class ElasticSearchClientService_IT extends AbstractElasticsearch_IT {
@Test
void testGet() {
for (int index = 1; index <= 15; index++) {
- String id = String.valueOf(index);
- Map<String, Object> doc = service.get(INDEX, TYPE, id, null);
+ final String id = String.valueOf(index);
+ final Map<String, Object> doc = service.get(INDEX, type, id, null);
assertNotNull(doc, "Doc was null");
assertNotNull(doc.get("msg"), "${doc.toString()}\t${doc.keySet().toString()}");
}
@@ -550,13 +546,19 @@ public class ElasticSearchClientService_IT extends AbstractElasticsearch_IT {
@Test
void testGetNotFound() {
- final ElasticsearchException ee = assertThrows(ElasticsearchException.class, () -> service.get(INDEX, TYPE, "not_found", null));
+ final ElasticsearchException ee = assertThrows(ElasticsearchException.class, () -> service.get(INDEX, type, "not_found", null));
assertTrue(ee.isNotFound());
}
+ @Test
+ void testExists() {
+ assertTrue(service.exists(INDEX, null), "index does not exist");
+ assertFalse(service.exists("index-does-not-exist", null), "index exists");
+ }
+
@Test
void testNullSuppression() throws InterruptedException {
- Map<String, Object> doc = new HashMap<>();
+ final Map<String, Object> doc = new HashMap<>();
doc.put("msg", "test");
doc.put("is_null", null);
doc.put("is_empty", "");
@@ -566,27 +568,27 @@ public class ElasticSearchClientService_IT extends AbstractElasticsearch_IT {
// index with nulls
suppressNulls(false);
- IndexOperationResponse response = service.bulk(Arrays.asList(new IndexOperationRequest("nulls", TYPE, "1", doc, IndexOperationRequest.Operation.Index)), null);
+ IndexOperationResponse response = service.bulk(Collections.singletonList(new IndexOperationRequest("nulls", type, "1", doc, IndexOperationRequest.Operation.Index)), null);
assertNotNull(response);
assertTrue(response.getTook() > 0);
waitForIndexRefresh();
- Map<String, Object> result = service.get("nulls", TYPE, "1", null);
+ Map<String, Object> result = service.get("nulls", type, "1", null);
assertEquals(doc, result);
// suppress nulls
suppressNulls(true);
- response = service.bulk(Arrays.asList(new IndexOperationRequest("nulls", TYPE, "2", doc, IndexOperationRequest.Operation.Index)), null);
+ response = service.bulk(Collections.singletonList(new IndexOperationRequest("nulls", type, "2", doc, IndexOperationRequest.Operation.Index)), null);
assertNotNull(response);
assertTrue(response.getTook() > 0);
waitForIndexRefresh();
- result = service.get("nulls", TYPE, "2", null);
+ result = service.get("nulls", type, "2", null);
assertTrue(result.keySet().containsAll(Arrays.asList("msg", "is_blank")), "Non-nulls (present): " + result);
- assertFalse(result.keySet().contains("is_null"), "is_null (should be omitted): " + result);
- assertFalse(result.keySet().contains("is_empty"), "is_empty (should be omitted): " + result);
- assertFalse(result.keySet().contains("empty_nested"), "empty_nested (should be omitted): " + result);
- assertFalse(result.keySet().contains("empty_array"), "empty_array (should be omitted): " + result);
+ assertFalse(result.containsKey("is_null"), "is_null (should be omitted): " + result);
+ assertFalse(result.containsKey("is_empty"), "is_empty (should be omitted): " + result);
+ assertFalse(result.containsKey("empty_nested"), "empty_nested (should be omitted): " + result);
+ assertFalse(result.containsKey("empty_array"), "empty_array (should be omitted): " + result);
}
private void suppressNulls(final boolean suppressNulls) {
@@ -601,19 +603,19 @@ public class ElasticSearchClientService_IT extends AbstractElasticsearch_IT {
@Test
void testBulkAddTwoIndexes() throws Exception {
- List<IndexOperationRequest> payload = new ArrayList<>();
+ final List<IndexOperationRequest> payload = new ArrayList<>();
for (int x = 0; x < 20; x++) {
- String index = x % 2 == 0 ? "bulk_a": "bulk_b";
- payload.add(new IndexOperationRequest(index, TYPE, String.valueOf(x), new HashMap<String, Object>(){{
+ final String index = x % 2 == 0 ? "bulk_a": "bulk_b";
+ payload.add(new IndexOperationRequest(index, type, String.valueOf(x), new HashMap<String, Object>(){{
put("msg", "test");
}}, IndexOperationRequest.Operation.Index));
}
for (int x = 0; x < 5; x++) {
- payload.add(new IndexOperationRequest("bulk_c", TYPE, String.valueOf(x), new HashMap<String, Object>(){{
+ payload.add(new IndexOperationRequest("bulk_c", type, String.valueOf(x), new HashMap<String, Object>(){{
put("msg", "test");
}}, IndexOperationRequest.Operation.Index));
}
- IndexOperationResponse response = service.bulk(payload, createParameters("refresh", "true"));
+ final IndexOperationResponse response = service.bulk(payload, createParameters("refresh", "true"));
assertNotNull(response);
assertTrue(response.getTook() > 0);
waitForIndexRefresh();
@@ -621,10 +623,10 @@ public class ElasticSearchClientService_IT extends AbstractElasticsearch_IT {
/*
* Now, check to ensure that both indexes got populated appropriately.
*/
- String query = "{ \"query\": { \"match_all\": {}}}";
- Long indexA = service.count(query, "bulk_a", TYPE, null);
- Long indexB = service.count(query, "bulk_b", TYPE, null);
- Long indexC = service.count(query, "bulk_c", TYPE, null);
+ final String query = "{ \"query\": { \"match_all\": {}}}";
+ final Long indexA = service.count(query, "bulk_a", type, null);
+ final Long indexB = service.count(query, "bulk_b", type, null);
+ final Long indexC = service.count(query, "bulk_c", type, null);
assertNotNull(indexA);
assertNotNull(indexB);
@@ -634,32 +636,32 @@ public class ElasticSearchClientService_IT extends AbstractElasticsearch_IT {
assertEquals(10, indexB.intValue());
assertEquals(5, indexC.intValue());
- Long total = service.count(query, "bulk_*", TYPE, null);
+ final Long total = service.count(query, "bulk_*", type, null);
assertNotNull(total);
assertEquals(25, total.intValue());
}
@Test
- void testBulkRequestParameters() throws Exception {
- List<IndexOperationRequest> payload = new ArrayList<>();
+ void testBulkRequestParameters() {
+ final List<IndexOperationRequest> payload = new ArrayList<>();
for (int x = 0; x < 20; x++) {
- String index = x % 2 == 0 ? "bulk_a": "bulk_b";
- payload.add(new IndexOperationRequest(index, TYPE, String.valueOf(x), new MapBuilder().of("msg", "test").build(), IndexOperationRequest.Operation.Index));
+ final String index = x % 2 == 0 ? "bulk_a": "bulk_b";
+ payload.add(new IndexOperationRequest(index, type, String.valueOf(x), new MapBuilder().of("msg", "test").build(), IndexOperationRequest.Operation.Index));
}
for (int x = 0; x < 5; x++) {
- payload.add(new IndexOperationRequest("bulk_c", TYPE, String.valueOf(x), new MapBuilder().of("msg", "test").build(), IndexOperationRequest.Operation.Index));
+ payload.add(new IndexOperationRequest("bulk_c", type, String.valueOf(x), new MapBuilder().of("msg", "test").build(), IndexOperationRequest.Operation.Index));
}
- IndexOperationResponse response = service.bulk(payload, createParameters("refresh", "true"));
+ final IndexOperationResponse response = service.bulk(payload, createParameters("refresh", "true"));
assertNotNull(response);
assertTrue(response.getTook() > 0);
/*
* Now, check to ensure that both indexes got populated and refreshed appropriately.
*/
- String query = "{ \"query\": { \"match_all\": {}}}";
- Long indexA = service.count(query, "bulk_a", TYPE, null);
- Long indexB = service.count(query, "bulk_b", TYPE, null);
- Long indexC = service.count(query, "bulk_c", TYPE, null);
+ final String query = "{ \"query\": { \"match_all\": {}}}";
+ final Long indexA = service.count(query, "bulk_a", type, null);
+ final Long indexB = service.count(query, "bulk_b", type, null);
+ final Long indexC = service.count(query, "bulk_c", type, null);
assertNotNull(indexA);
assertNotNull(indexB);
@@ -669,7 +671,7 @@ public class ElasticSearchClientService_IT extends AbstractElasticsearch_IT {
assertEquals(10, indexB.intValue());
assertEquals(5, indexC.intValue());
- Long total = service.count(query, "bulk_*", TYPE, null);
+ final Long total = service.count(query, "bulk_*", type, null);
assertNotNull(total);
assertEquals(25, total.intValue());
}
@@ -677,66 +679,72 @@ public class ElasticSearchClientService_IT extends AbstractElasticsearch_IT {
@Test
void testUpdateAndUpsert() throws InterruptedException {
final String TEST_ID = "update-test";
- Map<String, Object> doc = new HashMap<>();
+ final Map<String, Object> doc = new HashMap<>();
doc.put("msg", "Buongiorno, mondo");
- service.add(new IndexOperationRequest(INDEX, TYPE, TEST_ID, doc, IndexOperationRequest.Operation.Index), createParameters("refresh", "true"));
- Map<String, Object> result = service.get(INDEX, TYPE, TEST_ID, null);
+ service.add(new IndexOperationRequest(INDEX, type, TEST_ID, doc, IndexOperationRequest.Operation.Index), createParameters("refresh", "true"));
+ Map<String, Object> result = service.get(INDEX, type, TEST_ID, null);
assertEquals(doc, result, "Not the same");
- Map<String, Object> updates = new HashMap<>();
+ final Map<String, Object> updates = new HashMap<>();
updates.put("from", "john.smith");
- Map<String, Object> merged = new HashMap<>();
+ final Map<String, Object> merged = new HashMap<>();
merged.putAll(updates);
merged.putAll(doc);
- IndexOperationRequest request = new IndexOperationRequest(INDEX, TYPE, TEST_ID, updates, IndexOperationRequest.Operation.Update);
+ IndexOperationRequest request = new IndexOperationRequest(INDEX, type, TEST_ID, updates, IndexOperationRequest.Operation.Update);
service.add(request, createParameters("refresh", "true"));
- result = service.get(INDEX, TYPE, TEST_ID, null);
+ result = service.get(INDEX, type, TEST_ID, null);
assertTrue(result.containsKey("from"));
assertTrue(result.containsKey("msg"));
assertEquals(merged, result, "Not the same after update.");
final String UPSERTED_ID = "upsert-ftw";
- Map<String, Object> upsertItems = new HashMap<>();
+ final Map<String, Object> upsertItems = new HashMap<>();
upsertItems.put("upsert_1", "hello");
upsertItems.put("upsert_2", 1);
upsertItems.put("upsert_3", true);
- request = new IndexOperationRequest(INDEX, TYPE, UPSERTED_ID, upsertItems, IndexOperationRequest.Operation.Upsert);
+ request = new IndexOperationRequest(INDEX, type, UPSERTED_ID, upsertItems, IndexOperationRequest.Operation.Upsert);
service.add(request, createParameters("refresh", "true"));
- result = service.get(INDEX, TYPE, UPSERTED_ID, null);
+ result = service.get(INDEX, type, UPSERTED_ID, null);
assertEquals(upsertItems, result);
- List<IndexOperationRequest> deletes = new ArrayList<>();
- deletes.add(new IndexOperationRequest(INDEX, TYPE, TEST_ID, null, IndexOperationRequest.Operation.Delete));
- deletes.add(new IndexOperationRequest(INDEX, TYPE, UPSERTED_ID, null, IndexOperationRequest.Operation.Delete));
+ final List<IndexOperationRequest> deletes = new ArrayList<>();
+ deletes.add(new IndexOperationRequest(INDEX, type, TEST_ID, null, IndexOperationRequest.Operation.Delete));
+ deletes.add(new IndexOperationRequest(INDEX, type, UPSERTED_ID, null, IndexOperationRequest.Operation.Delete));
assertFalse(service.bulk(deletes, createParameters("refresh", "true")).hasErrors());
waitForIndexRefresh(); // wait 1s for index refresh (doesn't prevent GET but affects later tests using _search or _bulk)
- ElasticsearchException ee = assertThrows(ElasticsearchException.class, () -> service.get(INDEX, TYPE, TEST_ID, null) );
+ ElasticsearchException ee = assertThrows(ElasticsearchException.class, () -> service.get(INDEX, type, TEST_ID, null) );
assertTrue(ee.isNotFound());
- ee = assertThrows(ElasticsearchException.class, () -> service.get(INDEX, TYPE, UPSERTED_ID, null));
+ ee = assertThrows(ElasticsearchException.class, () -> service.get(INDEX, type, UPSERTED_ID, null));
assertTrue(ee.isNotFound());
}
+ @SuppressWarnings("unchecked")
@Test
void testGetBulkResponsesWithErrors() {
- List ops = Arrays.asList(
- new IndexOperationRequest(INDEX, TYPE, "1", new MapBuilder().of("msg", "one", "intField", 1).build(), IndexOperationRequest.Operation.Index), // OK
- new IndexOperationRequest(INDEX, TYPE, "2", new MapBuilder().of("msg", "two", "intField", 1).build(), IndexOperationRequest.Operation.Create), // already exists
- new IndexOperationRequest(INDEX, TYPE, "1", new MapBuilder().of("msg", "one", "intField", "notaninteger").build(), IndexOperationRequest.Operation.Index) // can't parse int field
+ final List<IndexOperationRequest> ops = Arrays.asList(
+ new IndexOperationRequest(INDEX, type, "1", new MapBuilder().of("msg", "one", "intField", 1).build(), IndexOperationRequest.Operation.Index), // OK
+ new IndexOperationRequest(INDEX, type, "2", new MapBuilder().of("msg", "two", "intField", 1).build(), IndexOperationRequest.Operation.Create), // already exists
+ new IndexOperationRequest(INDEX, type, "1", new MapBuilder().of("msg", "one", "intField", "notaninteger").build(), IndexOperationRequest.Operation.Index) // can't parse int field
);
- IndexOperationResponse response = service.bulk(ops, createParameters("refresh", "true"));
+ final IndexOperationResponse response = service.bulk(ops, createParameters("refresh", "true"));
assertTrue(response.hasErrors());
assertEquals(2, response.getItems().stream().filter(it -> {
- String key = it.keySet().stream().findFirst().get();
- return ((Map<String, Object>)it.get(key)).containsKey("error");
+ final Optional<String> first = it.keySet().stream().findFirst();
+ if (first.isPresent()) {
+ final String key = first.get();
+ return ((Map<String, Object>) it.get(key)).containsKey("error");
+ } else {
+ throw new IllegalStateException("Cannot find index response operation items");
+ }
}).count());
}
- private Map<String, String> createParameters(String... extra) {
+ private Map<String, String> createParameters(final String... extra) {
if (extra.length % 2 == 1) { //Putting this here to help maintainers catch stupid bugs before they happen
throw new RuntimeException("createParameters must have an even number of String parameters.");
}
- Map<String, String> parameters = new HashMap<>();
+ final Map<String, String> parameters = new HashMap<>();
for (int index = 0; index < extra.length; index += 2) {
parameters.put(extra[index], extra[index + 1]);
}
diff --git a/nifi-nar-bundles/nifi-elasticsearch-bundle/nifi-elasticsearch-client-service/src/test/java/org/apache/nifi/elasticsearch/integration/ElasticSearchLookupService_IT.java b/nifi-nar-bundles/nifi-elasticsearch-bundle/nifi-elasticsearch-client-service/src/test/java/org/apache/nifi/elasticsearch/integration/ElasticSearchLookupService_IT.java
index 6cb061903f..d4dd22b012 100644
--- a/nifi-nar-bundles/nifi-elasticsearch-bundle/nifi-elasticsearch-client-service/src/test/java/org/apache/nifi/elasticsearch/integration/ElasticSearchLookupService_IT.java
+++ b/nifi-nar-bundles/nifi-elasticsearch-bundle/nifi-elasticsearch-client-service/src/test/java/org/apache/nifi/elasticsearch/integration/ElasticSearchLookupService_IT.java
@@ -14,31 +14,25 @@
* See the License for the specific language governing permissions and
* limitations under the License.
*/
-
package org.apache.nifi.elasticsearch.integration;
-import org.apache.nifi.elasticsearch.ElasticSearchClientService;
-import org.apache.nifi.elasticsearch.ElasticSearchClientServiceImpl;
import org.apache.nifi.elasticsearch.ElasticSearchLookupService;
import org.apache.nifi.elasticsearch.TestControllerServiceProcessor;
import org.apache.nifi.elasticsearch.TestSchemaRegistry;
import org.apache.nifi.lookup.LookupFailureException;
+import org.apache.nifi.record.path.FieldValue;
import org.apache.nifi.record.path.RecordPath;
import org.apache.nifi.record.path.RecordPathResult;
import org.apache.nifi.schema.access.SchemaAccessUtils;
import org.apache.nifi.schemaregistry.services.SchemaRegistry;
import org.apache.nifi.serialization.record.MapRecord;
import org.apache.nifi.serialization.record.Record;
+import org.apache.nifi.serialization.record.RecordField;
import org.apache.nifi.serialization.record.RecordSchema;
import org.apache.nifi.serialization.record.type.RecordDataType;
-import org.apache.nifi.util.TestRunner;
-import org.apache.nifi.util.TestRunners;
-import org.junit.jupiter.api.AfterAll;
-import org.junit.jupiter.api.BeforeAll;
import org.junit.jupiter.api.BeforeEach;
import org.junit.jupiter.api.Test;
-import java.io.IOException;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
@@ -49,54 +43,29 @@ import static org.junit.jupiter.api.Assertions.assertEquals;
import static org.junit.jupiter.api.Assertions.assertNotNull;
import static org.junit.jupiter.api.Assertions.assertTrue;
-public class ElasticSearchLookupService_IT extends AbstractElasticsearch_IT {
- private TestRunner runner;
- private ElasticSearchClientService service;
- private ElasticSearchLookupService lookupService;
-
- @BeforeAll
- public static void beforeAll() throws IOException {
- startTestcontainer();
- setupTestData();
- }
+class ElasticSearchLookupService_IT extends AbstractElasticsearch_IT {
+ private static final String LOOKUP_SERVICE_NAME = "Lookup Service";
- @AfterAll
- public static void afterAll() throws IOException {
- tearDownTestData();
- stopTestcontainer();
- }
+ private ElasticSearchLookupService lookupService;
@BeforeEach
void before() throws Exception {
- runner = TestRunners.newTestRunner(TestControllerServiceProcessor.class);
- service = new ElasticSearchClientServiceImpl();
+ super.before();
+
lookupService = new ElasticSearchLookupService();
- runner.addControllerService("Client Service", service);
- runner.addControllerService("Lookup Service", lookupService);
- runner.setProperty(service, ElasticSearchClientService.HTTP_HOSTS, ELASTIC_HOST);
- runner.setProperty(service, ElasticSearchClientService.CONNECT_TIMEOUT, "10000");
- runner.setProperty(service, ElasticSearchClientService.SOCKET_TIMEOUT, "60000");
- runner.setProperty(service, ElasticSearchClientService.USERNAME, "elastic");
- runner.setProperty(service, ElasticSearchClientService.PASSWORD, ELASTIC_USER_PASSWORD);
- runner.setProperty(TestControllerServiceProcessor.CLIENT_SERVICE, "Client Service");
- runner.setProperty(TestControllerServiceProcessor.LOOKUP_SERVICE, "Lookup Service");
- runner.setProperty(lookupService, ElasticSearchLookupService.CLIENT_SERVICE, "Client Service");
+ runner.addControllerService(LOOKUP_SERVICE_NAME, lookupService);
+ runner.setProperty(TestControllerServiceProcessor.CLIENT_SERVICE, CLIENT_SERVICE_NAME);
+ runner.setProperty(TestControllerServiceProcessor.LOOKUP_SERVICE, LOOKUP_SERVICE_NAME);
+ runner.setProperty(lookupService, ElasticSearchLookupService.CLIENT_SERVICE, CLIENT_SERVICE_NAME);
runner.setProperty(lookupService, ElasticSearchLookupService.INDEX, "user_details");
setTypeOnLookupService();
- try {
- runner.enableControllerService(service);
- runner.enableControllerService(lookupService);
- } catch (Exception ex) {
- throw ex;
- }
-
- service.refresh(null, null);
+ runner.enableControllerService(lookupService);
}
void setTypeOnLookupService() {
if (getElasticMajorVersion() == 6) {
- runner.setProperty(lookupService, ElasticSearchLookupService.TYPE, ElasticSearchClientService_IT.TYPE);
+ runner.setProperty(lookupService, ElasticSearchLookupService.TYPE, type);
} else {
runner.removeProperty(lookupService, ElasticSearchLookupService.TYPE);
}
@@ -110,7 +79,7 @@ public class ElasticSearchLookupService_IT extends AbstractElasticsearch_IT {
private void setDefaultSchema() throws Exception {
runner.disableControllerService(lookupService);
- SchemaRegistry registry = new TestSchemaRegistry();
+ final SchemaRegistry registry = new TestSchemaRegistry();
runner.addControllerService("registry", registry);
runner.setProperty(lookupService, SchemaAccessUtils.SCHEMA_REGISTRY, "registry");
runner.enableControllerService(registry);
@@ -119,13 +88,13 @@ public class ElasticSearchLookupService_IT extends AbstractElasticsearch_IT {
@Test
void lookupById() throws LookupFailureException {
- Map<String, Object> coordinates = new HashMap<>();
+ final Map<String, Object> coordinates = new HashMap<>();
coordinates.put("_id", "2");
- Optional<Record> result = lookupService.lookup(coordinates);
+ final Optional<Record> result = lookupService.lookup(coordinates);
assertNotNull(result);
assertTrue(result.isPresent());
- Record record = result.get();
+ final Record record = result.get();
assertEquals("jane.doe@company.com", record.getAsString("email"));
assertEquals("098-765-4321", record.getAsString("phone"));
assertEquals("GHIJK", record.getAsString("accessKey"));
@@ -133,7 +102,7 @@ public class ElasticSearchLookupService_IT extends AbstractElasticsearch_IT {
@Test
void testInvalidIdScenarios() {
- List<Map<String, Object>> coordinates = new ArrayList<>();
+ final List<Map<String, Object>> coordinates = new ArrayList<>();
Map<String, Object> temp = new HashMap<>();
temp.put("_id", 1);
coordinates.add(temp);
@@ -142,12 +111,12 @@ public class ElasticSearchLookupService_IT extends AbstractElasticsearch_IT {
temp.put("email", "john.smith@company.com");
coordinates.add(temp);
- for (Map<String, Object> coordinate: coordinates) {
+ for (final Map<String, Object> coordinate: coordinates) {
Exception exception = null;
try {
lookupService.lookup(coordinate);
- } catch (Exception ex) {
+ } catch (final Exception ex) {
exception = ex;
}
@@ -158,14 +127,14 @@ public class ElasticSearchLookupService_IT extends AbstractElasticsearch_IT {
@Test
void lookupByQuery() throws LookupFailureException {
- Map<String, Object> coordinates = new HashMap<>();
+ final Map<String, Object> coordinates = new HashMap<>();
coordinates.put("phone", "098-765-4321");
coordinates.put("email", "jane.doe@company.com");
- Optional<Record> result = lookupService.lookup(coordinates);
+ final Optional<Record> result = lookupService.lookup(coordinates);
assertNotNull(result);
assertTrue(result.isPresent());
- Record record = result.get();
+ final Record record = result.get();
assertEquals("jane.doe@company.com", record.getAsString("email"));
assertEquals("098-765-4321", record.getAsString("phone"));
assertEquals("GHIJK", record.getAsString("accessKey"));
@@ -173,7 +142,7 @@ public class ElasticSearchLookupService_IT extends AbstractElasticsearch_IT {
@Test
void testNestedSchema() throws LookupFailureException {
- Map<String, Object> coordinates = new HashMap<>();
+ final Map<String, Object> coordinates = new HashMap<>();
coordinates.put("subField.deeper.deepest.super_secret", "The sky is blue");
coordinates.put("subField.deeper.secretz", "Buongiorno, mondo!!");
coordinates.put("msg", "Hello, world");
@@ -183,16 +152,16 @@ public class ElasticSearchLookupService_IT extends AbstractElasticsearch_IT {
setTypeOnLookupService();
runner.enableControllerService(lookupService);
- Optional<Record> response = lookupService.lookup(coordinates);
+ final Optional<Record> response = lookupService.lookup(coordinates);
assertNotNull(response);
assertTrue(response.isPresent());
- Record rec = response.get();
+ final Record rec = response.get();
assertEquals("Hello, world", rec.getValue("msg"));
- Record subRec = getSubRecord(rec, "subField");
+ final Record subRec = getSubRecord(rec, "subField");
assertNotNull(subRec);
- Record deeper = getSubRecord(subRec, "deeper");
+ final Record deeper = getSubRecord(subRec, "deeper");
assertNotNull(deeper);
- Record deepest = getSubRecord(deeper, "deepest");
+ final Record deepest = getSubRecord(deeper, "deepest");
assertNotNull(deepest);
assertEquals("The sky is blue", deepest.getAsString("super_secret"));
}
@@ -203,19 +172,21 @@ public class ElasticSearchLookupService_IT extends AbstractElasticsearch_IT {
runner.setProperty(lookupService, ElasticSearchLookupService.INDEX, "complex");
setTypeOnLookupService();
runner.enableControllerService(lookupService);
- Map<String, Object> coordinates = new HashMap<>();
+ final Map<String, Object> coordinates = new HashMap<>();
coordinates.put("_id", "1");
- Optional<Record> response = lookupService.lookup(coordinates);
+ final Optional<Record> response = lookupService.lookup(coordinates);
assertNotNull(response);
assertTrue(response.isPresent());
- Record rec = response.get();
- Record subRec = getSubRecord(rec, "subField");
+ final Record rec = response.get();
+ final Record subRec = getSubRecord(rec, "subField");
- Record r2 = new MapRecord(rec.getSchema(), new HashMap<>());
- RecordPath path = RecordPath.compile("/subField/longField");
- RecordPathResult result = path.evaluate(r2);
- result.getSelectedFields().findFirst().get().updateValue(1234567890L);
+ final Record r2 = new MapRecord(rec.getSchema(), new HashMap<>());
+ final RecordPath path = RecordPath.compile("/subField/longField");
+ final RecordPathResult result = path.evaluate(r2);
+ final Optional<FieldValue> fieldValue = result.getSelectedFields().findFirst();
+ assertTrue(fieldValue.isPresent());
+ fieldValue.get().updateValue(1234567890L);
assertNotNull(rec);
assertNotNull(subRec);
@@ -225,10 +196,15 @@ public class ElasticSearchLookupService_IT extends AbstractElasticsearch_IT {
assertEquals("2018-04-10T12:18:05Z", subRec.getValue("dateField"));
}
- public static Record getSubRecord(Record rec, String fieldName) {
- RecordSchema schema = rec.getSchema();
- RecordSchema subSchema = ((RecordDataType)schema.getField(fieldName).get().getDataType()).getChildSchema();
- return rec.getAsRecord(fieldName, subSchema);
+ static Record getSubRecord(final Record rec, final String fieldName) {
+ final RecordSchema schema = rec.getSchema();
+ final Optional<RecordField> recordField = schema.getField(fieldName);
+ if (recordField.isPresent()) {
+ final RecordSchema subSchema = ((RecordDataType) recordField.get().getDataType()).getChildSchema();
+ return rec.getAsRecord(fieldName, subSchema);
+ } else {
+ throw new IllegalStateException(String.format("Field %s not present in Record", fieldName));
+ }
}
@Test
@@ -240,17 +216,17 @@ public class ElasticSearchLookupService_IT extends AbstractElasticsearch_IT {
setTypeOnLookupService();
runner.enableControllerService(lookupService);
- Map<String, Object> coordinates = new HashMap<>();
+ final Map<String, Object> coordinates = new HashMap<>();
coordinates.put("msg", "Hello, world");
- Optional<Record> result = lookupService.lookup(coordinates);
+ final Optional<Record> result = lookupService.lookup(coordinates);
assertTrue(result.isPresent());
- Record rec = result.get();
- Map<String, Object> entries = new HashMap<>();
+ final Record rec = result.get();
+ final Map<String, Object> entries = new HashMap<>();
entries.put("dateField2", "2018-08-14T10:08:00Z");
entries.put("longField2", 150000);
- entries.entrySet().forEach( (field) -> {
- Object value = rec.getValue(field.getKey());
- assertEquals(field.getValue(), value);
+ entries.forEach((key, value1) -> {
+ final Object value = rec.getValue(key);
+ assertEquals(value1, value);
});
}
}
diff --git a/nifi-nar-bundles/nifi-elasticsearch-bundle/nifi-elasticsearch-client-service/src/test/resources/conf-5/jvm.options b/nifi-nar-bundles/nifi-elasticsearch-bundle/nifi-elasticsearch-client-service/src/test/resources/conf-5/jvm.options
deleted file mode 100644
index a1fae3beef..0000000000
--- a/nifi-nar-bundles/nifi-elasticsearch-bundle/nifi-elasticsearch-client-service/src/test/resources/conf-5/jvm.options
+++ /dev/null
@@ -1,127 +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.
-##
-
-## JVM configuration
-
-################################################################
-## IMPORTANT: JVM heap size
-################################################################
-##
-## You should always set the min and max JVM heap
-## size to the same value. For example, to set
-## the heap to 4 GB, set:
-##
-## -Xms4g
-## -Xmx4g
-##
-## See https://www.elastic.co/guide/en/elasticsearch/reference/current/heap-size.html
-## for more information
-##
-################################################################
-
-# Xms represents the initial size of total heap space
-# Xmx represents the maximum size of total heap space
-
--Xms2g
--Xmx2g
-
-################################################################
-## Expert settings
-################################################################
-##
-## All settings below this section are considered
-## expert settings. Don't tamper with them unless
-## you understand what you are doing
-##
-################################################################
-
-## GC configuration
-#-XX:+UseConcMarkSweepGC
-#-XX:CMSInitiatingOccupancyFraction=75
-#-XX:+UseCMSInitiatingOccupancyOnly
--XX:+UseSerialGC
-
-## optimizations
-
-# pre-touch memory pages used by the JVM during initialization
--XX:+AlwaysPreTouch
-
-## basic
-
-# force the server VM (remove on 32-bit client JVMs)
--server
-
-# explicitly set the stack size (reduce to 320k on 32-bit client JVMs)
--Xss1m
-
-# set to headless, just in case
--Djava.awt.headless=true
-
-# ensure UTF-8 encoding by default (e.g. filenames)
--Dfile.encoding=UTF-8
-
-# use our provided JNA always versus the system one
--Djna.nosys=true
-
-# use old-style file permissions on JDK9
--Djdk.io.permissionsUseCanonicalPath=true
-
-# flags to configure Netty
--Dio.netty.noUnsafe=true
--Dio.netty.noKeySetOptimization=true
--Dio.netty.recycler.maxCapacityPerThread=0
-
-# log4j 2
--Dlog4j.shutdownHookEnabled=false
--Dlog4j2.disable.jmx=true
--Dlog4j.skipJansi=true
-
-## heap dumps
-
-# generate a heap dump when an allocation from the Java heap fails
-# heap dumps are created in the working directory of the JVM
--XX:+HeapDumpOnOutOfMemoryError
-
-# specify an alternative path for heap dumps
-# ensure the directory exists and has sufficient space
-#-XX:HeapDumpPath=${heap.dump.path}
-
-## GC logging
-
-#-XX:+PrintGCDetails
-#-XX:+PrintGCTimeStamps
-#-XX:+PrintGCDateStamps
-#-XX:+PrintClassHistogram
-#-XX:+PrintTenuringDistribution
-#-XX:+PrintGCApplicationStoppedTime
-
-# log GC status to a file with time stamps
-# ensure the directory exists
-#-Xloggc:${loggc}
-
-# By default, the GC log file will not rotate.
-# By uncommenting the lines below, the GC log file
-# will be rotated every 128MB at most 32 times.
-#-XX:+UseGCLogFileRotation
-#-XX:NumberOfGCLogFiles=32
-#-XX:GCLogFileSize=128M
-
-# Elasticsearch 5.0.0 will throw an exception on unquoted field names in JSON.
-# If documents were already indexed with unquoted fields in a previous version
-# of Elasticsearch, some operations may throw errors.
-#
-# WARNING: This option will be removed in Elasticsearch 6.0.0 and is provided
-# only for migration purposes.
-#-Delasticsearch.json.allow_unquoted_field_names=true
diff --git a/nifi-nar-bundles/nifi-elasticsearch-bundle/nifi-elasticsearch-client-service/src/test/resources/setup-5.script b/nifi-nar-bundles/nifi-elasticsearch-bundle/nifi-elasticsearch-client-service/src/test/resources/setup-5.script
deleted file mode 100644
index c2cffc483c..0000000000
--- a/nifi-nar-bundles/nifi-elasticsearch-bundle/nifi-elasticsearch-client-service/src/test/resources/setup-5.script
+++ /dev/null
@@ -1,48 +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.
-
-#create mapping
-PUT:user_details/:{ "mappings":{"faketype":{ "properties":{ "email":{"type":"keyword"},"phone":{"type": "keyword"},"accessKey":{"type": "keyword"}}}}}
-PUT:messages/:{ "mappings":{"faketype":{ "properties":{ "msg":{"type":"keyword"}}}}}
-PUT:complex/:{"mappings":{"faketype":{"properties":{"msg":{"type":"keyword"},"subField":{"type":"nested","properties":{"longField":{"type":"long"},"dateField":{"type":"date"}}}}}}}
-PUT:nested/:{"mappings":{"faketype":{"properties":{"msg":{"type":"keyword"},"subField":{"type":"nested","properties":{"longField":{"type":"long"},"dateField":{"type":"date"},"deeper":{"type":"nested","properties":{"secretz":{"type":"keyword"},"deepest":{"type":"nested","properties":{"super_secret":{"type":"keyword"}}}}}}}}}}}
-PUT:bulk_a/:{ "mappings":{"faketype":{ "properties":{ "msg":{"type":"keyword"}}}}}
-PUT:bulk_b/:{ "mappings":{"faketype":{ "properties":{ "msg":{"type":"keyword"}}}}}
-PUT:bulk_c/:{ "mappings":{"faketype":{ "properties":{ "msg":{"type":"keyword"}}}}}
-PUT:error_handler:{ "mappings": { "faketype": { "properties": { "msg": { "type": "keyword" }, "intField": { "type": "integer" }}}}}
-
-#add document
-PUT:messages/faketype/1:{ "msg":"one" }
-PUT:messages/faketype/2:{ "msg":"two" }
-PUT:messages/faketype/3:{ "msg":"two" }
-PUT:messages/faketype/4:{ "msg":"three" }
-PUT:messages/faketype/5:{ "msg":"three" }
-PUT:messages/faketype/6:{ "msg":"three" }
-PUT:messages/faketype/7:{ "msg":"four" }
-PUT:messages/faketype/8:{ "msg":"four" }
-PUT:messages/faketype/9:{ "msg":"four" }
-PUT:messages/faketype/10:{ "msg":"four" }
-PUT:messages/faketype/11:{ "msg":"five" }
-PUT:messages/faketype/12:{ "msg":"five" }
-PUT:messages/faketype/13:{ "msg":"five" }
-PUT:messages/faketype/14:{ "msg":"five" }
-PUT:messages/faketype/15:{ "msg":"five" }
-PUT:complex/faketype/1:{"msg":"Hello, world","subField":{"longField":100000,"dateField":"2018-04-10T12:18:05Z"}}
-PUT:user_details/faketype/1:{ "email": "john.smith@company.com", "phone": "123-456-7890", "accessKey": "ABCDE"}
-PUT:user_details/faketype/2:{ "email": "jane.doe@company.com", "phone": "098-765-4321", "accessKey": "GHIJK"}
-PUT:nested/faketype/1:{"msg":"Hello, world","subField":{"longField":150000,"dateField":"2018-08-14T10:08:00Z","deeper":{"secretz":"No one should see this!","deepest":{"super_secret":"Got nothin to hide"}}}}
-PUT:nested/faketype/2:{"msg":"Hello, world","subField":{"longField":150000,"dateField":"2018-08-14T10:08:00Z","deeper":{"secretz":"Hello, world!","deepest":{"super_secret":"I could tell, but then I would have to kill you"}}}}
-PUT:nested/faketype/3:{"msg":"Hello, world","subField":{"longField":150000,"dateField":"2018-08-14T10:08:00Z","deeper":{"secretz":"Buongiorno, mondo!!","deepest":{"super_secret":"The sky is blue"}}}}
-
diff --git a/nifi-nar-bundles/nifi-elasticsearch-bundle/nifi-elasticsearch-restapi-processors/pom.xml b/nifi-nar-bundles/nifi-elasticsearch-bundle/nifi-elasticsearch-restapi-processors/pom.xml
index 581c91b5bd..ecec4ec40e 100644
--- a/nifi-nar-bundles/nifi-elasticsearch-bundle/nifi-elasticsearch-restapi-processors/pom.xml
+++ b/nifi-nar-bundles/nifi-elasticsearch-bundle/nifi-elasticsearch-restapi-processors/pom.xml
@@ -40,55 +40,51 @@ language governing permissions and limitations under the License. -->
</dependency>
<dependency>
<groupId>org.apache.nifi</groupId>
- <artifactId>nifi-mock</artifactId>
+ <artifactId>nifi-ssl-context-service-api</artifactId>
<version>1.19.0-SNAPSHOT</version>
- <scope>test</scope>
</dependency>
<dependency>
<groupId>org.apache.nifi</groupId>
- <artifactId>nifi-ssl-context-service-api</artifactId>
+ <artifactId>nifi-schema-registry-service-api</artifactId>
<version>1.19.0-SNAPSHOT</version>
</dependency>
<dependency>
<groupId>org.apache.nifi</groupId>
- <artifactId>nifi-schema-registry-service-api</artifactId>
+ <artifactId>nifi-elasticsearch-client-service-api</artifactId>
<version>1.19.0-SNAPSHOT</version>
+ <scope>provided</scope>
</dependency>
<dependency>
- <groupId>commons-io</groupId>
- <artifactId>commons-io</artifactId>
+ <groupId>org.apache.nifi</groupId>
+ <artifactId>nifi-json-utils</artifactId>
+ <version>1.19.0-SNAPSHOT</version>
</dependency>
<dependency>
<groupId>org.apache.nifi</groupId>
- <artifactId>nifi-ssl-context-service</artifactId>
- <version>1.19.0-SNAPSHOT</version>
- <scope>test</scope>
+ <artifactId>nifi-record-serialization-service-api</artifactId>
</dependency>
<dependency>
- <groupId>com.fasterxml.jackson.core</groupId>
- <artifactId>jackson-databind</artifactId>
+ <groupId>org.apache.nifi</groupId>
+ <artifactId>nifi-record</artifactId>
</dependency>
<dependency>
<groupId>org.apache.nifi</groupId>
- <artifactId>nifi-elasticsearch-client-service-api</artifactId>
+ <artifactId>nifi-record-path</artifactId>
<version>1.19.0-SNAPSHOT</version>
- <scope>provided</scope>
</dependency>
<dependency>
<groupId>org.apache.nifi</groupId>
- <artifactId>nifi-json-utils</artifactId>
+ <artifactId>nifi-standard-record-utils</artifactId>
<version>1.19.0-SNAPSHOT</version>
- <scope>compile</scope>
</dependency>
+
<dependency>
- <groupId>org.apache.nifi</groupId>
- <artifactId>nifi-record-serialization-service-api</artifactId>
- <scope>compile</scope>
+ <groupId>commons-io</groupId>
+ <artifactId>commons-io</artifactId>
</dependency>
<dependency>
- <groupId>org.apache.nifi</groupId>
- <artifactId>nifi-record</artifactId>
- <scope>compile</scope>
+ <groupId>com.fasterxml.jackson.core</groupId>
+ <artifactId>jackson-databind</artifactId>
</dependency>
<dependency>
<groupId>com.jayway.jsonpath</groupId>
@@ -96,17 +92,19 @@ language governing permissions and limitations under the License. -->
<version>2.7.0</version>
<scope>compile</scope>
</dependency>
+
+ <!-- test dependencies -->
<dependency>
<groupId>org.apache.nifi</groupId>
- <artifactId>nifi-record-path</artifactId>
+ <artifactId>nifi-mock</artifactId>
<version>1.19.0-SNAPSHOT</version>
- <scope>compile</scope>
+ <scope>test</scope>
</dependency>
<dependency>
<groupId>org.apache.nifi</groupId>
- <artifactId>nifi-standard-record-utils</artifactId>
+ <artifactId>nifi-ssl-context-service</artifactId>
<version>1.19.0-SNAPSHOT</version>
- <scope>compile</scope>
+ <scope>test</scope>
</dependency>
<dependency>
<groupId>org.apache.nifi</groupId>
@@ -132,6 +130,23 @@ language governing permissions and limitations under the License. -->
<version>1.19.0-SNAPSHOT</version>
<scope>test</scope>
</dependency>
+ <dependency>
+ <groupId>org.apache.nifi</groupId>
+ <artifactId>nifi-elasticsearch-test-utils</artifactId>
+ <version>1.19.0-SNAPSHOT</version>
+ <scope>test</scope>
+ </dependency>
+ <dependency>
+ <groupId>org.elasticsearch.client</groupId>
+ <artifactId>elasticsearch-rest-client</artifactId>
+ <scope>test</scope>
+ </dependency>
+ <dependency>
+ <groupId>org.slf4j</groupId>
+ <artifactId>jcl-over-slf4j</artifactId>
+ <version>${org.slf4j.version}</version>
+ <scope>test</scope>
+ </dependency>
</dependencies>
<build>
diff --git a/nifi-nar-bundles/nifi-elasticsearch-bundle/nifi-elasticsearch-restapi-processors/src/main/java/org/apache/nifi/processors/elasticsearch/AbstractByQueryElasticsearch.java b/nifi-nar-bundles/nifi-elasticsearch-bundle/nifi-elasticsearch-restapi-processors/src/main/java/org/apache/nifi/processors/elasticsearch/AbstractByQueryElasticsearch.java
index 2336a9c6ba..5d0e5031f1 100644
--- a/nifi-nar-bundles/nifi-elasticsearch-bundle/nifi-elasticsearch-restapi-processors/src/main/java/org/apache/nifi/processors/elasticsearch/AbstractByQueryElasticsearch.java
+++ b/nifi-nar-bundles/nifi-elasticsearch-bundle/nifi-elasticsearch-restapi-processors/src/main/java/org/apache/nifi/processors/elasticsearch/AbstractByQueryElasticsearch.java
@@ -18,6 +18,7 @@
package org.apache.nifi.processors.elasticsearch;
import org.apache.nifi.annotation.lifecycle.OnScheduled;
+import org.apache.nifi.annotation.lifecycle.OnStopped;
import org.apache.nifi.components.PropertyDescriptor;
import org.apache.nifi.elasticsearch.ElasticSearchClientService;
import org.apache.nifi.elasticsearch.ElasticsearchException;
@@ -38,6 +39,7 @@ import java.util.HashSet;
import java.util.List;
import java.util.Map;
import java.util.Set;
+import java.util.concurrent.atomic.AtomicReference;
public abstract class AbstractByQueryElasticsearch extends AbstractProcessor implements ElasticsearchRestProcessor {
public static final Relationship REL_FAILURE = new Relationship.Builder().name("failure")
@@ -51,7 +53,7 @@ public abstract class AbstractByQueryElasticsearch extends AbstractProcessor imp
private static final Set<Relationship> relationships;
private static final List<PropertyDescriptor> propertyDescriptors;
- private volatile ElasticSearchClientService clientService;
+ private final AtomicReference<ElasticSearchClientService> clientService = new AtomicReference<>(null);
static {
final Set<Relationship> rels = new HashSet<>();
@@ -99,9 +101,19 @@ public abstract class AbstractByQueryElasticsearch extends AbstractProcessor imp
.build();
}
+ @Override
+ public boolean isIndexNotExistSuccessful() {
+ return false;
+ }
+
@OnScheduled
public void onScheduled(final ProcessContext context) {
- clientService = context.getProperty(CLIENT_SERVICE).asControllerService(ElasticSearchClientService.class);
+ clientService.set(context.getProperty(CLIENT_SERVICE).asControllerService(ElasticSearchClientService.class));
+ }
+
+ @OnStopped
+ public void onStopped() {
+ clientService.set(null);
}
@Override
@@ -125,7 +137,7 @@ public abstract class AbstractByQueryElasticsearch extends AbstractProcessor imp
? context.getProperty(QUERY_ATTRIBUTE).evaluateAttributeExpressions(input).getValue()
: null;
- final OperationResponse or = performOperation(this.clientService, query, index, type, getUrlQueryParameters(context, input));
+ final OperationResponse or = performOperation(clientService.get(), query, index, type, getUrlQueryParameters(context, input));
if (input == null) {
input = session.create();
diff --git a/nifi-nar-bundles/nifi-elasticsearch-bundle/nifi-elasticsearch-restapi-processors/src/main/java/org/apache/nifi/processors/elasticsearch/AbstractJsonQueryElasticsearch.java b/nifi-nar-bundles/nifi-elasticsearch-bundle/nifi-elasticsearch-restapi-processors/src/main/java/org/apache/nifi/processors/elasticsearch/AbstractJsonQueryElasticsearch.java
index 25c9417e4b..1ce0f7f414 100644
--- a/nifi-nar-bundles/nifi-elasticsearch-bundle/nifi-elasticsearch-restapi-processors/src/main/java/org/apache/nifi/processors/elasticsearch/AbstractJsonQueryElasticsearch.java
+++ b/nifi-nar-bundles/nifi-elasticsearch-bundle/nifi-elasticsearch-restapi-processors/src/main/java/org/apache/nifi/processors/elasticsearch/AbstractJsonQueryElasticsearch.java
@@ -101,13 +101,14 @@ public abstract class AbstractJsonQueryElasticsearch<Q extends JsonQueryParamete
private static final Set<Relationship> relationships;
private static final List<PropertyDescriptor> propertyDescriptors;
- AtomicReference<ElasticSearchClientService> clientService;
String splitUpHits;
private String splitUpAggregations;
private boolean outputNoHits;
final ObjectMapper mapper = new ObjectMapper();
+ final AtomicReference<ElasticSearchClientService> clientService = new AtomicReference<>(null);
+
static {
final Set<Relationship> rels = new HashSet<>();
rels.add(REL_ORIGINAL);
@@ -150,9 +151,14 @@ public abstract class AbstractJsonQueryElasticsearch<Q extends JsonQueryParamete
.build();
}
+ @Override
+ public boolean isIndexNotExistSuccessful() {
+ return false;
+ }
+
@OnScheduled
public void onScheduled(final ProcessContext context) {
- clientService = new AtomicReference<>(context.getProperty(CLIENT_SERVICE).asControllerService(ElasticSearchClientService.class));
+ clientService.set(context.getProperty(CLIENT_SERVICE).asControllerService(ElasticSearchClientService.class));
splitUpHits = context.getProperty(SEARCH_RESULTS_SPLIT).getValue();
splitUpAggregations = context.getProperty(AGGREGATION_RESULTS_SPLIT).getValue();
@@ -162,7 +168,7 @@ public abstract class AbstractJsonQueryElasticsearch<Q extends JsonQueryParamete
@OnStopped
public void onStopped() {
- this.clientService = null;
+ clientService.set(null);
}
@Override
@@ -286,6 +292,7 @@ public abstract class AbstractJsonQueryElasticsearch<Q extends JsonQueryParamete
* for paginated queries, the List could contain one (or more) FlowFiles, to which further hits may be appended when the next
* SearchResponse is processed, i.e. this approach allows recursion for paginated queries, but is unnecessary for single-response queries.
*/
+ @SuppressWarnings("unused")
List<FlowFile> handleHits(final List<Map<String, Object>> hits, final boolean newQuery, final Q queryJsonParameters, final ProcessSession session,
final FlowFile parent, final Map<String, String> attributes, final List<FlowFile> hitsFlowFiles,
final String transitUri, final StopWatch stopWatch) throws IOException {
diff --git a/nifi-nar-bundles/nifi-elasticsearch-bundle/nifi-elasticsearch-restapi-processors/src/main/java/org/apache/nifi/processors/elasticsearch/AbstractPutElasticsearch.java b/nifi-nar-bundles/nifi-elasticsearch-bundle/nifi-elasticsearch-restapi-processors/src/main/java/org/apache/nifi/processors/elasticsearch/AbstractPutElasticsearch.java
index 638ad97e2c..af12d18102 100644
--- a/nifi-nar-bundles/nifi-elasticsearch-bundle/nifi-elasticsearch-restapi-processors/src/main/java/org/apache/nifi/processors/elasticsearch/AbstractPutElasticsearch.java
+++ b/nifi-nar-bundles/nifi-elasticsearch-bundle/nifi-elasticsearch-restapi-processors/src/main/java/org/apache/nifi/processors/elasticsearch/AbstractPutElasticsearch.java
@@ -21,6 +21,7 @@ import com.fasterxml.jackson.core.JsonProcessingException;
import com.fasterxml.jackson.databind.ObjectMapper;
import com.fasterxml.jackson.databind.SerializationFeature;
import org.apache.nifi.annotation.lifecycle.OnScheduled;
+import org.apache.nifi.annotation.lifecycle.OnStopped;
import org.apache.nifi.components.PropertyDescriptor;
import org.apache.nifi.components.PropertyValue;
import org.apache.nifi.components.ValidationContext;
@@ -42,6 +43,7 @@ import java.util.Collection;
import java.util.Collections;
import java.util.List;
import java.util.Map;
+import java.util.concurrent.atomic.AtomicReference;
import java.util.function.Predicate;
public abstract class AbstractPutElasticsearch extends AbstractProcessor implements ElasticsearchRestProcessor {
@@ -84,7 +86,7 @@ public abstract class AbstractPutElasticsearch extends AbstractProcessor impleme
boolean notFoundIsSuccessful;
ObjectMapper errorMapper;
- volatile ElasticSearchClientService clientService;
+ final AtomicReference<ElasticSearchClientService> clientService = new AtomicReference<>(null);
@Override
protected PropertyDescriptor getSupportedDynamicPropertyDescriptor(final String propertyDescriptorName) {
@@ -97,9 +99,16 @@ public abstract class AbstractPutElasticsearch extends AbstractProcessor impleme
.build();
}
+ @Override
+ public boolean isIndexNotExistSuccessful() {
+ // index can be created during _bulk index/create operation
+ return true;
+ }
+
@OnScheduled
public void onScheduled(final ProcessContext context) {
- this.clientService = context.getProperty(CLIENT_SERVICE).asControllerService(ElasticSearchClientService.class);
+ clientService.set(context.getProperty(CLIENT_SERVICE).asControllerService(ElasticSearchClientService.class));
+
this.logErrors = context.getProperty(LOG_ERROR_RESPONSES).asBoolean();
if (errorMapper == null && (logErrors || getLogger().isDebugEnabled())) {
@@ -108,6 +117,11 @@ public abstract class AbstractPutElasticsearch extends AbstractProcessor impleme
}
}
+ @OnStopped
+ public void onStopped() {
+ clientService.set(null);
+ }
+
@Override
protected Collection<ValidationResult> customValidate(final ValidationContext validationContext) {
final List<ValidationResult> validationResults = new ArrayList<>();
diff --git a/nifi-nar-bundles/nifi-elasticsearch-bundle/nifi-elasticsearch-restapi-processors/src/main/java/org/apache/nifi/processors/elasticsearch/DeleteByQueryElasticsearch.java b/nifi-nar-bundles/nifi-elasticsearch-bundle/nifi-elasticsearch-restapi-processors/src/main/java/org/apache/nifi/processors/elasticsearch/DeleteByQueryElasticsearch.java
index c6fcbeef75..414750c14b 100644
--- a/nifi-nar-bundles/nifi-elasticsearch-bundle/nifi-elasticsearch-restapi-processors/src/main/java/org/apache/nifi/processors/elasticsearch/DeleteByQueryElasticsearch.java
+++ b/nifi-nar-bundles/nifi-elasticsearch-bundle/nifi-elasticsearch-restapi-processors/src/main/java/org/apache/nifi/processors/elasticsearch/DeleteByQueryElasticsearch.java
@@ -34,7 +34,7 @@ import java.util.Map;
@WritesAttribute(attribute = "elasticsearch.delete.error", description = "The error message provided by Elasticsearch if there is an error running the delete.")
})
@InputRequirement(InputRequirement.Requirement.INPUT_ALLOWED)
-@Tags({ "elastic", "elasticsearch", "elasticsearch5", "elasticsearch6", "elasticsearch7", "delete", "query"})
+@Tags({ "elastic", "elasticsearch", "elasticsearch5", "elasticsearch6", "elasticsearch7", "elasticsearch8", "delete", "query"})
@CapabilityDescription("Delete from an Elasticsearch index using a query. The query can be loaded from a flowfile body " +
"or from the Query parameter.")
@DynamicProperty(
diff --git a/nifi-nar-bundles/nifi-elasticsearch-bundle/nifi-elasticsearch-restapi-processors/src/main/java/org/apache/nifi/processors/elasticsearch/ElasticsearchRestProcessor.java b/nifi-nar-bundles/nifi-elasticsearch-bundle/nifi-elasticsearch-restapi-processors/src/main/java/org/apache/nifi/processors/elasticsearch/ElasticsearchRestProcessor.java
index 9133c0aa1b..a2e22f8382 100644
--- a/nifi-nar-bundles/nifi-elasticsearch-bundle/nifi-elasticsearch-restapi-processors/src/main/java/org/apache/nifi/processors/elasticsearch/ElasticsearchRestProcessor.java
+++ b/nifi-nar-bundles/nifi-elasticsearch-bundle/nifi-elasticsearch-restapi-processors/src/main/java/org/apache/nifi/processors/elasticsearch/ElasticsearchRestProcessor.java
@@ -17,24 +17,31 @@
package org.apache.nifi.processors.elasticsearch;
+import org.apache.nifi.components.ConfigVerificationResult;
import org.apache.nifi.components.PropertyDescriptor;
import org.apache.nifi.components.Validator;
import org.apache.nifi.elasticsearch.ElasticSearchClientService;
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.VerifiableProcessor;
import org.apache.nifi.processor.util.JsonValidator;
import org.apache.nifi.processor.util.StandardValidators;
import java.io.ByteArrayOutputStream;
import java.io.IOException;
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.List;
import java.util.Map;
import java.util.stream.Collectors;
-public interface ElasticsearchRestProcessor {
+public interface ElasticsearchRestProcessor extends VerifiableProcessor {
String ATTR_RECORD_COUNT = "record.count";
+ String VERIFICATION_STEP_INDEX_EXISTS = "Elasticsearch Index Exists";
PropertyDescriptor INDEX = new PropertyDescriptor.Builder()
.name("el-rest-fetch-index")
@@ -117,13 +124,70 @@ public interface ElasticsearchRestProcessor {
}
default Map<String, String> getUrlQueryParameters(final ProcessContext context, final FlowFile flowFile) {
+ return getUrlQueryParameters(context, flowFile != null ? flowFile.getAttributes() : null);
+ }
+
+ default Map<String, String> getUrlQueryParameters(final ProcessContext context, final Map<String, String> attributes) {
return context.getProperties().entrySet().stream()
// filter non-null dynamic properties
.filter(e -> e.getKey().isDynamic() && e.getValue() != null)
// convert to Map of URL parameter keys and values
.collect(Collectors.toMap(
- e -> e.getKey().getName(),
- e -> context.getProperty(e.getKey()).evaluateAttributeExpressions(flowFile).getValue()
+ e -> e.getKey().getName(),
+ e -> context.getProperty(e.getKey()).evaluateAttributeExpressions(attributes).getValue()
));
}
+
+ @Override
+ default List<ConfigVerificationResult> verify(final ProcessContext context, final ComponentLog verificationLogger, final Map<String, String> attributes) {
+ final List<ConfigVerificationResult> results = new ArrayList<>();
+ final ConfigVerificationResult.Builder indexExistsResult = new ConfigVerificationResult.Builder()
+ .verificationStepName(VERIFICATION_STEP_INDEX_EXISTS);
+
+ ElasticSearchClientService verifyClientService = null;
+ String index = null;
+ boolean indexExists = false;
+ if (context.getProperty(CLIENT_SERVICE).isSet()) {
+ verifyClientService = context.getProperty(CLIENT_SERVICE).asControllerService(ElasticSearchClientService.class);
+ if (context.getProperty(INDEX).isSet()) {
+ index = context.getProperty(INDEX).evaluateAttributeExpressions(attributes).getValue();
+ try {
+ if (verifyClientService.exists(index, getUrlQueryParameters(context, attributes))) {
+ indexExistsResult.outcome(ConfigVerificationResult.Outcome.SUCCESSFUL)
+ .explanation(String.format("Index [%s] exists", index));
+ indexExists = true;
+ } else {
+ if (isIndexNotExistSuccessful()) {
+ indexExistsResult.outcome(ConfigVerificationResult.Outcome.SUCCESSFUL);
+ } else {
+ indexExistsResult.outcome(ConfigVerificationResult.Outcome.FAILED);
+ }
+ indexExistsResult.explanation(String.format("Index [%s] does not exist", index));
+ }
+ } catch (final Exception ex) {
+ verificationLogger.error("Error checking whether index [{}] exists", index, ex);
+ indexExistsResult.outcome(ConfigVerificationResult.Outcome.FAILED)
+ .explanation(String.format("Failed to check whether index [%s] exists", index));
+ }
+ } else {
+ indexExistsResult.outcome(ConfigVerificationResult.Outcome.SKIPPED)
+ .explanation(String.format("No [%s] specified for existence check", INDEX.getDisplayName()));
+ }
+ } else {
+ indexExistsResult.outcome(ConfigVerificationResult.Outcome.SKIPPED)
+ .explanation(CLIENT_SERVICE.getDisplayName() + " not configured, cannot check index existence");
+ }
+ results.add(indexExistsResult.build());
+ results.addAll(verifyAfterIndex(context, verificationLogger, attributes, verifyClientService, index, indexExists));
+
+ return results;
+ }
+
+ boolean isIndexNotExistSuccessful();
+
+ @SuppressWarnings("unused")
+ default List<ConfigVerificationResult> verifyAfterIndex(final ProcessContext context, final ComponentLog verificationLogger, final Map<String, String> attributes,
+ final ElasticSearchClientService verifyClientService, final String index, final boolean indexExists) {
+ return Collections.emptyList();
+ }
}
diff --git a/nifi-nar-bundles/nifi-elasticsearch-bundle/nifi-elasticsearch-restapi-processors/src/main/java/org/apache/nifi/processors/elasticsearch/GetElasticsearch.java b/nifi-nar-bundles/nifi-elasticsearch-bundle/nifi-elasticsearch-restapi-processors/src/main/java/org/apache/nifi/processors/elasticsearch/GetElasticsearch.java
index 45d818afcf..af3f85fe16 100644
--- a/nifi-nar-bundles/nifi-elasticsearch-bundle/nifi-elasticsearch-restapi-processors/src/main/java/org/apache/nifi/processors/elasticsearch/GetElasticsearch.java
+++ b/nifi-nar-bundles/nifi-elasticsearch-bundle/nifi-elasticsearch-restapi-processors/src/main/java/org/apache/nifi/processors/elasticsearch/GetElasticsearch.java
@@ -25,12 +25,15 @@ import org.apache.nifi.annotation.behavior.WritesAttributes;
import org.apache.nifi.annotation.documentation.CapabilityDescription;
import org.apache.nifi.annotation.documentation.Tags;
import org.apache.nifi.annotation.lifecycle.OnScheduled;
+import org.apache.nifi.annotation.lifecycle.OnStopped;
import org.apache.nifi.components.AllowableValue;
+import org.apache.nifi.components.ConfigVerificationResult;
import org.apache.nifi.components.PropertyDescriptor;
import org.apache.nifi.elasticsearch.ElasticSearchClientService;
import org.apache.nifi.elasticsearch.ElasticsearchException;
import org.apache.nifi.expression.ExpressionLanguageScope;
import org.apache.nifi.flowfile.FlowFile;
+import org.apache.nifi.logging.ComponentLog;
import org.apache.nifi.processor.AbstractProcessor;
import org.apache.nifi.processor.ProcessContext;
import org.apache.nifi.processor.ProcessSession;
@@ -40,6 +43,7 @@ import org.apache.nifi.processor.util.StandardValidators;
import org.apache.nifi.util.StopWatch;
import org.apache.nifi.util.StringUtils;
+import java.util.ArrayList;
import java.util.Arrays;
import java.util.Collections;
import java.util.HashMap;
@@ -48,9 +52,10 @@ import java.util.List;
import java.util.Map;
import java.util.Set;
import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicReference;
@InputRequirement(InputRequirement.Requirement.INPUT_ALLOWED)
-@Tags({"json", "elasticsearch", "elasticsearch5", "elasticsearch6", "elasticsearch7", "put", "index", "record"})
+@Tags({"json", "elasticsearch", "elasticsearch5", "elasticsearch6", "elasticsearch7", "elasticsearch8", "put", "index", "record"})
@CapabilityDescription("Elasticsearch get processor that uses the official Elastic REST client libraries " +
"to fetch a single document from Elasticsearch by _id. " +
"Note that the full body of the document will be read into memory before being written to a FlowFile for transfer.")
@@ -78,7 +83,7 @@ public class GetElasticsearch extends AbstractProcessor implements Elasticsearch
"Output the retrieved document as a FlowFile attribute specified by the Attribute Name."
);
- static final PropertyDescriptor ID = new PropertyDescriptor.Builder()
+ public static final PropertyDescriptor ID = new PropertyDescriptor.Builder()
.name("get-es-id")
.displayName("Document Id")
.description("The _id of the document to retrieve.")
@@ -115,6 +120,8 @@ public class GetElasticsearch extends AbstractProcessor implements Elasticsearch
.description("A FlowFile is routed to this relationship if the specified document does not exist in the Elasticsearch cluster.")
.build();
+ public static final String VERIFICATION_STEP_DOCUMENT_EXISTS = "Elasticsearch Document Exists";
+
static final List<PropertyDescriptor> DESCRIPTORS = Collections.unmodifiableList(Arrays.asList(
ID, INDEX, TYPE, DESTINATION, ATTRIBUTE_NAME, CLIENT_SERVICE
));
@@ -122,9 +129,10 @@ public class GetElasticsearch extends AbstractProcessor implements Elasticsearch
REL_DOC, REL_FAILURE, REL_RETRY, REL_NOT_FOUND
)));
- private volatile ElasticSearchClientService clientService;
private final ObjectMapper mapper = new ObjectMapper();
+ private final AtomicReference<ElasticSearchClientService> clientService = new AtomicReference<>(null);
+
@Override
public Set<Relationship> getRelationships() {
return RELATIONSHIPS;
@@ -146,9 +154,66 @@ public class GetElasticsearch extends AbstractProcessor implements Elasticsearch
.build();
}
+ @Override
+ public boolean isIndexNotExistSuccessful() {
+ return false;
+ }
+
@OnScheduled
public void onScheduled(final ProcessContext context) {
- this.clientService = context.getProperty(CLIENT_SERVICE).asControllerService(ElasticSearchClientService.class);
+ clientService.set(context.getProperty(CLIENT_SERVICE).asControllerService(ElasticSearchClientService.class));
+ }
+
+ @OnStopped
+ public void onStopped() {
+ clientService.set(null);
+ }
+
+ @Override
+ public List<ConfigVerificationResult> verifyAfterIndex(final ProcessContext context, final ComponentLog verificationLogger, final Map<String, String> attributes,
+ final ElasticSearchClientService verifyClientService, final String index, final boolean indexExists) {
+ final List<ConfigVerificationResult> results = new ArrayList<>();
+ final ConfigVerificationResult.Builder documentExistsResult = new ConfigVerificationResult.Builder()
+ .verificationStepName(VERIFICATION_STEP_DOCUMENT_EXISTS);
+
+ if (indexExists && context.getProperty(ID).isSet()) {
+ final String type = context.getProperty(TYPE).evaluateAttributeExpressions(attributes).getValue();
+ final String id = context.getProperty(ID).evaluateAttributeExpressions(attributes).getValue();
+ try {
+ final Map<String, String> requestParameters = new HashMap<>(getUrlQueryParameters(context, attributes));
+ requestParameters.putIfAbsent("_source", "false");
+ verifyClientService.get(index, type, id, requestParameters);
+ documentExistsResult.outcome(ConfigVerificationResult.Outcome.SUCCESSFUL)
+ .explanation(String.format("Document [%s] exists in index [%s]", id, index));
+ } catch (final ElasticsearchException ee) {
+ if (ee.isNotFound()) {
+ documentExistsResult.outcome(ConfigVerificationResult.Outcome.SUCCESSFUL)
+ .explanation(String.format("Document [%s] does not exist in index [%s]", id, index));
+ } else {
+ handleDocumentExistsCheckException(ee, documentExistsResult, verificationLogger, index, id);
+ }
+ } catch (final Exception ex) {
+ handleDocumentExistsCheckException(ex, documentExistsResult, verificationLogger, index, id);
+ }
+ } else {
+ String skippedReason;
+ if (indexExists) {
+ skippedReason = String.format("No %s specified for document existence check", ID.getDisplayName());
+ } else {
+ skippedReason = String.format("Index %s does not exist for document existence check", index);
+ }
+ documentExistsResult.outcome(ConfigVerificationResult.Outcome.SKIPPED).explanation(skippedReason);
+ }
+ results.add(documentExistsResult.build());
+
+ return results;
+ }
+
+ private void handleDocumentExistsCheckException(final Exception ex, final ConfigVerificationResult.Builder documentExistsResult,
+ final ComponentLog verificationLogger, final String index, final String id) {
+ verificationLogger.error("Error checking whether document [{}] exists in index [{}]", id, index, ex);
+ documentExistsResult.outcome(ConfigVerificationResult.Outcome.FAILED)
+ .explanation(String.format("Failed to check whether document [%s] exists in index [%s]", id, index));
}
@Override
@@ -168,7 +233,7 @@ public class GetElasticsearch extends AbstractProcessor implements Elasticsearch
}
final StopWatch stopWatch = new StopWatch(true);
- final Map<String, Object> doc = clientService.get(index, type, id, getUrlQueryParameters(context, input));
+ final Map<String, Object> doc = clientService.get().get(index, type, id, getUrlQueryParameters(context, input));
final Map<String, String> attributes = new HashMap<>(4, 1);
attributes.put("filename", id);
@@ -185,7 +250,7 @@ public class GetElasticsearch extends AbstractProcessor implements Elasticsearch
}
documentFlowFile = session.putAllAttributes(documentFlowFile, attributes);
- session.getProvenanceReporter().receive(documentFlowFile, clientService.getTransitUrl(index, type), stopWatch.getElapsed(TimeUnit.MILLISECONDS));
+ session.getProvenanceReporter().receive(documentFlowFile, clientService.get().getTransitUrl(index, type), stopWatch.getElapsed(TimeUnit.MILLISECONDS));
session.transfer(documentFlowFile, REL_DOC);
} catch (final ElasticsearchException ese) {
handleElasticsearchException(ese, input, session, index, type, id);
diff --git a/nifi-nar-bundles/nifi-elasticsearch-bundle/nifi-elasticsearch-restapi-processors/src/main/java/org/apache/nifi/processors/elasticsearch/JsonQueryElasticsearch.java b/nifi-nar-bundles/nifi-elasticsearch-bundle/nifi-elasticsearch-restapi-processors/src/main/java/org/apache/nifi/processors/elasticsearch/JsonQueryElasticsearch.java
index a0bce42741..d72f13cd27 100644
--- a/nifi-nar-bundles/nifi-elasticsearch-bundle/nifi-elasticsearch-restapi-processors/src/main/java/org/apache/nifi/processors/elasticsearch/JsonQueryElasticsearch.java
+++ b/nifi-nar-bundles/nifi-elasticsearch-bundle/nifi-elasticsearch-restapi-processors/src/main/java/org/apache/nifi/processors/elasticsearch/JsonQueryElasticsearch.java
@@ -44,7 +44,7 @@ import java.util.concurrent.TimeUnit;
})
@InputRequirement(InputRequirement.Requirement.INPUT_ALLOWED)
@EventDriven
-@Tags({"elasticsearch", "elasticsearch5", "elasticsearch6", "elasticsearch7", "query", "read", "get", "json"})
+@Tags({"elasticsearch", "elasticsearch5", "elasticsearch6", "elasticsearch7", "elasticsearch8", "query", "read", "get", "json"})
@CapabilityDescription("A processor that allows the user to run a query (with aggregations) written with the " +
"Elasticsearch JSON DSL. It does not automatically paginate queries for the user. If an incoming relationship is added to this " +
"processor, it will use the flowfile's content for the query. Care should be taken on the size of the query because the entire response " +
diff --git a/nifi-nar-bundles/nifi-elasticsearch-bundle/nifi-elasticsearch-restapi-processors/src/main/java/org/apache/nifi/processors/elasticsearch/PaginatedJsonQueryElasticsearch.java b/nifi-nar-bundles/nifi-elasticsearch-bundle/nifi-elasticsearch-restapi-processors/src/main/java/org/apache/nifi/processors/elasticsearch/PaginatedJsonQueryElasticsearch.java
index f2384b6ee0..5a8f137a48 100644
--- a/nifi-nar-bundles/nifi-elasticsearch-bundle/nifi-elasticsearch-restapi-processors/src/main/java/org/apache/nifi/processors/elasticsearch/PaginatedJsonQueryElasticsearch.java
+++ b/nifi-nar-bundles/nifi-elasticsearch-bundle/nifi-elasticsearch-restapi-processors/src/main/java/org/apache/nifi/processors/elasticsearch/PaginatedJsonQueryElasticsearch.java
@@ -47,7 +47,7 @@ import java.util.List;
})
@InputRequirement(InputRequirement.Requirement.INPUT_REQUIRED)
@EventDriven
-@Tags({"elasticsearch", "elasticsearch5", "elasticsearch6", "elasticsearch7", "query", "scroll", "page", "read", "json"})
+@Tags({"elasticsearch", "elasticsearch5", "elasticsearch6", "elasticsearch7", "elasticsearch8", "query", "scroll", "page", "read", "json"})
@CapabilityDescription("A processor that allows the user to run a paginated query (with aggregations) written with the Elasticsearch JSON DSL. " +
"It will use the flowfile's content for the query unless the QUERY attribute is populated. " +
"Search After/Point in Time queries must include a valid \"sort\" field.")
diff --git a/nifi-nar-bundles/nifi-elasticsearch-bundle/nifi-elasticsearch-restapi-processors/src/main/java/org/apache/nifi/processors/elasticsearch/PutElasticsearchJson.java b/nifi-nar-bundles/nifi-elasticsearch-bundle/nifi-elasticsearch-restapi-processors/src/main/java/org/apache/nifi/processors/elasticsearch/PutElasticsearchJson.java
index 7b23d1a22d..aac7e00d1f 100644
--- a/nifi-nar-bundles/nifi-elasticsearch-bundle/nifi-elasticsearch-restapi-processors/src/main/java/org/apache/nifi/processors/elasticsearch/PutElasticsearchJson.java
+++ b/nifi-nar-bundles/nifi-elasticsearch-bundle/nifi-elasticsearch-restapi-processors/src/main/java/org/apache/nifi/processors/elasticsearch/PutElasticsearchJson.java
@@ -55,7 +55,7 @@ import java.util.function.Predicate;
import java.util.stream.Collectors;
@InputRequirement(InputRequirement.Requirement.INPUT_REQUIRED)
-@Tags({"json", "elasticsearch", "elasticsearch5", "elasticsearch6", "elasticsearch7", "put", "index"})
+@Tags({"json", "elasticsearch", "elasticsearch5", "elasticsearch6", "elasticsearch7", "elasticsearch8", "put", "index"})
@CapabilityDescription("An Elasticsearch put processor that uses the official Elastic REST client libraries.")
@WritesAttributes({
@WritesAttribute(attribute = "elasticsearch.put.error", description = "The error message provided by Elasticsearch if there is an error indexing the document.")
@@ -200,7 +200,7 @@ public class PutElasticsearchJson extends AbstractPutElasticsearch {
errorDocuments.forEach(e ->
session.getProvenanceReporter().send(
e,
- clientService.getTransitUrl(
+ clientService.get().getTransitUrl(
context.getProperty(INDEX).evaluateAttributeExpressions(e).getValue(),
context.getProperty(TYPE).evaluateAttributeExpressions(e).getValue()
),
@@ -213,7 +213,7 @@ public class PutElasticsearchJson extends AbstractPutElasticsearch {
successfulDocuments.forEach(s ->
session.getProvenanceReporter().send(
s,
- clientService.getTransitUrl(
+ clientService.get().getTransitUrl(
context.getProperty(INDEX).evaluateAttributeExpressions(s).getValue(),
context.getProperty(TYPE).evaluateAttributeExpressions(s).getValue()
)
@@ -241,7 +241,7 @@ public class PutElasticsearchJson extends AbstractPutElasticsearch {
@SuppressWarnings("unchecked")
private List<FlowFile> indexDocuments(final List<IndexOperationRequest> operations, final List<FlowFile> originals, final ProcessContext context) throws JsonProcessingException {
- final IndexOperationResponse response = clientService.bulk(operations, getUrlQueryParameters(context, originals.get(0)));
+ final IndexOperationResponse response = clientService.get().bulk(operations, getUrlQueryParameters(context, originals.get(0)));
final List<FlowFile> errorDocuments = new ArrayList<>(response.getItems() == null ? 0 : response.getItems().size());
List<Predicate<Map<String, Object>>> errorItemFilters = new ArrayList<>(2);
diff --git a/nifi-nar-bundles/nifi-elasticsearch-bundle/nifi-elasticsearch-restapi-processors/src/main/java/org/apache/nifi/processors/elasticsearch/PutElasticsearchRecord.java b/nifi-nar-bundles/nifi-elasticsearch-bundle/nifi-elasticsearch-restapi-processors/src/main/java/org/apache/nifi/processors/elasticsearch/PutElasticsearchRecord.java
index 1b572815d3..2b682c2c24 100644
--- a/nifi-nar-bundles/nifi-elasticsearch-bundle/nifi-elasticsearch-restapi-processors/src/main/java/org/apache/nifi/processors/elasticsearch/PutElasticsearchRecord.java
+++ b/nifi-nar-bundles/nifi-elasticsearch-bundle/nifi-elasticsearch-restapi-processors/src/main/java/org/apache/nifi/processors/elasticsearch/PutElasticsearchRecord.java
@@ -79,7 +79,7 @@ import java.util.concurrent.atomic.AtomicLong;
import java.util.function.Predicate;
@InputRequirement(InputRequirement.Requirement.INPUT_REQUIRED)
-@Tags({"json", "elasticsearch", "elasticsearch5", "elasticsearch6", "elasticsearch7", "put", "index", "record"})
+@Tags({"json", "elasticsearch", "elasticsearch5", "elasticsearch6", "elasticsearch7", "elasticsearch8", "put", "index", "record"})
@CapabilityDescription("A record-aware Elasticsearch put processor that uses the official Elastic REST client libraries.")
@WritesAttributes({
@WritesAttribute(attribute = "elasticsearch.put.error", description = "The error message provided by Elasticsearch if there is an error indexing the documents."),
@@ -412,7 +412,7 @@ public class PutElasticsearchRecord extends AbstractPutElasticsearch {
stopWatch.stop();
session.getProvenanceReporter().send(
input,
- clientService.getTransitUrl(String.join(",", indices), types.isEmpty() ? null : String.join(",", types)),
+ clientService.get().getTransitUrl(String.join(",", indices), types.isEmpty() ? null : String.join(",", types)),
String.format(Locale.getDefault(), "%d Elasticsearch _bulk operation batch(es) [%d error(s), %d success(es)]", batches, erroredRecords.get(), successfulRecords.get()),
stopWatch.getDuration(TimeUnit.MILLISECONDS)
);
@@ -456,7 +456,7 @@ public class PutElasticsearchRecord extends AbstractPutElasticsearch {
}
private ResponseDetails indexDocuments(final BulkOperation bundle, final ProcessContext context, final ProcessSession session, final FlowFile input) throws IOException, SchemaNotFoundException {
- final IndexOperationResponse response = clientService.bulk(bundle.getOperationList(), getUrlQueryParameters(context, input));
+ final IndexOperationResponse response = clientService.get().bulk(bundle.getOperationList(), getUrlQueryParameters(context, input));
List<Predicate<Map<String, Object>>> errorItemFilters = new ArrayList<>(2);
if (response.hasErrors()) {
diff --git a/nifi-nar-bundles/nifi-elasticsearch-bundle/nifi-elasticsearch-restapi-processors/src/main/java/org/apache/nifi/processors/elasticsearch/SearchElasticsearch.java b/nifi-nar-bundles/nifi-elasticsearch-bundle/nifi-elasticsearch-restapi-processors/src/main/java/org/apache/nifi/processors/elasticsearch/SearchElasticsearch.java
index 93adcfe945..9636526c7a 100644
--- a/nifi-nar-bundles/nifi-elasticsearch-bundle/nifi-elasticsearch-restapi-processors/src/main/java/org/apache/nifi/processors/elasticsearch/SearchElasticsearch.java
+++ b/nifi-nar-bundles/nifi-elasticsearch-bundle/nifi-elasticsearch-restapi-processors/src/main/java/org/apache/nifi/processors/elasticsearch/SearchElasticsearch.java
@@ -62,7 +62,7 @@ import java.util.Set;
@TriggerSerially
@PrimaryNodeOnly
@DefaultSchedule(period="1 min")
-@Tags({"elasticsearch", "elasticsearch5", "elasticsearch6", "elasticsearch7", "query", "scroll", "page", "search", "json"})
+@Tags({"elasticsearch", "elasticsearch5", "elasticsearch6", "elasticsearch7", "elasticsearch8", "query", "scroll", "page", "search", "json"})
@CapabilityDescription("A processor that allows the user to repeatedly run a paginated query (with aggregations) written with the Elasticsearch JSON DSL. " +
"Search After/Point in Time queries must include a valid \"sort\" field. The processor will retrieve multiple pages of results " +
"until either no more results are available or the Pagination Keep Alive expiration is reached, after which the query will " +
diff --git a/nifi-nar-bundles/nifi-elasticsearch-bundle/nifi-elasticsearch-restapi-processors/src/main/java/org/apache/nifi/processors/elasticsearch/UpdateByQueryElasticsearch.java b/nifi-nar-bundles/nifi-elasticsearch-bundle/nifi-elasticsearch-restapi-processors/src/main/java/org/apache/nifi/processors/elasticsearch/UpdateByQueryElasticsearch.java
index a64babaed6..77f5a4555c 100644
--- a/nifi-nar-bundles/nifi-elasticsearch-bundle/nifi-elasticsearch-restapi-processors/src/main/java/org/apache/nifi/processors/elasticsearch/UpdateByQueryElasticsearch.java
+++ b/nifi-nar-bundles/nifi-elasticsearch-bundle/nifi-elasticsearch-restapi-processors/src/main/java/org/apache/nifi/processors/elasticsearch/UpdateByQueryElasticsearch.java
@@ -34,7 +34,7 @@ import java.util.Map;
@WritesAttribute(attribute = "elasticsearch.update.error", description = "The error message provided by Elasticsearch if there is an error running the update.")
})
@InputRequirement(InputRequirement.Requirement.INPUT_ALLOWED)
-@Tags({ "elastic", "elasticsearch", "elasticsearch5", "elasticsearch6", "elasticsearch7", "update", "query"})
+@Tags({ "elastic", "elasticsearch", "elasticsearch5", "elasticsearch6", "elasticsearch7", "elasticsearch8", "update", "query"})
@CapabilityDescription("Update documents in an Elasticsearch index using a query. The query can be loaded from a flowfile body " +
"or from the Query parameter.")
@DynamicProperty(
diff --git a/nifi-nar-bundles/nifi-elasticsearch-bundle/nifi-elasticsearch-restapi-processors/src/test/groovy/org/apache/nifi/processors/elasticsearch/TestElasticsearchClientService.groovy b/nifi-nar-bundles/nifi-elasticsearch-bundle/nifi-elasticsearch-restapi-processors/src/test/groovy/org/apache/nifi/processors/elasticsearch/TestElasticsearchClientService.groovy
index 02eed951f7..2d3b715ce8 100644
--- a/nifi-nar-bundles/nifi-elasticsearch-bundle/nifi-elasticsearch-restapi-processors/src/test/groovy/org/apache/nifi/processors/elasticsearch/TestElasticsearchClientService.groovy
+++ b/nifi-nar-bundles/nifi-elasticsearch-bundle/nifi-elasticsearch-restapi-processors/src/test/groovy/org/apache/nifi/processors/elasticsearch/TestElasticsearchClientService.groovy
@@ -18,7 +18,9 @@
package org.apache.nifi.processors.elasticsearch
import groovy.json.JsonSlurper
+import org.apache.nifi.components.ConfigVerificationResult
import org.apache.nifi.controller.AbstractControllerService
+import org.apache.nifi.controller.ConfigurationContext
import org.apache.nifi.elasticsearch.DeleteOperationResponse
import org.apache.nifi.elasticsearch.ElasticSearchClientService
import org.apache.nifi.elasticsearch.ElasticsearchException
@@ -26,6 +28,7 @@ import org.apache.nifi.elasticsearch.IndexOperationRequest
import org.apache.nifi.elasticsearch.IndexOperationResponse
import org.apache.nifi.elasticsearch.SearchResponse
import org.apache.nifi.elasticsearch.UpdateOperationResponse
+import org.apache.nifi.logging.ComponentLog
import org.apache.nifi.processors.elasticsearch.mock.MockElasticsearchException
import org.elasticsearch.client.Response
import org.elasticsearch.client.ResponseException
@@ -58,6 +61,11 @@ class TestElasticsearchClientService extends AbstractControllerService implement
this.requestParameters = requestParameters
}
+ @Override
+ List<ConfigVerificationResult> verify(ConfigurationContext context, ComponentLog verificationLogger, Map<String, String> variables) {
+ return null
+ }
+
@Override
IndexOperationResponse add(IndexOperationRequest operation, Map<String, String> requestParameters) {
return bulk(Arrays.asList(operation) as List<IndexOperationRequest>, requestParameters)
@@ -101,6 +109,11 @@ class TestElasticsearchClientService extends AbstractControllerService implement
void refresh(final String index, final Map<String, String> requestParameters) {
}
+ @Override
+ boolean exists(final String index, final Map<String, String> requestParameters) {
+ return true
+ }
+
@Override
Map<String, Object> get(String index, String type, String id, Map<String, String> requestParameters) {
common(throwErrorInGet || throwNotFoundInGet, requestParameters)
diff --git a/nifi-nar-bundles/nifi-elasticsearch-bundle/nifi-elasticsearch-restapi-processors/src/test/groovy/org/apache/nifi/processors/elasticsearch/mock/AbstractMockElasticsearchClient.groovy b/nifi-nar-bundles/nifi-elasticsearch-bundle/nifi-elasticsearch-restapi-processors/src/test/groovy/org/apache/nifi/processors/elasticsearch/mock/AbstractMockElasticsearchClient.groovy
index 52a0316ab6..9498b8da5a 100644
--- a/nifi-nar-bundles/nifi-elasticsearch-bundle/nifi-elasticsearch-restapi-processors/src/test/groovy/org/apache/nifi/processors/elasticsearch/mock/AbstractMockElasticsearchClient.groovy
+++ b/nifi-nar-bundles/nifi-elasticsearch-bundle/nifi-elasticsearch-restapi-processors/src/test/groovy/org/apache/nifi/processors/elasticsearch/mock/AbstractMockElasticsearchClient.groovy
@@ -17,19 +17,27 @@
package org.apache.nifi.processors.elasticsearch.mock
+import org.apache.nifi.components.ConfigVerificationResult
import org.apache.nifi.controller.AbstractControllerService
+import org.apache.nifi.controller.ConfigurationContext
import org.apache.nifi.elasticsearch.DeleteOperationResponse
import org.apache.nifi.elasticsearch.ElasticSearchClientService
import org.apache.nifi.elasticsearch.IndexOperationRequest
import org.apache.nifi.elasticsearch.IndexOperationResponse
import org.apache.nifi.elasticsearch.SearchResponse
import org.apache.nifi.elasticsearch.UpdateOperationResponse
+import org.apache.nifi.logging.ComponentLog
import org.apache.nifi.util.StringUtils
class AbstractMockElasticsearchClient extends AbstractControllerService implements ElasticSearchClientService {
boolean throwRetriableError
boolean throwFatalError
+ @Override
+ List<ConfigVerificationResult> verify(ConfigurationContext context, ComponentLog verificationLogger, Map<String, String> variables) {
+ return null
+ }
+
@Override
IndexOperationResponse add(IndexOperationRequest operation, Map<String, String> requestParameters) {
return null
@@ -69,6 +77,11 @@ class AbstractMockElasticsearchClient extends AbstractControllerService implemen
void refresh(final String index, final Map<String, String> requestParameters) {
}
+ @Override
+ boolean exists(final String index, final Map<String, String> requestParameters) {
+ return true
+ }
+
@Override
Map<String, Object> get(String index, String type, String id, Map<String, String> requestParameters) {
return null
diff --git a/nifi-nar-bundles/nifi-elasticsearch-bundle/nifi-elasticsearch-restapi-processors/src/test/java/.gitignore b/nifi-nar-bundles/nifi-elasticsearch-bundle/nifi-elasticsearch-restapi-processors/src/test/java/.gitignore
deleted file mode 100644
index 74f8eb7863..0000000000
--- a/nifi-nar-bundles/nifi-elasticsearch-bundle/nifi-elasticsearch-restapi-processors/src/test/java/.gitignore
+++ /dev/null
@@ -1 +0,0 @@
-# Placeholder to force compilation
diff --git a/nifi-nar-bundles/nifi-elasticsearch-bundle/nifi-elasticsearch-restapi-processors/src/test/java/org/apache/nifi/processors/elasticsearch/integration/AbstractElasticsearch_IT.java b/nifi-nar-bundles/nifi-elasticsearch-bundle/nifi-elasticsearch-restapi-processors/src/test/java/org/apache/nifi/processors/elasticsearch/integration/AbstractElasticsearch_IT.java
new file mode 100644
index 0000000000..58218910cd
--- /dev/null
+++ b/nifi-nar-bundles/nifi-elasticsearch-bundle/nifi-elasticsearch-restapi-processors/src/test/java/org/apache/nifi/processors/elasticsearch/integration/AbstractElasticsearch_IT.java
@@ -0,0 +1,127 @@
+/*
+ * 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.integration;
+
+import org.apache.nifi.components.ConfigVerificationResult;
+import org.apache.nifi.elasticsearch.ElasticSearchClientService;
+import org.apache.nifi.elasticsearch.ElasticSearchClientServiceImpl;
+import org.apache.nifi.elasticsearch.integration.AbstractElasticsearchITBase;
+import org.apache.nifi.processor.Processor;
+import org.apache.nifi.processor.VerifiableProcessor;
+import org.apache.nifi.processors.elasticsearch.ElasticsearchRestProcessor;
+import org.apache.nifi.util.TestRunners;
+import org.junit.jupiter.api.AfterAll;
+import org.junit.jupiter.api.AfterEach;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+
+import java.lang.reflect.InvocationTargetException;
+import java.util.Collections;
+import java.util.List;
+import java.util.stream.Collectors;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
+abstract class AbstractElasticsearch_IT extends AbstractElasticsearchITBase {
+ static final List<String> TEST_INDICES = Collections.singletonList("messages");
+
+ static final String ID = "1";
+
+ ElasticSearchClientServiceImpl service;
+
+ abstract Class<? extends Processor> getTestProcessorClass();
+
+ @BeforeEach
+ void before() throws Exception {
+ runner = TestRunners.newTestRunner(getTestProcessorClass());
+
+ service = new ElasticSearchClientServiceImpl();
+ runner.addControllerService(CLIENT_SERVICE_NAME, service);
+ runner.setProperty(service, ElasticSearchClientService.HTTP_HOSTS, elasticsearchHost);
+ runner.setProperty(service, ElasticSearchClientService.CONNECT_TIMEOUT, "10000");
+ runner.setProperty(service, ElasticSearchClientService.SOCKET_TIMEOUT, "60000");
+ runner.setProperty(service, ElasticSearchClientService.SUPPRESS_NULLS, ElasticSearchClientService.ALWAYS_SUPPRESS.getValue());
+ runner.setProperty(service, ElasticSearchClientService.USERNAME, "elastic");
+ runner.setProperty(service, ElasticSearchClientService.PASSWORD, ELASTIC_USER_PASSWORD);
+
+ runner.enableControllerService(service);
+
+ runner.setProperty(ElasticsearchRestProcessor.CLIENT_SERVICE, CLIENT_SERVICE_NAME);
+ runner.setProperty(ElasticsearchRestProcessor.INDEX, INDEX);
+ runner.setProperty(ElasticsearchRestProcessor.TYPE, type);
+
+ service.refresh(null, null);
+ }
+
+ @AfterEach
+ void after() {
+ runner.disableControllerService(service);
+ }
+
+ @AfterAll
+ static void afterAll() {
+ tearDownTestData(TEST_INDICES);
+ stopTestcontainer();
+ }
+
+ @Test
+ void testVerifyIndexExists() throws Exception {
+ final List<ConfigVerificationResult> results = ((VerifiableProcessor) runner.getProcessor()).verify(
+ runner.getProcessContext(), runner.getLogger(), Collections.emptyMap()
+ );
+
+ assertIndexVerificationResults(results, true, String.format("Index [%s] exists", INDEX));
+ }
+
+ @Test
+ void testVerifyIndexDoesNotExist() throws Exception {
+ final String notExists = "not-exists";
+ runner.setProperty(ElasticsearchRestProcessor.INDEX, notExists);
+
+ final List<ConfigVerificationResult> results = ((VerifiableProcessor) runner.getProcessor()).verify(
+ runner.getProcessContext(), runner.getLogger(), Collections.emptyMap()
+ );
+
+ assertIndexVerificationResults(results, false, String.format("Index [%s] does not exist", notExists));
+ }
+
+ private void assertIndexVerificationResults(final List<ConfigVerificationResult> results, final boolean expectedExists, final String expectedExplanation)
+ throws NoSuchMethodException, InvocationTargetException, InstantiationException, IllegalAccessException {
+ // allow for extra verification test results beyond Index Exist
+ assertTrue(results.size() >= 1);
+ final List<ConfigVerificationResult> indexResults = results.stream()
+ .filter(result -> ElasticsearchRestProcessor.VERIFICATION_STEP_INDEX_EXISTS.equals(result.getVerificationStepName()))
+ .collect(Collectors.toList());
+ assertEquals(1, indexResults.size(), results.toString());
+ final ConfigVerificationResult result = indexResults.get(0);
+
+ final ElasticsearchRestProcessor processor = (ElasticsearchRestProcessor) getTestProcessorClass().getConstructor().newInstance();
+ final ConfigVerificationResult.Outcome expectedOutcome;
+ if (processor.isIndexNotExistSuccessful()) {
+ expectedOutcome = ConfigVerificationResult.Outcome.SUCCESSFUL;
+ } else {
+ if (expectedExists) {
+ expectedOutcome = ConfigVerificationResult.Outcome.SUCCESSFUL;
+ } else {
+ expectedOutcome = ConfigVerificationResult.Outcome.FAILED;
+ }
+ }
+ assertEquals(expectedOutcome, result.getOutcome(), results.toString());
+ assertEquals(expectedExplanation, result.getExplanation(), results.toString());
+ }
+}
diff --git a/nifi-nar-bundles/nifi-elasticsearch-bundle/nifi-elasticsearch-restapi-processors/src/test/java/org/apache/nifi/processors/elasticsearch/integration/GetElasticsearch_IT.java b/nifi-nar-bundles/nifi-elasticsearch-bundle/nifi-elasticsearch-restapi-processors/src/test/java/org/apache/nifi/processors/elasticsearch/integration/GetElasticsearch_IT.java
new file mode 100644
index 0000000000..f290b6e80f
--- /dev/null
+++ b/nifi-nar-bundles/nifi-elasticsearch-bundle/nifi-elasticsearch-restapi-processors/src/test/java/org/apache/nifi/processors/elasticsearch/integration/GetElasticsearch_IT.java
@@ -0,0 +1,106 @@
+/*
+ * 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.integration;
+
+import org.apache.nifi.components.ConfigVerificationResult;
+import org.apache.nifi.processor.Processor;
+import org.apache.nifi.processor.VerifiableProcessor;
+import org.apache.nifi.processors.elasticsearch.ElasticsearchRestProcessor;
+import org.apache.nifi.processors.elasticsearch.GetElasticsearch;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+
+import java.util.Collections;
+import java.util.List;
+import java.util.Objects;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+
+class GetElasticsearch_IT extends AbstractElasticsearch_IT {
+ Class<? extends Processor> getTestProcessorClass() {
+ return GetElasticsearch.class;
+ }
+
+ @BeforeEach
+ void before() throws Exception {
+ super.before();
+
+ runner.setProperty(GetElasticsearch.ID, ID);
+ }
+
+ @Test
+ void testVerifyIndexAndDocumentExist() {
+ final List<ConfigVerificationResult> results = ((VerifiableProcessor) runner.getProcessor()).verify(
+ runner.getProcessContext(), runner.getLogger(), Collections.emptyMap()
+ );
+ assertEquals(2, results.size());
+ assertEquals(2, results.stream().filter(result -> result.getOutcome() == ConfigVerificationResult.Outcome.SUCCESSFUL).count(), results.toString());
+ // check Document result, index covered in AbstractElasticsearch_IT#testVerifyIndexExists
+ assertEquals(1, results.stream().filter(
+ result -> Objects.equals(result.getVerificationStepName(), GetElasticsearch.VERIFICATION_STEP_DOCUMENT_EXISTS)
+ && Objects.equals(result.getExplanation(), String.format("Document [%s] exists in index [%s]", ID, INDEX))
+ && result.getOutcome() == ConfigVerificationResult.Outcome.SUCCESSFUL).count(),
+ results.toString()
+ );
+ }
+
+ @Test
+ void testVerifyIndexNotExists() {
+ final String notExists = "not-exists";
+ runner.setProperty(ElasticsearchRestProcessor.INDEX, notExists);
+
+ final List<ConfigVerificationResult> results = ((VerifiableProcessor) runner.getProcessor()).verify(
+ runner.getProcessContext(), runner.getLogger(), Collections.emptyMap()
+ );
+ assertEquals(2, results.size());
+ assertEquals(1, results.stream().filter(
+ result -> Objects.equals(result.getVerificationStepName(), ElasticsearchRestProcessor.VERIFICATION_STEP_INDEX_EXISTS)
+ && Objects.equals(result.getExplanation(), String.format("Index [%s] does not exist", notExists))
+ && result.getOutcome() == ConfigVerificationResult.Outcome.FAILED).count(),
+ results.toString()
+ );
+ assertEquals(1, results.stream().filter(
+ result -> Objects.equals(result.getVerificationStepName(), GetElasticsearch.VERIFICATION_STEP_DOCUMENT_EXISTS)
+ && Objects.equals(result.getExplanation(), String.format("Index %s does not exist for document existence check", notExists))
+ && result.getOutcome() == ConfigVerificationResult.Outcome.SKIPPED).count(),
+ results.toString()
+ );
+ }
+
+ @Test
+ void testVerifyIndexExistsDocumentNotExists() {
+ final String notExists = "not-exists";
+ runner.setProperty(GetElasticsearch.ID, notExists);
+
+ final List<ConfigVerificationResult> results = ((VerifiableProcessor) runner.getProcessor()).verify(
+ runner.getProcessContext(), runner.getLogger(), Collections.emptyMap()
+ );
+ assertEquals(2, results.size());
+ assertEquals(1, results.stream().filter(
+ result -> Objects.equals(result.getVerificationStepName(), ElasticsearchRestProcessor.VERIFICATION_STEP_INDEX_EXISTS)
+ && Objects.equals(result.getExplanation(), String.format("Index [%s] exists", INDEX))
+ && result.getOutcome() == ConfigVerificationResult.Outcome.SUCCESSFUL).count(),
+ results.toString()
+ );
+ assertEquals(1, results.stream().filter(
+ result -> Objects.equals(result.getVerificationStepName(), GetElasticsearch.VERIFICATION_STEP_DOCUMENT_EXISTS)
+ && Objects.equals(result.getExplanation(), String.format("Document [%s] does not exist in index [%s]", notExists, INDEX))
+ && result.getOutcome() == ConfigVerificationResult.Outcome.SUCCESSFUL).count(),
+ results.toString()
+ );
+ }
+}
diff --git a/nifi-nar-bundles/nifi-elasticsearch-bundle/nifi-elasticsearch-restapi-processors/src/test/java/org/apache/nifi/processors/elasticsearch/integration/PutElasticsearchJson_IT.java b/nifi-nar-bundles/nifi-elasticsearch-bundle/nifi-elasticsearch-restapi-processors/src/test/java/org/apache/nifi/processors/elasticsearch/integration/PutElasticsearchJson_IT.java
new file mode 100644
index 0000000000..e5fc530eaa
--- /dev/null
+++ b/nifi-nar-bundles/nifi-elasticsearch-bundle/nifi-elasticsearch-restapi-processors/src/test/java/org/apache/nifi/processors/elasticsearch/integration/PutElasticsearchJson_IT.java
@@ -0,0 +1,26 @@
+/*
+ * 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.integration;
+
+import org.apache.nifi.processor.Processor;
+import org.apache.nifi.processors.elasticsearch.PutElasticsearchJson;
+
+class PutElasticsearchJson_IT extends AbstractElasticsearch_IT {
+ Class<? extends Processor> getTestProcessorClass() {
+ return PutElasticsearchJson.class;
+ }
+}
diff --git a/nifi-nar-bundles/nifi-elasticsearch-bundle/nifi-elasticsearch-restapi-processors/src/test/resources/setup-6.script b/nifi-nar-bundles/nifi-elasticsearch-bundle/nifi-elasticsearch-restapi-processors/src/test/resources/setup-6.script
new file mode 100644
index 0000000000..77ac6850c4
--- /dev/null
+++ b/nifi-nar-bundles/nifi-elasticsearch-bundle/nifi-elasticsearch-restapi-processors/src/test/resources/setup-6.script
@@ -0,0 +1,21 @@
+# 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.
+
+#create mapping
+PUT:messages/:{ "mappings":{"_doc":{ "properties":{ "msg":{"type":"keyword"}}}}}
+
+#add document
+PUT:messages/_doc/1:{ "msg":"one" }
+
diff --git a/nifi-nar-bundles/nifi-elasticsearch-bundle/nifi-elasticsearch-restapi-processors/src/test/resources/setup-7.script b/nifi-nar-bundles/nifi-elasticsearch-bundle/nifi-elasticsearch-restapi-processors/src/test/resources/setup-7.script
new file mode 100644
index 0000000000..ee3d2cbd8a
--- /dev/null
+++ b/nifi-nar-bundles/nifi-elasticsearch-bundle/nifi-elasticsearch-restapi-processors/src/test/resources/setup-7.script
@@ -0,0 +1,20 @@
+# 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.
+
+#create mapping
+PUT:messages/:{ "mappings":{ "properties":{ "msg":{"type":"keyword"}}}}
+
+#add document
+POST:messages/_doc/1:{ "msg":"one" }
diff --git a/nifi-nar-bundles/nifi-elasticsearch-bundle/nifi-elasticsearch-restapi-processors/src/test/resources/setup-8.script b/nifi-nar-bundles/nifi-elasticsearch-bundle/nifi-elasticsearch-restapi-processors/src/test/resources/setup-8.script
new file mode 100644
index 0000000000..ee3d2cbd8a
--- /dev/null
+++ b/nifi-nar-bundles/nifi-elasticsearch-bundle/nifi-elasticsearch-restapi-processors/src/test/resources/setup-8.script
@@ -0,0 +1,20 @@
+# 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.
+
+#create mapping
+PUT:messages/:{ "mappings":{ "properties":{ "msg":{"type":"keyword"}}}}
+
+#add document
+POST:messages/_doc/1:{ "msg":"one" }
diff --git a/nifi-nar-bundles/nifi-elasticsearch-bundle/nifi-elasticsearch-test-utils/pom.xml b/nifi-nar-bundles/nifi-elasticsearch-bundle/nifi-elasticsearch-test-utils/pom.xml
new file mode 100644
index 0000000000..693de1e995
--- /dev/null
+++ b/nifi-nar-bundles/nifi-elasticsearch-bundle/nifi-elasticsearch-test-utils/pom.xml
@@ -0,0 +1,73 @@
+<?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>1.19.0-SNAPSHOT</version>
+ </parent>
+
+ <artifactId>nifi-elasticsearch-test-utils</artifactId>
+ <packaging>jar</packaging>
+
+ <dependencies>
+ <dependency>
+ <groupId>org.apache.nifi</groupId>
+ <artifactId>nifi-mock</artifactId>
+ <version>1.19.0-SNAPSHOT</version>
+ <scope>compile</scope>
+ </dependency>
+
+ <dependency>
+ <groupId>org.apache.httpcomponents</groupId>
+ <artifactId>httpclient</artifactId>
+ </dependency>
+ <dependency>
+ <groupId>com.fasterxml.jackson.core</groupId>
+ <artifactId>jackson-databind</artifactId>
+ </dependency>
+
+ <dependency>
+ <groupId>org.elasticsearch.client</groupId>
+ <artifactId>elasticsearch-rest-client</artifactId>
+ </dependency>
+
+ <dependency>
+ <groupId>org.junit.jupiter</groupId>
+ <artifactId>junit-jupiter-api</artifactId>
+ </dependency>
+ <dependency>
+ <groupId>org.testcontainers</groupId>
+ <artifactId>testcontainers</artifactId>
+ <scope>compile</scope>
+ </dependency>
+ <dependency>
+ <groupId>org.testcontainers</groupId>
+ <artifactId>elasticsearch</artifactId>
+ <scope>compile</scope>
+ </dependency>
+ <dependency>
+ <groupId>org.testcontainers</groupId>
+ <artifactId>junit-jupiter</artifactId>
+ <scope>compile</scope>
+ </dependency>
+ <dependency>
+ <!-- testcontainers requires elements JUnit 4 (even when running in JUnit 5 mode), expected to be removed in 2.x
+ see https://github.com/testcontainers/testcontainers-java/issues/970
+ -->
+ <groupId>junit</groupId>
+ <artifactId>junit</artifactId>
+ <scope>compile</scope>
+ </dependency>
+ </dependencies>
+</project>
\ No newline at end of file
diff --git a/nifi-nar-bundles/nifi-elasticsearch-bundle/nifi-elasticsearch-test-utils/src/main/java/org/apache/nifi/elasticsearch/MapBuilder.java b/nifi-nar-bundles/nifi-elasticsearch-bundle/nifi-elasticsearch-test-utils/src/main/java/org/apache/nifi/elasticsearch/MapBuilder.java
new file mode 100644
index 0000000000..7aea00b772
--- /dev/null
+++ b/nifi-nar-bundles/nifi-elasticsearch-bundle/nifi-elasticsearch-test-utils/src/main/java/org/apache/nifi/elasticsearch/MapBuilder.java
@@ -0,0 +1,70 @@
+/*
+ * 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.elasticsearch;
+
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.Map;
+
+public class MapBuilder {
+ private final Map<String, Object> toBuild;
+
+ public MapBuilder() {
+ toBuild = new HashMap<>();
+ }
+
+ public MapBuilder of(final String key, final Object value) {
+ toBuild.put(key, value);
+ return this;
+ }
+
+ public MapBuilder of(final String key, final Object value, final String key2, final Object value2) {
+ toBuild.put(key, value);
+ toBuild.put(key2, value2);
+ return this;
+ }
+
+ public MapBuilder of(final String key, final Object value, final String key2, final Object value2, final String key3, final Object value3) {
+ toBuild.put(key, value);
+ toBuild.put(key2, value2);
+ toBuild.put(key3, value3);
+ return this;
+ }
+
+ public MapBuilder of(final String key, final Object value, final String key2, final Object value2, final String key3, final Object value3,
+ final String key4, final Object value4) {
+ toBuild.put(key, value);
+ toBuild.put(key2, value2);
+ toBuild.put(key3, value3);
+ toBuild.put(key4, value4);
+ return this;
+ }
+
+ public MapBuilder of(final String key, final Object value, final String key2, final Object value2, final String key3, final Object value3,
+ final String key4, final Object value4, final String key5, final Object value5) {
+ toBuild.put(key, value);
+ toBuild.put(key2, value2);
+ toBuild.put(key3, value3);
+ toBuild.put(key4, value4);
+ toBuild.put(key5, value5);
+ return this;
+ }
+
+ public Map<String, Object> build() {
+ return Collections.unmodifiableMap(toBuild);
+ }
+}
diff --git a/nifi-nar-bundles/nifi-elasticsearch-bundle/nifi-elasticsearch-test-utils/src/main/java/org/apache/nifi/elasticsearch/integration/AbstractElasticsearchITBase.java b/nifi-nar-bundles/nifi-elasticsearch-bundle/nifi-elasticsearch-test-utils/src/main/java/org/apache/nifi/elasticsearch/integration/AbstractElasticsearchITBase.java
new file mode 100644
index 0000000000..989844d3df
--- /dev/null
+++ b/nifi-nar-bundles/nifi-elasticsearch-bundle/nifi-elasticsearch-test-utils/src/main/java/org/apache/nifi/elasticsearch/integration/AbstractElasticsearchITBase.java
@@ -0,0 +1,192 @@
+/*
+ * 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.elasticsearch.integration;
+
+import com.fasterxml.jackson.databind.ObjectMapper;
+import org.apache.http.HttpEntity;
+import org.apache.http.HttpHost;
+import org.apache.http.auth.UsernamePasswordCredentials;
+import org.apache.http.entity.ContentType;
+import org.apache.http.impl.client.BasicCredentialsProvider;
+import org.apache.http.nio.entity.NStringEntity;
+import org.apache.nifi.util.TestRunner;
+import org.elasticsearch.client.Request;
+import org.elasticsearch.client.ResponseException;
+import org.elasticsearch.client.RestClient;
+import org.junit.jupiter.api.BeforeAll;
+import org.testcontainers.elasticsearch.ElasticsearchContainer;
+import org.testcontainers.shaded.org.apache.commons.lang3.RandomStringUtils;
+import org.testcontainers.utility.DockerImageName;
+
+import java.io.BufferedReader;
+import java.io.IOException;
+import java.io.InputStreamReader;
+import java.net.URL;
+import java.nio.file.Files;
+import java.nio.file.Paths;
+import java.util.ArrayList;
+import java.util.List;
+
+import static org.apache.http.auth.AuthScope.ANY;
+
+public abstract class AbstractElasticsearchITBase {
+ protected static final DockerImageName IMAGE = DockerImageName
+ .parse(System.getProperty("elasticsearch.docker.image", "docker.elastic.co/elasticsearch/elasticsearch:8.4.3"));
+ protected static final String ELASTIC_USER_PASSWORD = System.getProperty("elasticsearch.elastic_user.password", RandomStringUtils.randomAlphanumeric(10, 20));
+ protected static final ElasticsearchContainer ELASTICSEARCH_CONTAINER = new ElasticsearchContainer(IMAGE)
+ .withPassword(ELASTIC_USER_PASSWORD)
+ .withEnv("xpack.security.enabled", "true");
+ protected static final String CLIENT_SERVICE_NAME = "Client Service";
+ protected static final String INDEX = "messages";
+
+ protected static final ObjectMapper MAPPER = new ObjectMapper();
+
+ protected static final boolean ENABLE_TEST_CONTAINERS = "true".equalsIgnoreCase(System.getProperty("elasticsearch.testcontainers.enabled"));
+ protected static String elasticsearchHost;
+ protected static void startTestcontainer() {
+ if (ENABLE_TEST_CONTAINERS) {
+ ELASTICSEARCH_CONTAINER.start();
+ elasticsearchHost = String.format("http://%s", ELASTICSEARCH_CONTAINER.getHttpHostAddress());
+ } else {
+ elasticsearchHost = System.getProperty("elasticsearch.endpoint", "http://localhost:9200");
+ }
+ }
+
+ protected TestRunner runner;
+
+ protected static String type;
+
+ private static RestClient testDataManagementClient;
+
+ protected static void stopTestcontainer() {
+ if (ENABLE_TEST_CONTAINERS) {
+ ELASTICSEARCH_CONTAINER.stop();
+ }
+ }
+
+ @BeforeAll
+ static void beforeAll() throws IOException {
+ startTestcontainer();
+ type = getElasticMajorVersion() == 6 ? "_doc" : "";
+ System.out.printf("%n%n%n%n%n%n%n%n%n%n%n%n%n%n%nTYPE: %s%nIMAGE: %s:%s%n%n%n%n%n%n%n%n%n%n%n%n%n%n%n%n",
+ type, IMAGE.getRepository(), IMAGE.getVersionPart());
+
+ setupTestData();
+ }
+
+ private static String[] getElasticVersion() {
+ final String fullVersion = IMAGE.getVersionPart();
+ final String[] parts = fullVersion.split("\\.");
+ if (parts.length == 1) {
+ throw new IllegalArgumentException("The elasticsearch version should have at least a major and minor version ex. 7.17");
+ }
+
+ return parts;
+ }
+
+ protected static int getElasticMajorVersion() {
+ return Integer.parseInt(getElasticVersion()[0]);
+ }
+
+ protected static int getElasticMinorVersion() {
+ return Integer.parseInt(getElasticVersion()[1]);
+ }
+
+ protected static void setupTestData() throws IOException {
+ final int majorVersion = getElasticMajorVersion();
+ final URL url = new URL(elasticsearchHost);
+ testDataManagementClient = RestClient
+ .builder(new HttpHost(url.getHost(), url.getPort(), url.getProtocol()))
+ .setHttpClientConfigCallback(httpClientBuilder -> {
+ final UsernamePasswordCredentials credentials = new UsernamePasswordCredentials("elastic", ELASTIC_USER_PASSWORD);
+ final BasicCredentialsProvider provider = new BasicCredentialsProvider();
+ provider.setCredentials(ANY, credentials);
+ httpClientBuilder.setDefaultCredentialsProvider(provider);
+ return httpClientBuilder;
+ })
+ .build();
+ final String script = String.format("src/test/resources/setup-%s.script", majorVersion);
+
+ final List<SetupAction> actions = readSetupActions(script);
+
+ for (final SetupAction action : actions) {
+ final String endpoint = String.format("%s/%s", elasticsearchHost, action.path);
+ final Request request = new Request(action.verb, endpoint);
+ final HttpEntity jsonBody = new NStringEntity(action.json, ContentType.APPLICATION_JSON);
+ request.setEntity(jsonBody);
+
+ try {
+ testDataManagementClient.performRequest(request);
+ } catch (final ResponseException re) {
+ throw new IllegalStateException("Error performing action " + action, re);
+ }
+ }
+ }
+
+ protected static void tearDownTestData(final List<String> testIndices) {
+ testIndices.forEach(AbstractElasticsearchITBase::deleteIndex);
+ }
+
+ protected static void deleteIndex(final String index) {
+ final Request request = new Request("DELETE", String.format("%s/%s", elasticsearchHost, index));
+ try {
+ testDataManagementClient.performRequest(request);
+ } catch (final IOException ioe) {
+ throw new IllegalStateException("Error deleting index " + index, ioe);
+ }
+ }
+
+ private static List<SetupAction> readSetupActions(final String scriptPath) throws IOException {
+ final List<SetupAction> actions = new ArrayList<>();
+ try (BufferedReader reader = new BufferedReader(new InputStreamReader(Files.newInputStream(Paths.get(scriptPath))))) {
+ String line = reader.readLine();
+ while (line != null) {
+ if (!line.trim().isEmpty() && !line.trim().startsWith("#")) {
+ final String verb = line.substring(0, line.indexOf(":"));
+ final String path = line.substring(verb.length() + 1, line.indexOf(":", verb.length() + 1));
+ final int loc = verb.length() + path.length() + 2;
+ final String json = line.substring(loc);
+
+ actions.add(new SetupAction(verb, path, json));
+ }
+ line = reader.readLine();
+ }
+ }
+
+ return actions;
+ }
+
+ private static final class SetupAction {
+ private final String verb;
+ private final String path;
+ private final String json;
+
+ public SetupAction(final String verb, final String path, final String json) {
+ this.verb = verb;
+ this.path = path;
+ this.json = json;
+ }
+
+ @Override
+ public String toString() {
+ return "SetupAction{" +
+ "verb='" + verb + '\'' +
+ ", path='" + path + '\'' +
+ '}';
+ }
+ }
+}
diff --git a/nifi-nar-bundles/nifi-elasticsearch-bundle/pom.xml b/nifi-nar-bundles/nifi-elasticsearch-bundle/pom.xml
index 84d672302b..e12841cad0 100644
--- a/nifi-nar-bundles/nifi-elasticsearch-bundle/pom.xml
+++ b/nifi-nar-bundles/nifi-elasticsearch-bundle/pom.xml
@@ -22,8 +22,9 @@ language governing permissions and limitations under the License. -->
<packaging>pom</packaging>
<modules>
+ <module>nifi-elasticsearch-test-utils</module>
<module>nifi-elasticsearch-client-service-api</module>
- <module>nifi-elasticsearch-client-service-api-nar</module>
+ <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>
@@ -50,6 +51,74 @@ language governing permissions and limitations under the License. -->
<artifactId>commons-compress</artifactId>
<version>1.21</version>
</dependency>
+
+ <dependency>
+ <groupId>org.elasticsearch.client</groupId>
+ <artifactId>elasticsearch-rest-client</artifactId>
+ <!-- pinned at 7.13.4 as it is the last version prior to Elastic forcing the client to check it is connecting
+ to an Elastic-provided Elasticsearch instead of an instance provided by someone else (e.g. AWS OpenSearch)
+ see: https://opensearch.org/blog/community/2021/08/community-clients/ for more info.
+
+ Note: the low-level elasticsearch-rest-client remains licensed with Apache 2.0
+ (https://www.elastic.co/guide/en/elasticsearch/client/java-rest/current/_license.html) even after the move
+ of the main Elasticsearch product and elasticsearch-rest-high-level-client to Elastic 2.0/SSPL 1.0 in v7.11.0+ -->
+ <version>7.13.4</version>
+ <scope>compile</scope>
+ <exclusions>
+ <exclusion>
+ <groupId>com.vividsolutions</groupId>
+ <artifactId>jts</artifactId>
+ </exclusion>
+ <exclusion>
+ <groupId>commons-logging</groupId>
+ <artifactId>commons-logging</artifactId>
+ </exclusion>
+ </exclusions>
+ </dependency>
</dependencies>
</dependencyManagement>
+
+ <profiles>
+ <!-- currently used by nifi-elasticsearch-client-service and nifi-elasticsearch-restapi-processors -->
+ <profile>
+ <id>integration-tests</id>
+ <activation>
+ <activeByDefault>false</activeByDefault>
+ </activation>
+ <properties>
+ <elasticsearch_docker_image>8.5.0</elasticsearch_docker_image>
+ <elasticsearch.elastic.password>s3cret</elasticsearch.elastic.password>
+ </properties>
+ <build>
+ <pluginManagement>
+ <plugins>
+ <plugin>
+ <groupId>org.apache.maven.plugins</groupId>
+ <artifactId>maven-failsafe-plugin</artifactId>
+ <configuration>
+ <systemPropertyVariables>
+ <elasticsearch.endpoint>http://localhost:9200</elasticsearch.endpoint>
+ <elasticsearch.testcontainers.enabled>true</elasticsearch.testcontainers.enabled>
+ <elasticsearch.docker.image>docker.elastic.co/elasticsearch/elasticsearch:${elasticsearch_docker_image}</elasticsearch.docker.image>
+ <elasticsearch.elastic_user.password>${elasticsearch.elastic.password}</elasticsearch.elastic_user.password>
+ </systemPropertyVariables>
+ </configuration>
+ </plugin>
+ </plugins>
+ </pluginManagement>
+ </build>
+ </profile>
+ <profile>
+ <id>elasticsearch6</id>
+ <properties>
+ <elasticsearch_docker_image>6.8.23</elasticsearch_docker_image>
+ </properties>
+ </profile>
+ <profile>
+ <id>elasticsearch7</id>
+ <properties>
+ <elasticsearch_docker_image>7.17.7</elasticsearch_docker_image>
+ </properties>
+ </profile>
+ </profiles>
</project>
diff --git a/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-file-authorizer/pom.xml b/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-file-authorizer/pom.xml
index 1e977bb8b3..f6b7dc6236 100644
--- a/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-file-authorizer/pom.xml
+++ b/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-file-authorizer/pom.xml
@@ -239,7 +239,6 @@
<dependency>
<groupId>org.testcontainers</groupId>
<artifactId>testcontainers</artifactId>
- <version>1.17.3</version>
<scope>test</scope>
</dependency>
</dependencies>
diff --git a/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-shell-authorizer/pom.xml b/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-shell-authorizer/pom.xml
index 089088ce19..65a86b3660 100644
--- a/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-shell-authorizer/pom.xml
+++ b/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-shell-authorizer/pom.xml
@@ -30,7 +30,6 @@
<dependency>
<groupId>org.testcontainers</groupId>
<artifactId>testcontainers</artifactId>
- <version>1.17.3</version>
<scope>test</scope>
</dependency>
<dependency>
diff --git a/nifi-nar-bundles/nifi-smb-bundle/pom.xml b/nifi-nar-bundles/nifi-smb-bundle/pom.xml
index 2f9e13e602..bf0570ef5d 100644
--- a/nifi-nar-bundles/nifi-smb-bundle/pom.xml
+++ b/nifi-nar-bundles/nifi-smb-bundle/pom.xml
@@ -61,16 +61,6 @@
<artifactId>nifi-mock</artifactId>
<version>1.19.0-SNAPSHOT</version>
</dependency>
- <dependency>
- <groupId>org.testcontainers</groupId>
- <artifactId>testcontainers</artifactId>
- <version>${testcontainers.version}</version>
- </dependency>
- <dependency>
- <groupId>org.testcontainers</groupId>
- <artifactId>toxiproxy</artifactId>
- <version>${testcontainers.version}</version>
- </dependency>
</dependencies>
</dependencyManagement>
</project>
diff --git a/nifi-registry/nifi-registry-core/nifi-registry-test/pom.xml b/nifi-registry/nifi-registry-core/nifi-registry-test/pom.xml
index 8fa9d2fe8d..f82beea378 100644
--- a/nifi-registry/nifi-registry-core/nifi-registry-test/pom.xml
+++ b/nifi-registry/nifi-registry-core/nifi-registry-test/pom.xml
@@ -45,6 +45,7 @@
<dependency>
<groupId>org.testcontainers</groupId>
<artifactId>mysql</artifactId>
+ <version>${testcontainers.version}</version>
</dependency>
<dependency>
<groupId>org.testcontainers</groupId>
diff --git a/nifi-registry/pom.xml b/nifi-registry/pom.xml
index 4dfa2f11ee..bf211598f9 100644
--- a/nifi-registry/pom.xml
+++ b/nifi-registry/pom.xml
@@ -231,15 +231,9 @@
<version>${nifi.groovy.version}</version>
<type>test</type>
</dependency>
- <dependency>
- <groupId>org.testcontainers</groupId>
- <artifactId>testcontainers</artifactId>
- <version>${testcontainers.version}</version>
- </dependency>
<dependency>
<groupId>org.testcontainers</groupId>
<artifactId>mysql</artifactId>
- <version>${testcontainers.version}</version>
<exclusions>
<exclusion>
<groupId>org.slf4j</groupId>
@@ -247,16 +241,6 @@
</exclusion>
</exclusions>
</dependency>
- <dependency>
- <groupId>org.testcontainers</groupId>
- <artifactId>mariadb</artifactId>
- <version>${testcontainers.version}</version>
- </dependency>
- <dependency>
- <groupId>org.testcontainers</groupId>
- <artifactId>postgresql</artifactId>
- <version>${testcontainers.version}</version>
- </dependency>
</dependencies>
</dependencyManagement>
diff --git a/pom.xml b/pom.xml
index a966d40785..1a17359327 100644
--- a/pom.xml
+++ b/pom.xml
@@ -354,9 +354,10 @@
</dependency>
<dependency>
<groupId>org.testcontainers</groupId>
- <artifactId>testcontainers</artifactId>
+ <artifactId>testcontainers-bom</artifactId>
<version>${testcontainers.version}</version>
- <scope>test</scope>
+ <type>pom</type>
+ <scope>import</scope>
</dependency>
<dependency>
<groupId>org.testcontainers</groupId>
@@ -784,6 +785,7 @@
<releaseProfiles>apache-release</releaseProfiles>
<autoVersionSubmodules>true</autoVersionSubmodules>
<goals>deploy</goals>
+ <!--suppress UnresolvedMavenProperty -->
<tagNameFormat>@{project.artifactId}-@{project.version}</tagNameFormat>
<pushChanges>false</pushChanges>
<localCheckout>true</localCheckout>
@@ -1035,7 +1037,7 @@
<id>check-style</id>
<phase>validate</phase>
<configuration>
- <encoding>UTF-8</encoding>
+ <inputEncoding>UTF-8</inputEncoding>
<excludes>**/generated-sources/**/*</excludes>
<sourceDirectories>
<sourceDirectory>${project.build.sourceDirectory}</sourceDirectory>