You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@camel.apache.org by nf...@apache.org on 2022/09/20 08:45:47 UTC
[camel] 01/01: CAMEL-17780: camel-elasticsearch - Implementation based on the Java Api Client
This is an automated email from the ASF dual-hosted git repository.
nfilotto pushed a commit to branch CAMEL-17780/new-elasticsearch-comp
in repository https://gitbox.apache.org/repos/asf/camel.git
commit fdfbda372bc2a1ec31f176bc6aa5bc819277de26
Author: Nicolas Filotto <nf...@talend.com>
AuthorDate: Tue Sep 20 10:44:46 2022 +0200
CAMEL-17780: camel-elasticsearch - Implementation based on the Java Api Client
---
bom/camel-bom/pom.xml | 5 +
camel-dependencies/pom.xml | 2 +
catalog/camel-allcomponents/pom.xml | 4 +
components/camel-elasticsearch/pom.xml | 114 +++
.../ElasticsearchComponentConfigurer.java | 123 +++
.../ElasticsearchEndpointConfigurer.java | 154 ++++
.../ElasticsearchEndpointUriFactory.java | 87 ++
.../ElasticsearchActionRequestConverterLoader.java | 66 ++
.../services/org/apache/camel/TypeConverterLoader | 2 +
.../services/org/apache/camel/component.properties | 7 +
.../org/apache/camel/component/elasticsearch | 2 +
.../camel/configurer/elasticsearch-component | 2 +
.../apache/camel/configurer/elasticsearch-endpoint | 2 +
.../apache/camel/urifactory/elasticsearch-endpoint | 2 +
.../component/elasticsearch/elasticsearch.json | 73 ++
.../src/main/docs/elasticsearch-component.adoc | 277 +++++++
.../elasticsearch/ElasticsearchComponent.java | 251 ++++++
.../ElasticsearchComponentVerifierExtension.java | 89 ++
.../elasticsearch/ElasticsearchConfiguration.java | 315 +++++++
.../elasticsearch/ElasticsearchConstants.java | 55 ++
.../elasticsearch/ElasticsearchEndpoint.java | 65 ++
.../elasticsearch/ElasticsearchOperation.java | 64 ++
.../elasticsearch/ElasticsearchProducer.java | 391 +++++++++
.../ElasticsearchScrollRequestIterator.java | 150 ++++
.../BulkRequestAggregationStrategy.java | 50 ++
.../ElasticsearchActionRequestConverter.java | 296 +++++++
...icsearchRestComponentVerifierExtensionTest.java | 65 ++
.../integration/ElasticsearchBulkIT.java | 255 ++++++
.../integration/ElasticsearchClusterIndexIT.java | 86 ++
...ElasticsearchGetSearchDeleteExistsUpdateIT.java | 901 +++++++++++++++++++++
.../integration/ElasticsearchIndexIT.java | 129 +++
.../integration/ElasticsearchPingIT.java | 42 +
.../integration/ElasticsearchScrollSearchIT.java | 169 ++++
.../integration/ElasticsearchSizeLimitIT.java | 68 ++
.../integration/ElasticsearchTestSupport.java | 170 ++++
.../src/test/resources/log4j2.properties | 29 +
components/pom.xml | 1 +
parent/pom.xml | 7 +
38 files changed, 4570 insertions(+)
diff --git a/bom/camel-bom/pom.xml b/bom/camel-bom/pom.xml
index 092e662202e..850534db91b 100644
--- a/bom/camel-bom/pom.xml
+++ b/bom/camel-bom/pom.xml
@@ -763,6 +763,11 @@
<artifactId>camel-ehcache</artifactId>
<version>${project.version}</version>
</dependency>
+ <dependency>
+ <groupId>org.apache.camel</groupId>
+ <artifactId>camel-elasticsearch</artifactId>
+ <version>${project.version}</version>
+ </dependency>
<dependency>
<groupId>org.apache.camel</groupId>
<artifactId>camel-elasticsearch-rest</artifactId>
diff --git a/camel-dependencies/pom.xml b/camel-dependencies/pom.xml
index 117e9de75e0..84eb3a96bd4 100644
--- a/camel-dependencies/pom.xml
+++ b/camel-dependencies/pom.xml
@@ -178,6 +178,8 @@
<eddsa-version>0.3.0</eddsa-version>
<egit-github-core-version>2.1.5</egit-github-core-version>
<ehcache3-version>3.9.6</ehcache3-version>
+ <elasticsearch-java-client-version>8.4.1</elasticsearch-java-client-version>
+ <elasticsearch-java-client-sniffer-version>8.4.1</elasticsearch-java-client-sniffer-version>
<elasticsearch-rest-sniffer-version>7.10.2</elasticsearch-rest-sniffer-version>
<elasticsearch-rest-version>7.10.2</elasticsearch-rest-version>
<elytron-web>1.10.2.Final</elytron-web>
diff --git a/catalog/camel-allcomponents/pom.xml b/catalog/camel-allcomponents/pom.xml
index 966be55838b..2ad1eafc28d 100644
--- a/catalog/camel-allcomponents/pom.xml
+++ b/catalog/camel-allcomponents/pom.xml
@@ -477,6 +477,10 @@
<groupId>org.apache.camel</groupId>
<artifactId>camel-elasticsearch-rest</artifactId>
</dependency>
+ <dependency>
+ <groupId>org.apache.camel</groupId>
+ <artifactId>camel-elasticsearch</artifactId>
+ </dependency>
<dependency>
<groupId>org.apache.camel</groupId>
<artifactId>camel-elytron</artifactId>
diff --git a/components/camel-elasticsearch/pom.xml b/components/camel-elasticsearch/pom.xml
new file mode 100644
index 00000000000..8da8dd69763
--- /dev/null
+++ b/components/camel-elasticsearch/pom.xml
@@ -0,0 +1,114 @@
+<?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 http://maven.apache.org/maven-v4_0_0.xsd">
+
+ <modelVersion>4.0.0</modelVersion>
+
+ <parent>
+ <groupId>org.apache.camel</groupId>
+ <artifactId>components</artifactId>
+ <version>3.19.0-SNAPSHOT</version>
+ </parent>
+
+ <artifactId>camel-elasticsearch</artifactId>
+ <packaging>jar</packaging>
+ <name>Camel :: ElasticSearch Java API Client</name>
+ <description>Camel ElasticSearch Java API Client support</description>
+
+ <properties>
+ </properties>
+
+ <dependencies>
+ <dependency>
+ <groupId>org.apache.camel</groupId>
+ <artifactId>camel-support</artifactId>
+ </dependency>
+
+ <dependency>
+ <groupId>co.elastic.clients</groupId>
+ <artifactId>elasticsearch-java</artifactId>
+ <version>${elasticsearch-java-client-version}</version>
+ </dependency>
+ <dependency>
+ <groupId>org.elasticsearch.client</groupId>
+ <artifactId>elasticsearch-rest-client-sniffer</artifactId>
+ <version>${elasticsearch-java-client-sniffer-version}</version>
+ </dependency>
+ <dependency>
+ <groupId>com.fasterxml.jackson.core</groupId>
+ <artifactId>jackson-databind</artifactId>
+ </dependency>
+
+ <!-- for testing -->
+ <dependency>
+ <groupId>org.apache.camel</groupId>
+ <artifactId>camel-test-junit5</artifactId>
+ <scope>test</scope>
+ </dependency>
+ <dependency>
+ <groupId>org.apache.camel</groupId>
+ <artifactId>camel-core-catalog</artifactId>
+ <scope>test</scope>
+ </dependency>
+ <dependency>
+ <groupId>org.awaitility</groupId>
+ <artifactId>awaitility</artifactId>
+ <scope>test</scope>
+ </dependency>
+ <dependency>
+ <groupId>org.elasticsearch.client</groupId>
+ <artifactId>transport</artifactId>
+ <version>${elasticsearch-rest-version}</version>
+ <scope>test</scope>
+ </dependency>
+
+ <!-- test infra -->
+ <dependency>
+ <groupId>org.apache.camel</groupId>
+ <artifactId>camel-test-infra-elasticsearch</artifactId>
+ <version>${project.version}</version>
+ <type>test-jar</type>
+ <scope>test</scope>
+ </dependency>
+
+ <!-- logging -->
+ <dependency>
+ <groupId>org.apache.logging.log4j</groupId>
+ <artifactId>log4j-slf4j-impl</artifactId>
+ <scope>test</scope>
+ </dependency>
+ </dependencies>
+
+ <build>
+ <plugins>
+ <plugin>
+ <groupId>org.apache.maven.plugins</groupId>
+ <artifactId>maven-surefire-plugin</artifactId>
+ <configuration>
+ <systemPropertyVariables>
+ <es.path.data>target/data</es.path.data>
+ </systemPropertyVariables>
+ <forkCount>1</forkCount>
+ <reuseForks>false</reuseForks>
+ </configuration>
+ </plugin>
+ </plugins>
+ </build>
+</project>
diff --git a/components/camel-elasticsearch/src/generated/java/org/apache/camel/component/elasticsearch/ElasticsearchComponentConfigurer.java b/components/camel-elasticsearch/src/generated/java/org/apache/camel/component/elasticsearch/ElasticsearchComponentConfigurer.java
new file mode 100644
index 00000000000..dcdbec69d92
--- /dev/null
+++ b/components/camel-elasticsearch/src/generated/java/org/apache/camel/component/elasticsearch/ElasticsearchComponentConfigurer.java
@@ -0,0 +1,123 @@
+/* Generated by camel build tools - do NOT edit this file! */
+package org.apache.camel.component.elasticsearch;
+
+import java.util.Map;
+
+import org.apache.camel.CamelContext;
+import org.apache.camel.spi.ExtendedPropertyConfigurerGetter;
+import org.apache.camel.spi.PropertyConfigurerGetter;
+import org.apache.camel.spi.ConfigurerStrategy;
+import org.apache.camel.spi.GeneratedPropertyConfigurer;
+import org.apache.camel.util.CaseInsensitiveMap;
+import org.apache.camel.support.component.PropertyConfigurerSupport;
+
+/**
+ * Generated by camel build tools - do NOT edit this file!
+ */
+@SuppressWarnings("unchecked")
+public class ElasticsearchComponentConfigurer extends PropertyConfigurerSupport implements GeneratedPropertyConfigurer, PropertyConfigurerGetter {
+
+ @Override
+ public boolean configure(CamelContext camelContext, Object obj, String name, Object value, boolean ignoreCase) {
+ ElasticsearchComponent target = (ElasticsearchComponent) obj;
+ switch (ignoreCase ? name.toLowerCase() : name) {
+ case "autowiredenabled":
+ case "autowiredEnabled": target.setAutowiredEnabled(property(camelContext, boolean.class, value)); return true;
+ case "certificatepath":
+ case "certificatePath": target.setCertificatePath(property(camelContext, java.lang.String.class, value)); return true;
+ case "client": target.setClient(property(camelContext, org.elasticsearch.client.RestClient.class, value)); return true;
+ case "connectiontimeout":
+ case "connectionTimeout": target.setConnectionTimeout(property(camelContext, int.class, value)); return true;
+ case "enablessl":
+ case "enableSSL": target.setEnableSSL(property(camelContext, boolean.class, value)); return true;
+ case "enablesniffer":
+ case "enableSniffer": target.setEnableSniffer(property(camelContext, boolean.class, value)); return true;
+ case "hostaddresses":
+ case "hostAddresses": target.setHostAddresses(property(camelContext, java.lang.String.class, value)); return true;
+ case "lazystartproducer":
+ case "lazyStartProducer": target.setLazyStartProducer(property(camelContext, boolean.class, value)); return true;
+ case "maxretrytimeout":
+ case "maxRetryTimeout": target.setMaxRetryTimeout(property(camelContext, int.class, value)); return true;
+ case "password": target.setPassword(property(camelContext, java.lang.String.class, value)); return true;
+ case "sniffafterfailuredelay":
+ case "sniffAfterFailureDelay": target.setSniffAfterFailureDelay(property(camelContext, int.class, value)); return true;
+ case "snifferinterval":
+ case "snifferInterval": target.setSnifferInterval(property(camelContext, int.class, value)); return true;
+ case "sockettimeout":
+ case "socketTimeout": target.setSocketTimeout(property(camelContext, int.class, value)); return true;
+ case "user": target.setUser(property(camelContext, java.lang.String.class, value)); return true;
+ default: return false;
+ }
+ }
+
+ @Override
+ public String[] getAutowiredNames() {
+ return new String[]{"client"};
+ }
+
+ @Override
+ public Class<?> getOptionType(String name, boolean ignoreCase) {
+ switch (ignoreCase ? name.toLowerCase() : name) {
+ case "autowiredenabled":
+ case "autowiredEnabled": return boolean.class;
+ case "certificatepath":
+ case "certificatePath": return java.lang.String.class;
+ case "client": return org.elasticsearch.client.RestClient.class;
+ case "connectiontimeout":
+ case "connectionTimeout": return int.class;
+ case "enablessl":
+ case "enableSSL": return boolean.class;
+ case "enablesniffer":
+ case "enableSniffer": return boolean.class;
+ case "hostaddresses":
+ case "hostAddresses": return java.lang.String.class;
+ case "lazystartproducer":
+ case "lazyStartProducer": return boolean.class;
+ case "maxretrytimeout":
+ case "maxRetryTimeout": return int.class;
+ case "password": return java.lang.String.class;
+ case "sniffafterfailuredelay":
+ case "sniffAfterFailureDelay": return int.class;
+ case "snifferinterval":
+ case "snifferInterval": return int.class;
+ case "sockettimeout":
+ case "socketTimeout": return int.class;
+ case "user": return java.lang.String.class;
+ default: return null;
+ }
+ }
+
+ @Override
+ public Object getOptionValue(Object obj, String name, boolean ignoreCase) {
+ ElasticsearchComponent target = (ElasticsearchComponent) obj;
+ switch (ignoreCase ? name.toLowerCase() : name) {
+ case "autowiredenabled":
+ case "autowiredEnabled": return target.isAutowiredEnabled();
+ case "certificatepath":
+ case "certificatePath": return target.getCertificatePath();
+ case "client": return target.getClient();
+ case "connectiontimeout":
+ case "connectionTimeout": return target.getConnectionTimeout();
+ case "enablessl":
+ case "enableSSL": return target.isEnableSSL();
+ case "enablesniffer":
+ case "enableSniffer": return target.isEnableSniffer();
+ case "hostaddresses":
+ case "hostAddresses": return target.getHostAddresses();
+ case "lazystartproducer":
+ case "lazyStartProducer": return target.isLazyStartProducer();
+ case "maxretrytimeout":
+ case "maxRetryTimeout": return target.getMaxRetryTimeout();
+ case "password": return target.getPassword();
+ case "sniffafterfailuredelay":
+ case "sniffAfterFailureDelay": return target.getSniffAfterFailureDelay();
+ case "snifferinterval":
+ case "snifferInterval": return target.getSnifferInterval();
+ case "sockettimeout":
+ case "socketTimeout": return target.getSocketTimeout();
+ case "user": return target.getUser();
+ default: return null;
+ }
+ }
+}
+
diff --git a/components/camel-elasticsearch/src/generated/java/org/apache/camel/component/elasticsearch/ElasticsearchEndpointConfigurer.java b/components/camel-elasticsearch/src/generated/java/org/apache/camel/component/elasticsearch/ElasticsearchEndpointConfigurer.java
new file mode 100644
index 00000000000..88d01bbabaf
--- /dev/null
+++ b/components/camel-elasticsearch/src/generated/java/org/apache/camel/component/elasticsearch/ElasticsearchEndpointConfigurer.java
@@ -0,0 +1,154 @@
+/* Generated by camel build tools - do NOT edit this file! */
+package org.apache.camel.component.elasticsearch;
+
+import java.util.Map;
+
+import org.apache.camel.CamelContext;
+import org.apache.camel.spi.ExtendedPropertyConfigurerGetter;
+import org.apache.camel.spi.PropertyConfigurerGetter;
+import org.apache.camel.spi.ConfigurerStrategy;
+import org.apache.camel.spi.GeneratedPropertyConfigurer;
+import org.apache.camel.util.CaseInsensitiveMap;
+import org.apache.camel.support.component.PropertyConfigurerSupport;
+
+/**
+ * Generated by camel build tools - do NOT edit this file!
+ */
+@SuppressWarnings("unchecked")
+public class ElasticsearchEndpointConfigurer extends PropertyConfigurerSupport implements GeneratedPropertyConfigurer, PropertyConfigurerGetter {
+
+ @Override
+ public boolean configure(CamelContext camelContext, Object obj, String name, Object value, boolean ignoreCase) {
+ ElasticsearchEndpoint target = (ElasticsearchEndpoint) obj;
+ switch (ignoreCase ? name.toLowerCase() : name) {
+ case "certificatepath":
+ case "certificatePath": target.getConfiguration().setCertificatePath(property(camelContext, java.lang.String.class, value)); return true;
+ case "connectiontimeout":
+ case "connectionTimeout": target.getConfiguration().setConnectionTimeout(property(camelContext, int.class, value)); return true;
+ case "disconnect": target.getConfiguration().setDisconnect(property(camelContext, boolean.class, value)); return true;
+ case "documentclass":
+ case "documentClass": target.getConfiguration().setDocumentClass(property(camelContext, java.lang.Class.class, value)); return true;
+ case "enablessl":
+ case "enableSSL": target.getConfiguration().setEnableSSL(property(camelContext, boolean.class, value)); return true;
+ case "enablesniffer":
+ case "enableSniffer": target.getConfiguration().setEnableSniffer(property(camelContext, boolean.class, value)); return true;
+ case "from": target.getConfiguration().setFrom(property(camelContext, java.lang.Integer.class, value)); return true;
+ case "hostaddresses":
+ case "hostAddresses": target.getConfiguration().setHostAddresses(property(camelContext, java.lang.String.class, value)); return true;
+ case "indexname":
+ case "indexName": target.getConfiguration().setIndexName(property(camelContext, java.lang.String.class, value)); return true;
+ case "lazystartproducer":
+ case "lazyStartProducer": target.setLazyStartProducer(property(camelContext, boolean.class, value)); return true;
+ case "maxretrytimeout":
+ case "maxRetryTimeout": target.getConfiguration().setMaxRetryTimeout(property(camelContext, int.class, value)); return true;
+ case "operation": target.getConfiguration().setOperation(property(camelContext, org.apache.camel.component.elasticsearch.ElasticsearchOperation.class, value)); return true;
+ case "scrollkeepalivems":
+ case "scrollKeepAliveMs": target.getConfiguration().setScrollKeepAliveMs(property(camelContext, int.class, value)); return true;
+ case "size": target.getConfiguration().setSize(property(camelContext, java.lang.Integer.class, value)); return true;
+ case "sniffafterfailuredelay":
+ case "sniffAfterFailureDelay": target.getConfiguration().setSniffAfterFailureDelay(property(camelContext, int.class, value)); return true;
+ case "snifferinterval":
+ case "snifferInterval": target.getConfiguration().setSnifferInterval(property(camelContext, int.class, value)); return true;
+ case "sockettimeout":
+ case "socketTimeout": target.getConfiguration().setSocketTimeout(property(camelContext, int.class, value)); return true;
+ case "usescroll":
+ case "useScroll": target.getConfiguration().setUseScroll(property(camelContext, boolean.class, value)); return true;
+ case "waitforactiveshards":
+ case "waitForActiveShards": target.getConfiguration().setWaitForActiveShards(property(camelContext, int.class, value)); return true;
+ default: return false;
+ }
+ }
+
+ @Override
+ public Class<?> getOptionType(String name, boolean ignoreCase) {
+ switch (ignoreCase ? name.toLowerCase() : name) {
+ case "certificatepath":
+ case "certificatePath": return java.lang.String.class;
+ case "connectiontimeout":
+ case "connectionTimeout": return int.class;
+ case "disconnect": return boolean.class;
+ case "documentclass":
+ case "documentClass": return java.lang.Class.class;
+ case "enablessl":
+ case "enableSSL": return boolean.class;
+ case "enablesniffer":
+ case "enableSniffer": return boolean.class;
+ case "from": return java.lang.Integer.class;
+ case "hostaddresses":
+ case "hostAddresses": return java.lang.String.class;
+ case "indexname":
+ case "indexName": return java.lang.String.class;
+ case "lazystartproducer":
+ case "lazyStartProducer": return boolean.class;
+ case "maxretrytimeout":
+ case "maxRetryTimeout": return int.class;
+ case "operation": return org.apache.camel.component.elasticsearch.ElasticsearchOperation.class;
+ case "scrollkeepalivems":
+ case "scrollKeepAliveMs": return int.class;
+ case "size": return java.lang.Integer.class;
+ case "sniffafterfailuredelay":
+ case "sniffAfterFailureDelay": return int.class;
+ case "snifferinterval":
+ case "snifferInterval": return int.class;
+ case "sockettimeout":
+ case "socketTimeout": return int.class;
+ case "usescroll":
+ case "useScroll": return boolean.class;
+ case "waitforactiveshards":
+ case "waitForActiveShards": return int.class;
+ default: return null;
+ }
+ }
+
+ @Override
+ public Object getOptionValue(Object obj, String name, boolean ignoreCase) {
+ ElasticsearchEndpoint target = (ElasticsearchEndpoint) obj;
+ switch (ignoreCase ? name.toLowerCase() : name) {
+ case "certificatepath":
+ case "certificatePath": return target.getConfiguration().getCertificatePath();
+ case "connectiontimeout":
+ case "connectionTimeout": return target.getConfiguration().getConnectionTimeout();
+ case "disconnect": return target.getConfiguration().isDisconnect();
+ case "documentclass":
+ case "documentClass": return target.getConfiguration().getDocumentClass();
+ case "enablessl":
+ case "enableSSL": return target.getConfiguration().isEnableSSL();
+ case "enablesniffer":
+ case "enableSniffer": return target.getConfiguration().isEnableSniffer();
+ case "from": return target.getConfiguration().getFrom();
+ case "hostaddresses":
+ case "hostAddresses": return target.getConfiguration().getHostAddresses();
+ case "indexname":
+ case "indexName": return target.getConfiguration().getIndexName();
+ case "lazystartproducer":
+ case "lazyStartProducer": return target.isLazyStartProducer();
+ case "maxretrytimeout":
+ case "maxRetryTimeout": return target.getConfiguration().getMaxRetryTimeout();
+ case "operation": return target.getConfiguration().getOperation();
+ case "scrollkeepalivems":
+ case "scrollKeepAliveMs": return target.getConfiguration().getScrollKeepAliveMs();
+ case "size": return target.getConfiguration().getSize();
+ case "sniffafterfailuredelay":
+ case "sniffAfterFailureDelay": return target.getConfiguration().getSniffAfterFailureDelay();
+ case "snifferinterval":
+ case "snifferInterval": return target.getConfiguration().getSnifferInterval();
+ case "sockettimeout":
+ case "socketTimeout": return target.getConfiguration().getSocketTimeout();
+ case "usescroll":
+ case "useScroll": return target.getConfiguration().isUseScroll();
+ case "waitforactiveshards":
+ case "waitForActiveShards": return target.getConfiguration().getWaitForActiveShards();
+ default: return null;
+ }
+ }
+
+ @Override
+ public Object getCollectionValueType(Object target, String name, boolean ignoreCase) {
+ switch (ignoreCase ? name.toLowerCase() : name) {
+ case "documentclass":
+ case "documentClass": return java.lang.Object.class;
+ default: return null;
+ }
+ }
+}
+
diff --git a/components/camel-elasticsearch/src/generated/java/org/apache/camel/component/elasticsearch/ElasticsearchEndpointUriFactory.java b/components/camel-elasticsearch/src/generated/java/org/apache/camel/component/elasticsearch/ElasticsearchEndpointUriFactory.java
new file mode 100644
index 00000000000..02388f2af7d
--- /dev/null
+++ b/components/camel-elasticsearch/src/generated/java/org/apache/camel/component/elasticsearch/ElasticsearchEndpointUriFactory.java
@@ -0,0 +1,87 @@
+/* Generated by camel build tools - do NOT edit this file! */
+package org.apache.camel.component.elasticsearch;
+
+import java.net.URISyntaxException;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.HashSet;
+import java.util.Map;
+import java.util.Set;
+
+import org.apache.camel.spi.EndpointUriFactory;
+
+/**
+ * Generated by camel build tools - do NOT edit this file!
+ */
+public class ElasticsearchEndpointUriFactory extends org.apache.camel.support.component.EndpointUriFactorySupport implements EndpointUriFactory {
+
+ private static final String BASE = ":clusterName";
+
+ private static final Set<String> PROPERTY_NAMES;
+ private static final Set<String> SECRET_PROPERTY_NAMES;
+ private static final Set<String> MULTI_VALUE_PREFIXES;
+ static {
+ Set<String> props = new HashSet<>(20);
+ props.add("certificatePath");
+ props.add("clusterName");
+ props.add("connectionTimeout");
+ props.add("disconnect");
+ props.add("documentClass");
+ props.add("enableSSL");
+ props.add("enableSniffer");
+ props.add("from");
+ props.add("hostAddresses");
+ props.add("indexName");
+ props.add("lazyStartProducer");
+ props.add("maxRetryTimeout");
+ props.add("operation");
+ props.add("scrollKeepAliveMs");
+ props.add("size");
+ props.add("sniffAfterFailureDelay");
+ props.add("snifferInterval");
+ props.add("socketTimeout");
+ props.add("useScroll");
+ props.add("waitForActiveShards");
+ PROPERTY_NAMES = Collections.unmodifiableSet(props);
+ SECRET_PROPERTY_NAMES = Collections.emptySet();
+ MULTI_VALUE_PREFIXES = Collections.emptySet();
+ }
+
+ @Override
+ public boolean isEnabled(String scheme) {
+ return "elasticsearch".equals(scheme);
+ }
+
+ @Override
+ public String buildUri(String scheme, Map<String, Object> properties, boolean encode) throws URISyntaxException {
+ String syntax = scheme + BASE;
+ String uri = syntax;
+
+ Map<String, Object> copy = new HashMap<>(properties);
+
+ uri = buildPathParameter(syntax, uri, "clusterName", null, true, copy);
+ uri = buildQueryParameters(uri, copy, encode);
+ return uri;
+ }
+
+ @Override
+ public Set<String> propertyNames() {
+ return PROPERTY_NAMES;
+ }
+
+ @Override
+ public Set<String> secretPropertyNames() {
+ return SECRET_PROPERTY_NAMES;
+ }
+
+ @Override
+ public Set<String> multiValuePrefixes() {
+ return MULTI_VALUE_PREFIXES;
+ }
+
+ @Override
+ public boolean isLenientProperties() {
+ return false;
+ }
+}
+
diff --git a/components/camel-elasticsearch/src/generated/java/org/apache/camel/component/elasticsearch/converter/ElasticsearchActionRequestConverterLoader.java b/components/camel-elasticsearch/src/generated/java/org/apache/camel/component/elasticsearch/converter/ElasticsearchActionRequestConverterLoader.java
new file mode 100644
index 00000000000..2e65085edaa
--- /dev/null
+++ b/components/camel-elasticsearch/src/generated/java/org/apache/camel/component/elasticsearch/converter/ElasticsearchActionRequestConverterLoader.java
@@ -0,0 +1,66 @@
+/* Generated by camel build tools - do NOT edit this file! */
+package org.apache.camel.component.elasticsearch.converter;
+
+import org.apache.camel.CamelContext;
+import org.apache.camel.CamelContextAware;
+import org.apache.camel.DeferredContextBinding;
+import org.apache.camel.Exchange;
+import org.apache.camel.TypeConversionException;
+import org.apache.camel.TypeConverterLoaderException;
+import org.apache.camel.spi.TypeConverterLoader;
+import org.apache.camel.spi.TypeConverterRegistry;
+import org.apache.camel.support.SimpleTypeConverter;
+import org.apache.camel.support.TypeConverterSupport;
+import org.apache.camel.util.DoubleMap;
+
+/**
+ * Generated by camel build tools - do NOT edit this file!
+ */
+@SuppressWarnings("unchecked")
+@DeferredContextBinding
+public final class ElasticsearchActionRequestConverterLoader implements TypeConverterLoader, CamelContextAware {
+
+ private CamelContext camelContext;
+
+ public ElasticsearchActionRequestConverterLoader() {
+ }
+
+ @Override
+ public void setCamelContext(CamelContext camelContext) {
+ this.camelContext = camelContext;
+ }
+
+ @Override
+ public CamelContext getCamelContext() {
+ return camelContext;
+ }
+
+ @Override
+ public void load(TypeConverterRegistry registry) throws TypeConverterLoaderException {
+ registerConverters(registry);
+ }
+
+ private void registerConverters(TypeConverterRegistry registry) {
+ addTypeConverter(registry, co.elastic.clients.elasticsearch.core.BulkRequest.Builder.class, java.lang.Object.class, false,
+ (type, exchange, value) -> org.apache.camel.component.elasticsearch.converter.ElasticsearchActionRequestConverter.toBulkRequestBuilder(value, exchange));
+ addTypeConverter(registry, co.elastic.clients.elasticsearch.core.DeleteRequest.Builder.class, java.lang.Object.class, false,
+ (type, exchange, value) -> org.apache.camel.component.elasticsearch.converter.ElasticsearchActionRequestConverter.toDeleteRequestBuilder(value, exchange));
+ addTypeConverter(registry, co.elastic.clients.elasticsearch.core.GetRequest.Builder.class, java.lang.Object.class, false,
+ (type, exchange, value) -> org.apache.camel.component.elasticsearch.converter.ElasticsearchActionRequestConverter.toGetRequestBuilder(value, exchange));
+ addTypeConverter(registry, co.elastic.clients.elasticsearch.core.IndexRequest.Builder.class, java.lang.Object.class, false,
+ (type, exchange, value) -> org.apache.camel.component.elasticsearch.converter.ElasticsearchActionRequestConverter.toIndexRequestBuilder(value, exchange));
+ addTypeConverter(registry, co.elastic.clients.elasticsearch.core.MgetRequest.Builder.class, java.lang.Object.class, false,
+ (type, exchange, value) -> org.apache.camel.component.elasticsearch.converter.ElasticsearchActionRequestConverter.toMgetRequestBuilder(value, exchange));
+ addTypeConverter(registry, co.elastic.clients.elasticsearch.core.SearchRequest.Builder.class, java.lang.Object.class, false,
+ (type, exchange, value) -> org.apache.camel.component.elasticsearch.converter.ElasticsearchActionRequestConverter.toSearchRequestBuilder(value, exchange));
+ addTypeConverter(registry, co.elastic.clients.elasticsearch.core.UpdateRequest.Builder.class, java.lang.Object.class, false,
+ (type, exchange, value) -> org.apache.camel.component.elasticsearch.converter.ElasticsearchActionRequestConverter.toUpdateRequestBuilder(value, exchange));
+ addTypeConverter(registry, co.elastic.clients.elasticsearch.indices.DeleteIndexRequest.Builder.class, java.lang.Object.class, false,
+ (type, exchange, value) -> org.apache.camel.component.elasticsearch.converter.ElasticsearchActionRequestConverter.toDeleteIndexRequestBuilder(value, exchange));
+ }
+
+ private static void addTypeConverter(TypeConverterRegistry registry, Class<?> toType, Class<?> fromType, boolean allowNull, SimpleTypeConverter.ConversionMethod method) {
+ registry.addTypeConverter(toType, fromType, new SimpleTypeConverter(allowNull, method));
+ }
+
+}
diff --git a/components/camel-elasticsearch/src/generated/resources/META-INF/services/org/apache/camel/TypeConverterLoader b/components/camel-elasticsearch/src/generated/resources/META-INF/services/org/apache/camel/TypeConverterLoader
new file mode 100644
index 00000000000..6d4ca3c9adc
--- /dev/null
+++ b/components/camel-elasticsearch/src/generated/resources/META-INF/services/org/apache/camel/TypeConverterLoader
@@ -0,0 +1,2 @@
+# Generated by camel build tools - do NOT edit this file!
+org.apache.camel.component.elasticsearch.converter.ElasticsearchActionRequestConverterLoader
diff --git a/components/camel-elasticsearch/src/generated/resources/META-INF/services/org/apache/camel/component.properties b/components/camel-elasticsearch/src/generated/resources/META-INF/services/org/apache/camel/component.properties
new file mode 100644
index 00000000000..65eacc359cf
--- /dev/null
+++ b/components/camel-elasticsearch/src/generated/resources/META-INF/services/org/apache/camel/component.properties
@@ -0,0 +1,7 @@
+# Generated by camel build tools - do NOT edit this file!
+components=elasticsearch
+groupId=org.apache.camel
+artifactId=camel-elasticsearch
+version=3.19.0-SNAPSHOT
+projectName=Camel :: ElasticSearch Java API Client
+projectDescription=Camel ElasticSearch Java API Client support
diff --git a/components/camel-elasticsearch/src/generated/resources/META-INF/services/org/apache/camel/component/elasticsearch b/components/camel-elasticsearch/src/generated/resources/META-INF/services/org/apache/camel/component/elasticsearch
new file mode 100644
index 00000000000..64297dc714d
--- /dev/null
+++ b/components/camel-elasticsearch/src/generated/resources/META-INF/services/org/apache/camel/component/elasticsearch
@@ -0,0 +1,2 @@
+# Generated by camel build tools - do NOT edit this file!
+class=org.apache.camel.component.elasticsearch.ElasticsearchComponent
diff --git a/components/camel-elasticsearch/src/generated/resources/META-INF/services/org/apache/camel/configurer/elasticsearch-component b/components/camel-elasticsearch/src/generated/resources/META-INF/services/org/apache/camel/configurer/elasticsearch-component
new file mode 100644
index 00000000000..823abcb3d4b
--- /dev/null
+++ b/components/camel-elasticsearch/src/generated/resources/META-INF/services/org/apache/camel/configurer/elasticsearch-component
@@ -0,0 +1,2 @@
+# Generated by camel build tools - do NOT edit this file!
+class=org.apache.camel.component.elasticsearch.ElasticsearchComponentConfigurer
diff --git a/components/camel-elasticsearch/src/generated/resources/META-INF/services/org/apache/camel/configurer/elasticsearch-endpoint b/components/camel-elasticsearch/src/generated/resources/META-INF/services/org/apache/camel/configurer/elasticsearch-endpoint
new file mode 100644
index 00000000000..cab1b57233e
--- /dev/null
+++ b/components/camel-elasticsearch/src/generated/resources/META-INF/services/org/apache/camel/configurer/elasticsearch-endpoint
@@ -0,0 +1,2 @@
+# Generated by camel build tools - do NOT edit this file!
+class=org.apache.camel.component.elasticsearch.ElasticsearchEndpointConfigurer
diff --git a/components/camel-elasticsearch/src/generated/resources/META-INF/services/org/apache/camel/urifactory/elasticsearch-endpoint b/components/camel-elasticsearch/src/generated/resources/META-INF/services/org/apache/camel/urifactory/elasticsearch-endpoint
new file mode 100644
index 00000000000..b0bc87a69b6
--- /dev/null
+++ b/components/camel-elasticsearch/src/generated/resources/META-INF/services/org/apache/camel/urifactory/elasticsearch-endpoint
@@ -0,0 +1,2 @@
+# Generated by camel build tools - do NOT edit this file!
+class=org.apache.camel.component.elasticsearch.ElasticsearchEndpointUriFactory
diff --git a/components/camel-elasticsearch/src/generated/resources/org/apache/camel/component/elasticsearch/elasticsearch.json b/components/camel-elasticsearch/src/generated/resources/org/apache/camel/component/elasticsearch/elasticsearch.json
new file mode 100644
index 00000000000..5e219e1cb7d
--- /dev/null
+++ b/components/camel-elasticsearch/src/generated/resources/org/apache/camel/component/elasticsearch/elasticsearch.json
@@ -0,0 +1,73 @@
+{
+ "component": {
+ "kind": "component",
+ "name": "elasticsearch",
+ "title": "Elasticsearch",
+ "description": "Send requests to ElasticSearch via Java Client API.",
+ "deprecated": false,
+ "firstVersion": "3.20.0",
+ "label": "search,monitoring",
+ "javaType": "org.apache.camel.component.elasticsearch.ElasticsearchComponent",
+ "supportLevel": "Preview",
+ "groupId": "org.apache.camel",
+ "artifactId": "camel-elasticsearch",
+ "version": "3.19.0-SNAPSHOT",
+ "scheme": "elasticsearch",
+ "extendsScheme": "",
+ "syntax": "elasticsearch:clusterName",
+ "async": false,
+ "api": false,
+ "consumerOnly": false,
+ "producerOnly": true,
+ "lenientProperties": false
+ },
+ "componentProperties": {
+ "connectionTimeout": { "kind": "property", "displayName": "Connection Timeout", "group": "producer", "label": "", "required": false, "type": "integer", "javaType": "int", "deprecated": false, "autowired": false, "secret": false, "defaultValue": 30000, "description": "The time in ms to wait before connection will timeout." },
+ "hostAddresses": { "kind": "property", "displayName": "Host Addresses", "group": "producer", "label": "", "required": false, "type": "string", "javaType": "java.lang.String", "deprecated": false, "autowired": false, "secret": false, "description": "Comma separated list with ip:port formatted remote transport addresses to use. The ip and port options must be left blank for hostAddresses to be considered instead." },
+ "lazyStartProducer": { "kind": "property", "displayName": "Lazy Start Producer", "group": "producer", "label": "producer", "required": false, "type": "boolean", "javaType": "boolean", "deprecated": false, "autowired": false, "secret": false, "defaultValue": false, "description": "Whether the producer should be started lazy (on the first message). By starting lazy you can use this to allow CamelContext and routes to startup in situations where a producer may otherwise fail during star [...]
+ "maxRetryTimeout": { "kind": "property", "displayName": "Max Retry Timeout", "group": "producer", "label": "", "required": false, "type": "integer", "javaType": "int", "deprecated": false, "autowired": false, "secret": false, "defaultValue": 30000, "description": "The time in ms before retry" },
+ "socketTimeout": { "kind": "property", "displayName": "Socket Timeout", "group": "producer", "label": "", "required": false, "type": "integer", "javaType": "int", "deprecated": false, "autowired": false, "secret": false, "defaultValue": 30000, "description": "The timeout in ms to wait before the socket will timeout." },
+ "autowiredEnabled": { "kind": "property", "displayName": "Autowired Enabled", "group": "advanced", "label": "advanced", "required": false, "type": "boolean", "javaType": "boolean", "deprecated": false, "autowired": false, "secret": false, "defaultValue": true, "description": "Whether autowiring is enabled. This is used for automatic autowiring options (the option must be marked as autowired) by looking up in the registry to find if there is a single instance of matching type, which t [...]
+ "client": { "kind": "property", "displayName": "Client", "group": "advanced", "label": "advanced", "required": false, "type": "object", "javaType": "org.elasticsearch.client.RestClient", "deprecated": false, "autowired": true, "secret": false, "description": "To use an existing configured Elasticsearch client, instead of creating a client per endpoint. This allow to customize the client with specific settings." },
+ "enableSniffer": { "kind": "property", "displayName": "Enable Sniffer", "group": "advanced", "label": "advanced", "required": false, "type": "boolean", "javaType": "boolean", "deprecated": false, "autowired": false, "secret": false, "defaultValue": false, "description": "Enable automatically discover nodes from a running Elasticsearch cluster. If this option is used in conjunction with Spring Boot then it's managed by the Spring Boot configuration (see: Disable Sniffer in Spring Boot)." },
+ "sniffAfterFailureDelay": { "kind": "property", "displayName": "Sniff After Failure Delay", "group": "advanced", "label": "advanced", "required": false, "type": "integer", "javaType": "int", "deprecated": false, "autowired": false, "secret": false, "defaultValue": 60000, "description": "The delay of a sniff execution scheduled after a failure (in milliseconds)" },
+ "snifferInterval": { "kind": "property", "displayName": "Sniffer Interval", "group": "advanced", "label": "advanced", "required": false, "type": "integer", "javaType": "int", "deprecated": false, "autowired": false, "secret": false, "defaultValue": 300000, "description": "The interval between consecutive ordinary sniff executions in milliseconds. Will be honoured when sniffOnFailure is disabled or when there are no failures between consecutive sniff executions" },
+ "certificatePath": { "kind": "property", "displayName": "Certificate Path", "group": "security", "label": "security", "required": false, "type": "string", "javaType": "java.lang.String", "deprecated": false, "autowired": false, "secret": false, "description": "The path of the self-signed certificate to use to access to Elasticsearch." },
+ "enableSSL": { "kind": "property", "displayName": "Enable SSL", "group": "security", "label": "security", "required": false, "type": "boolean", "javaType": "boolean", "deprecated": false, "autowired": false, "secret": false, "defaultValue": false, "description": "Enable SSL" },
+ "password": { "kind": "property", "displayName": "Password", "group": "security", "label": "security", "required": false, "type": "string", "javaType": "java.lang.String", "deprecated": false, "autowired": false, "secret": true, "description": "Password for authenticate" },
+ "user": { "kind": "property", "displayName": "User", "group": "security", "label": "security", "required": false, "type": "string", "javaType": "java.lang.String", "deprecated": false, "autowired": false, "secret": true, "description": "Basic authenticate user" }
+ },
+ "headers": {
+ "operation": { "kind": "header", "displayName": "", "group": "producer", "label": "", "required": false, "javaType": "org.apache.camel.component.elasticsearch.ElasticsearchOperation", "enum": [ "Index", "Update", "Bulk", "GetById", "MultiGet", "MultiSearch", "Delete", "DeleteIndex", "Search", "Exists", "Ping" ], "deprecated": false, "deprecationNote": "", "autowired": false, "secret": false, "description": "The operation to perform", "constantName": "org.apache.camel.component.elasti [...]
+ "indexId": { "kind": "header", "displayName": "", "group": "producer", "label": "", "required": false, "javaType": "String", "deprecated": false, "deprecationNote": "", "autowired": false, "secret": false, "description": "The id of the indexed document.", "constantName": "org.apache.camel.component.elasticsearch.ElasticsearchConstants#PARAM_INDEX_ID" },
+ "indexName": { "kind": "header", "displayName": "", "group": "producer", "label": "", "required": false, "javaType": "String", "deprecated": false, "deprecationNote": "", "autowired": false, "secret": false, "description": "The name of the index to act against", "constantName": "org.apache.camel.component.elasticsearch.ElasticsearchConstants#PARAM_INDEX_NAME" },
+ "documentClass": { "kind": "header", "displayName": "", "group": "producer", "label": "", "required": false, "javaType": "Class", "deprecated": false, "deprecationNote": "", "autowired": false, "secret": false, "defaultValue": "ObjectNode", "description": "The full qualified name of the class of the document to unmarshall", "constantName": "org.apache.camel.component.elasticsearch.ElasticsearchConstants#PARAM_DOCUMENT_CLASS" },
+ "waitForActiveShards": { "kind": "header", "displayName": "", "group": "producer", "label": "", "required": false, "javaType": "Integer", "deprecated": false, "deprecationNote": "", "autowired": false, "secret": false, "description": "The index creation waits for the write consistency number of shards to be available", "constantName": "org.apache.camel.component.elasticsearch.ElasticsearchConstants#PARAM_WAIT_FOR_ACTIVE_SHARDS" },
+ "scrollKeepAliveMs": { "kind": "header", "displayName": "", "group": "producer", "label": "", "required": false, "javaType": "Integer", "deprecated": false, "deprecationNote": "", "autowired": false, "secret": false, "description": "The starting index of the response.", "constantName": "org.apache.camel.component.elasticsearch.ElasticsearchConstants#PARAM_SCROLL_KEEP_ALIVE_MS" },
+ "useScroll": { "kind": "header", "displayName": "", "group": "producer", "label": "", "required": false, "javaType": "Boolean", "deprecated": false, "deprecationNote": "", "autowired": false, "secret": false, "description": "Set to true to enable scroll usage", "constantName": "org.apache.camel.component.elasticsearch.ElasticsearchConstants#PARAM_SCROLL" },
+ "size": { "kind": "header", "displayName": "", "group": "producer", "label": "", "required": false, "javaType": "Integer", "deprecated": false, "deprecationNote": "", "autowired": false, "secret": false, "description": "The size of the response.", "constantName": "org.apache.camel.component.elasticsearch.ElasticsearchConstants#PARAM_SIZE" },
+ "from": { "kind": "header", "displayName": "", "group": "producer", "label": "", "required": false, "javaType": "Integer", "deprecated": false, "deprecationNote": "", "autowired": false, "secret": false, "description": "The starting index of the response.", "constantName": "org.apache.camel.component.elasticsearch.ElasticsearchConstants#PARAM_FROM" }
+ },
+ "properties": {
+ "clusterName": { "kind": "path", "displayName": "Cluster Name", "group": "producer", "label": "", "required": true, "type": "string", "javaType": "java.lang.String", "deprecated": false, "deprecationNote": "", "autowired": false, "secret": false, "configurationClass": "org.apache.camel.component.elasticsearch.ElasticsearchConfiguration", "configurationField": "configuration", "description": "Name of the cluster" },
+ "connectionTimeout": { "kind": "parameter", "displayName": "Connection Timeout", "group": "producer", "label": "", "required": false, "type": "integer", "javaType": "int", "deprecated": false, "autowired": false, "secret": false, "defaultValue": 30000, "configurationClass": "org.apache.camel.component.elasticsearch.ElasticsearchConfiguration", "configurationField": "configuration", "description": "The time in ms to wait before connection will timeout." },
+ "disconnect": { "kind": "parameter", "displayName": "Disconnect", "group": "producer", "label": "", "required": false, "type": "boolean", "javaType": "boolean", "deprecated": false, "autowired": false, "secret": false, "defaultValue": false, "configurationClass": "org.apache.camel.component.elasticsearch.ElasticsearchConfiguration", "configurationField": "configuration", "description": "Disconnect after it finish calling the producer" },
+ "from": { "kind": "parameter", "displayName": "From", "group": "producer", "label": "", "required": false, "type": "integer", "javaType": "java.lang.Integer", "deprecated": false, "autowired": false, "secret": false, "configurationClass": "org.apache.camel.component.elasticsearch.ElasticsearchConfiguration", "configurationField": "configuration", "description": "Starting index of the response." },
+ "hostAddresses": { "kind": "parameter", "displayName": "Host Addresses", "group": "producer", "label": "", "required": false, "type": "string", "javaType": "java.lang.String", "deprecated": false, "autowired": false, "secret": false, "configurationClass": "org.apache.camel.component.elasticsearch.ElasticsearchConfiguration", "configurationField": "configuration", "description": "Comma separated list with ip:port formatted remote transport addresses to use." },
+ "indexName": { "kind": "parameter", "displayName": "Index Name", "group": "producer", "label": "", "required": false, "type": "string", "javaType": "java.lang.String", "deprecated": false, "autowired": false, "secret": false, "configurationClass": "org.apache.camel.component.elasticsearch.ElasticsearchConfiguration", "configurationField": "configuration", "description": "The name of the index to act against" },
+ "maxRetryTimeout": { "kind": "parameter", "displayName": "Max Retry Timeout", "group": "producer", "label": "", "required": false, "type": "integer", "javaType": "int", "deprecated": false, "autowired": false, "secret": false, "defaultValue": 30000, "configurationClass": "org.apache.camel.component.elasticsearch.ElasticsearchConfiguration", "configurationField": "configuration", "description": "The time in ms before retry" },
+ "operation": { "kind": "parameter", "displayName": "Operation", "group": "producer", "label": "", "required": false, "type": "object", "javaType": "org.apache.camel.component.elasticsearch.ElasticsearchOperation", "enum": [ "Index", "Update", "Bulk", "GetById", "MultiGet", "MultiSearch", "Delete", "DeleteIndex", "Search", "Exists", "Ping" ], "deprecated": false, "autowired": false, "secret": false, "configurationClass": "org.apache.camel.component.elasticsearch.ElasticsearchConfigura [...]
+ "scrollKeepAliveMs": { "kind": "parameter", "displayName": "Scroll Keep Alive Ms", "group": "producer", "label": "", "required": false, "type": "integer", "javaType": "int", "deprecated": false, "autowired": false, "secret": false, "defaultValue": 60000, "configurationClass": "org.apache.camel.component.elasticsearch.ElasticsearchConfiguration", "configurationField": "configuration", "description": "Time in ms during which elasticsearch will keep search context alive" },
+ "size": { "kind": "parameter", "displayName": "Size", "group": "producer", "label": "", "required": false, "type": "integer", "javaType": "java.lang.Integer", "deprecated": false, "autowired": false, "secret": false, "configurationClass": "org.apache.camel.component.elasticsearch.ElasticsearchConfiguration", "configurationField": "configuration", "description": "Size of the response." },
+ "socketTimeout": { "kind": "parameter", "displayName": "Socket Timeout", "group": "producer", "label": "", "required": false, "type": "integer", "javaType": "int", "deprecated": false, "autowired": false, "secret": false, "defaultValue": 30000, "configurationClass": "org.apache.camel.component.elasticsearch.ElasticsearchConfiguration", "configurationField": "configuration", "description": "The timeout in ms to wait before the socket will timeout." },
+ "useScroll": { "kind": "parameter", "displayName": "Use Scroll", "group": "producer", "label": "", "required": false, "type": "boolean", "javaType": "boolean", "deprecated": false, "autowired": false, "secret": false, "defaultValue": false, "configurationClass": "org.apache.camel.component.elasticsearch.ElasticsearchConfiguration", "configurationField": "configuration", "description": "Enable scroll usage" },
+ "waitForActiveShards": { "kind": "parameter", "displayName": "Wait For Active Shards", "group": "producer", "label": "", "required": false, "type": "integer", "javaType": "int", "deprecated": false, "autowired": false, "secret": false, "defaultValue": 1, "configurationClass": "org.apache.camel.component.elasticsearch.ElasticsearchConfiguration", "configurationField": "configuration", "description": "Index creation waits for the write consistency number of shards to be available" },
+ "lazyStartProducer": { "kind": "parameter", "displayName": "Lazy Start Producer", "group": "producer (advanced)", "label": "producer,advanced", "required": false, "type": "boolean", "javaType": "boolean", "deprecated": false, "autowired": false, "secret": false, "defaultValue": false, "description": "Whether the producer should be started lazy (on the first message). By starting lazy you can use this to allow CamelContext and routes to startup in situations where a producer may other [...]
+ "documentClass": { "kind": "parameter", "displayName": "Document Class", "group": "advanced", "label": "advanced", "required": false, "type": "string", "javaType": "java.lang.Class<java.lang.Object>", "deprecated": false, "autowired": false, "secret": false, "defaultValue": "ObjectNode", "configurationClass": "org.apache.camel.component.elasticsearch.ElasticsearchConfiguration", "configurationField": "configuration", "description": "The class to use when deserializing the documents." },
+ "enableSniffer": { "kind": "parameter", "displayName": "Enable Sniffer", "group": "advanced", "label": "advanced", "required": false, "type": "boolean", "javaType": "boolean", "deprecated": false, "autowired": false, "secret": false, "defaultValue": false, "configurationClass": "org.apache.camel.component.elasticsearch.ElasticsearchConfiguration", "configurationField": "configuration", "description": "Enable automatically discover nodes from a running Elasticsearch cluster. If this o [...]
+ "sniffAfterFailureDelay": { "kind": "parameter", "displayName": "Sniff After Failure Delay", "group": "advanced", "label": "advanced", "required": false, "type": "integer", "javaType": "int", "deprecated": false, "autowired": false, "secret": false, "defaultValue": 60000, "configurationClass": "org.apache.camel.component.elasticsearch.ElasticsearchConfiguration", "configurationField": "configuration", "description": "The delay of a sniff execution scheduled after a failure (in millis [...]
+ "snifferInterval": { "kind": "parameter", "displayName": "Sniffer Interval", "group": "advanced", "label": "advanced", "required": false, "type": "integer", "javaType": "int", "deprecated": false, "autowired": false, "secret": false, "defaultValue": 300000, "configurationClass": "org.apache.camel.component.elasticsearch.ElasticsearchConfiguration", "configurationField": "configuration", "description": "The interval between consecutive ordinary sniff executions in milliseconds. Will b [...]
+ "certificatePath": { "kind": "parameter", "displayName": "Certificate Path", "group": "security", "label": "security", "required": false, "type": "string", "javaType": "java.lang.String", "deprecated": false, "autowired": false, "secret": false, "configurationClass": "org.apache.camel.component.elasticsearch.ElasticsearchConfiguration", "configurationField": "configuration", "description": "The path of the self-signed certificate to use to access to Elasticsearch." },
+ "enableSSL": { "kind": "parameter", "displayName": "Enable SSL", "group": "security", "label": "security", "required": false, "type": "boolean", "javaType": "boolean", "deprecated": false, "autowired": false, "secret": false, "defaultValue": false, "configurationClass": "org.apache.camel.component.elasticsearch.ElasticsearchConfiguration", "configurationField": "configuration", "description": "Enable SSL" }
+ }
+}
diff --git a/components/camel-elasticsearch/src/main/docs/elasticsearch-component.adoc b/components/camel-elasticsearch/src/main/docs/elasticsearch-component.adoc
new file mode 100644
index 00000000000..954f8de2a40
--- /dev/null
+++ b/components/camel-elasticsearch/src/main/docs/elasticsearch-component.adoc
@@ -0,0 +1,277 @@
+= Elasticsearch Component
+:doctitle: Elasticsearch
+:shortname: elasticsearch
+:artifactid: camel-elasticsearch
+:description: Send requests to ElasticSearch via Java Client API.
+:since: 3.20
+:supportlevel: Preview
+:component-header: Only producer is supported
+
+*Since Camel {since}*
+
+*{component-header}*
+
+The ElasticSearch component allows you to interface with an
+https://www.elastic.co/products/elasticsearch[ElasticSearch] 8.x API using the Java API Client library.
+
+Maven users will need to add the following dependency to their `pom.xml`
+for this component:
+
+[source,xml]
+----
+<dependency>
+ <groupId>org.apache.camel</groupId>
+ <artifactId>camel-elasticsearch</artifactId>
+ <version>x.x.x</version>
+ <!-- use the same version as your Camel core version -->
+</dependency>
+----
+
+== URI format
+
+----
+elasticsearch://clusterName[?options]
+----
+
+
+// component-configure options: START
+
+// component-configure options: END
+
+// component options: START
+include::partial$component-configure-options.adoc[]
+include::partial$component-endpoint-options.adoc[]
+// component options: END
+
+// endpoint options: START
+
+// endpoint options: END
+
+// component headers: START
+include::partial$component-endpoint-headers.adoc[]
+// component headers: END
+
+== Message Operations
+
+The following ElasticSearch operations are currently supported. Simply
+set an endpoint URI option or exchange header with a key of "operation"
+and a value set to one of the following. Some operations also require
+other parameters or the message body to be set.
+
+[width="100%",cols="10%,10%,80%",options="header",]
+|===
+|operation |message body |description
+
+|Index |*Map*, *String*, *byte[]*, *Reader*, *InputStream* or *IndexRequest.Builder* content to index |Adds content to an index and returns the content's indexId in the body.
+You can set the name of the target index by setting the message header with the key "indexName".
+You can set the indexId by setting the message header with
+the key "indexId".
+
+|GetById |*String* or *GetRequest.Builder* index id of content to retrieve |Retrieves the document corresponding to the given index id and returns a GetResponse object in the body.
+You can set the name of the target index by setting the message header with the key "indexName".
+You can set the type of document by setting the message header with
+the key "documentClass".
+
+|Delete |*String* or *DeleteRequest.Builder* index id of content to delete |Deletes the specified indexName and returns a Result object in the body.
+You can set the name of the target index by setting the message header with the key "indexName".
+
+|DeleteIndex |*String* or *DeleteIndexRequest.Builder* index name of the index to delete |Deletes the specified indexName and returns a status code in the body.
+You can set the name of the target index by setting the message header with the key "indexName".
+
+|Bulk |*Iterable* or *BulkRequest.Builder* of any type that is already accepted (DeleteOperation.Builder for delete operation, UpdateOperation.Builder for update operation, CreateOperation.Builder for create operation, byte[], InputStream, String, Reader, Map or any document type for index operation) | Adds/Updates/Deletes content from/to an index and returns a List<BulkResponseItem> object in the body
+You can set the name of the target index by setting the message header with the key "indexName".
+
+|Search |*Map*, *String* or *SearchRequest.Builder* |Search the content with the map of query string.
+You can set the name of the target index by setting the message header with the key "indexName".
+You can set the number of hits to return by setting the message header with the key "size".
+You can set the starting document offset by setting the message header with the key "from".
+
+|MultiSearch |*MsearchRequest.Builder* |Multiple search in one
+
+|MultiGet |*Iterable<String>* or *MgetRequest.Builder* the id of the document to retrieve |Multiple get in one
+
+You can set the name of the target index by setting the message header with the key "indexName".
+
+|Exists |None |Checks whether the index exists or not and returns a Boolean flag in the body.
+
+You must set the name of the target index by setting the message header with the key "indexName".
+
+|Update |*byte[]*, *InputStream*, *String*, *Reader*, *Map* or any document type content to update |Updates content to an index and returns the content's indexId in the body.
+You can set the name of the target index by setting the message header with the key "indexName".
+You can set the indexId by setting the message header with
+the key "indexId".
+
+|Ping |None |Pings the Elasticsearch cluster and returns true if the ping succeeded, false otherwise
+
+|===
+
+== Configure the component and enable basic authentication
+To use the Elasticsearch component it has to be configured with a minimum configuration.
+
+[source,java]
+----
+ElasticsearchComponent elasticsearchComponent = new ElasticsearchComponent();
+elasticsearchComponent.setHostAddresses("myelkhost:9200");
+camelContext.addComponent("elasticsearch", elasticsearchComponent);
+----
+
+For basic authentication with elasticsearch or using reverse http proxy in front of the elasticsearch cluster, simply setup
+basic authentication and SSL on the component like the example below
+
+[source,java]
+----
+ElasticsearchComponent elasticsearchComponent = new ElasticsearchComponent();
+elasticsearchComponent.setHostAddresses("myelkhost:9200");
+elasticsearchComponent.setUser("elkuser");
+elasticsearchComponent.setPassword("secure!!");
+elasticsearchComponent.setEnableSSL(true);
+elasticsearchComponent.setCertificatePath(certPath);
+
+camelContext.addComponent("elasticsearch", elasticsearchComponent);
+----
+
+== Index Example
+
+Below is a simple INDEX example
+
+[source,java]
+----
+from("direct:index")
+ .to("elasticsearch://elasticsearch?operation=Index&indexName=twitter");
+----
+
+[source,xml]
+----
+<route>
+ <from uri="direct:index"/>
+ <to uri="elasticsearch://elasticsearch?operation=Index&indexName=twitter"/>
+</route>
+----
+
+*For this operation you'll need to specify a indexId header.*
+
+A client would simply need to pass a body message containing a Map to
+the route. The result body contains the indexId created.
+
+[source,java]
+----
+Map<String, String> map = new HashMap<String, String>();
+map.put("content", "test");
+String indexId = template.requestBody("direct:index", map, String.class);
+----
+
+== Search Example
+
+Searching on specific field(s) and value use the Operation ´Search´.
+Pass in the query JSON String or the Map
+
+[source,java]
+----
+from("direct:search")
+ .to("elasticsearch://elasticsearch?operation=Search&indexName=twitter");
+----
+
+[source,xml]
+----
+<route>
+ <from uri="direct:search"/>
+ <to uri="elasticsearch://elasticsearch?operation=Search&indexName=twitter"/>
+</route>
+----
+
+[source,java]
+----
+String query = "{\"query\":{\"match\":{\"doc.content\":\"new release of ApacheCamel\"}}}";
+HitsMetadata<?> response = template.requestBody("direct:search", query, HitsMetadata.class);
+
+----
+
+Search on specific field(s) using Map.
+
+[source,java]
+----
+Map<String, Object> actualQuery = new HashMap<>();
+actualQuery.put("doc.content", "new release of ApacheCamel");
+
+Map<String, Object> match = new HashMap<>();
+match.put("match", actualQuery);
+
+Map<String, Object> query = new HashMap<>();
+query.put("query", match);
+HitsMetadata<?> response = template.requestBody("direct:search", query, HitsMetadata.class);
+
+----
+
+Search using Elasticsearch scroll api in order to fetch all results.
+
+[source,java]
+----
+from("direct:search")
+ .to("elasticsearch://elasticsearch?operation=Search&indexName=twitter&useScroll=true&scrollKeepAliveMs=30000");
+----
+
+[source,xml]
+----
+<route>
+ <from uri="direct:search"/>
+ <to uri="elasticsearch://elasticsearch?operation=Search&indexName=twitter&useScroll=true&scrollKeepAliveMs=30000"/>
+</route>
+----
+
+[source,java]
+----
+String query = "{\"query\":{\"match\":{\"doc.content\":\"new release of ApacheCamel\"}}}";
+try (ElasticsearchScrollRequestIterator response = template.requestBody("direct:search", query, ElasticsearchScrollRequestIterator.class)) {
+ // do something smart with results
+}
+----
+
+xref:eips:split-eip.adoc[Split EIP] can also be used.
+
+[source,java]
+----
+from("direct:search")
+ .to("elasticsearch://elasticsearch?operation=Search&indexName=twitter&useScroll=true&scrollKeepAliveMs=30000")
+ .split()
+ .body()
+ .streaming()
+ .to("mock:output")
+ .end();
+----
+
+== MultiSearch Example
+
+MultiSearching on specific field(s) and value use the Operation ´MultiSearch´.
+Pass in the MultiSearchRequest instance
+
+[source,java]
+----
+from("direct:multiSearch")
+ .to("elasticsearch://elasticsearch?operation=MultiSearch");
+----
+
+[source,xml]
+----
+<route>
+ <from uri="direct:multiSearch"/>
+ <to uri="elasticsearch://elasticsearch?operation=MultiSearch"/>
+</route>
+----
+
+MultiSearch on specific field(s)
+
+[source,java]
+----
+MsearchRequest.Builder builder = new MsearchRequest.Builder().index("twitter").searches(
+ new RequestItem.Builder().header(new MultisearchHeader.Builder().build())
+ .body(new MultisearchBody.Builder().query(b -> b.matchAll(x -> x)).build()).build(),
+ new RequestItem.Builder().header(new MultisearchHeader.Builder().build())
+ .body(new MultisearchBody.Builder().query(b -> b.matchAll(x -> x)).build()).build());
+List<MultiSearchResponseItem<?>> response = template.requestBody("direct:multiSearch", builder, List.class);
+----
+
+== Document type
+
+For all the search operations, it is possible to indicate the type of document to retrieve in order to get the result already unmarshalled with the expected type.
+
+The document type can be set using the header "documentClass" or via the uri parameter of the same name.
diff --git a/components/camel-elasticsearch/src/main/java/org/apache/camel/component/elasticsearch/ElasticsearchComponent.java b/components/camel-elasticsearch/src/main/java/org/apache/camel/component/elasticsearch/ElasticsearchComponent.java
new file mode 100644
index 00000000000..801a2d4ee3d
--- /dev/null
+++ b/components/camel-elasticsearch/src/main/java/org/apache/camel/component/elasticsearch/ElasticsearchComponent.java
@@ -0,0 +1,251 @@
+/*
+ * 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.camel.component.elasticsearch;
+
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.List;
+import java.util.Map;
+
+import org.apache.camel.CamelContext;
+import org.apache.camel.Endpoint;
+import org.apache.camel.spi.Metadata;
+import org.apache.camel.spi.annotations.Component;
+import org.apache.camel.support.DefaultComponent;
+import org.apache.http.HttpHost;
+import org.elasticsearch.client.RestClient;
+
+/**
+ * Represents the component that manages {@link ElasticsearchEndpoint}.
+ */
+@Component("elasticsearch")
+public class ElasticsearchComponent extends DefaultComponent {
+
+ @Metadata(label = "advanced", autowired = true)
+ private RestClient client;
+ @Metadata
+ private String hostAddresses;
+ @Metadata(defaultValue = "" + ElasticsearchConstants.DEFAULT_SOCKET_TIMEOUT)
+ private int socketTimeout = ElasticsearchConstants.DEFAULT_SOCKET_TIMEOUT;
+ @Metadata(defaultValue = "" + ElasticsearchConstants.MAX_RETRY_TIMEOUT)
+ private int maxRetryTimeout = ElasticsearchConstants.MAX_RETRY_TIMEOUT;
+ @Metadata(defaultValue = "" + ElasticsearchConstants.DEFAULT_CONNECTION_TIMEOUT)
+ private int connectionTimeout = ElasticsearchConstants.DEFAULT_CONNECTION_TIMEOUT;
+ @Metadata(label = "security", secret = true)
+ private String user;
+ @Metadata(label = "security", secret = true)
+ private String password;
+ @Metadata(label = "security")
+ private boolean enableSSL;
+ @Metadata(label = "security")
+ private String certificatePath;
+ @Metadata(label = "advanced")
+ private boolean enableSniffer;
+ @Metadata(label = "advanced", defaultValue = "" + ElasticsearchConstants.DEFAULT_SNIFFER_INTERVAL)
+ private int snifferInterval = ElasticsearchConstants.DEFAULT_SNIFFER_INTERVAL;
+ @Metadata(label = "advanced", defaultValue = "" + ElasticsearchConstants.DEFAULT_AFTER_FAILURE_DELAY)
+ private int sniffAfterFailureDelay = ElasticsearchConstants.DEFAULT_AFTER_FAILURE_DELAY;
+
+ public ElasticsearchComponent() {
+ this(null);
+ }
+
+ public ElasticsearchComponent(CamelContext context) {
+ super(context);
+ registerExtension(new ElasticsearchComponentVerifierExtension());
+ }
+
+ @Override
+ protected Endpoint createEndpoint(String uri, String remaining, Map<String, Object> parameters) throws Exception {
+ ElasticsearchConfiguration config = new ElasticsearchConfiguration();
+ config.setHostAddresses(this.getHostAddresses());
+ config.setSocketTimeout(this.getSocketTimeout());
+ config.setMaxRetryTimeout(this.getMaxRetryTimeout());
+ config.setConnectionTimeout(this.getConnectionTimeout());
+ config.setUser(this.getUser());
+ config.setEnableSSL(this.isEnableSSL());
+ config.setPassword(this.getPassword());
+ config.setCertificatePath(this.getCertificatePath());
+ config.setEnableSniffer(this.isEnableSniffer());
+ config.setSnifferInterval(this.getSnifferInterval());
+ config.setSniffAfterFailureDelay(this.getSniffAfterFailureDelay());
+ config.setClusterName(remaining);
+
+ Endpoint endpoint = new ElasticsearchEndpoint(uri, this, config, client);
+ setProperties(endpoint, parameters);
+ config.setHostAddressesList(parseHostAddresses(config.getHostAddresses(), config));
+
+ return endpoint;
+ }
+
+ private List<HttpHost> parseHostAddresses(String ipsString, ElasticsearchConfiguration config) {
+ if (ipsString == null || ipsString.isEmpty()) {
+ return null;
+ }
+ List<String> addressesStr = Arrays.asList(ipsString.split(","));
+ List<HttpHost> addressesTrAd = new ArrayList<>(addressesStr.size());
+ for (String address : addressesStr) {
+ String[] split = address.split(":");
+ String hostname;
+ if (split.length > 0) {
+ hostname = split[0];
+ } else {
+ throw new IllegalArgumentException();
+ }
+ int port = split.length > 1 ? Integer.parseInt(split[1]) : ElasticsearchConstants.DEFAULT_PORT;
+ addressesTrAd.add(new HttpHost(hostname, port, config.isEnableSSL() ? "HTTPS" : "HTTP"));
+ }
+ return addressesTrAd;
+ }
+
+ public RestClient getClient() {
+ return client;
+ }
+
+ /**
+ * To use an existing configured Elasticsearch client, instead of creating a client per endpoint. This allow to
+ * customize the client with specific settings.
+ */
+ public void setClient(RestClient client) {
+ this.client = client;
+ }
+
+ /**
+ * Comma separated list with ip:port formatted remote transport addresses to use. The ip and port options must be
+ * left blank for hostAddresses to be considered instead.
+ */
+ public String getHostAddresses() {
+ return hostAddresses;
+ }
+
+ public void setHostAddresses(String hostAddresses) {
+ this.hostAddresses = hostAddresses;
+ }
+
+ /**
+ * The timeout in ms to wait before the socket will timeout.
+ */
+ public int getSocketTimeout() {
+ return socketTimeout;
+ }
+
+ public void setSocketTimeout(int socketTimeout) {
+ this.socketTimeout = socketTimeout;
+ }
+
+ /**
+ * The time in ms to wait before connection will timeout.
+ */
+ public int getConnectionTimeout() {
+ return connectionTimeout;
+ }
+
+ public void setConnectionTimeout(int connectionTimeout) {
+ this.connectionTimeout = connectionTimeout;
+ }
+
+ /**
+ * Basic authenticate user
+ */
+ public String getUser() {
+ return user;
+ }
+
+ public void setUser(String user) {
+ this.user = user;
+ }
+
+ /**
+ * Password for authenticate
+ */
+ public String getPassword() {
+ return password;
+ }
+
+ public void setPassword(String password) {
+ this.password = password;
+ }
+
+ public boolean isEnableSSL() {
+ return enableSSL;
+ }
+
+ /**
+ * Enable SSL
+ */
+ public void setEnableSSL(boolean enableSSL) {
+ this.enableSSL = enableSSL;
+ }
+
+ /**
+ * The path of the self-signed certificate to use to access to Elasticsearch.
+ */
+ public String getCertificatePath() {
+ return certificatePath;
+ }
+
+ public void setCertificatePath(String certificatePath) {
+ this.certificatePath = certificatePath;
+ }
+
+ /**
+ * The time in ms before retry
+ */
+ public int getMaxRetryTimeout() {
+ return maxRetryTimeout;
+ }
+
+ public void setMaxRetryTimeout(int maxRetryTimeout) {
+ this.maxRetryTimeout = maxRetryTimeout;
+ }
+
+ public boolean isEnableSniffer() {
+ return enableSniffer;
+ }
+
+ /**
+ * Enable automatically discover nodes from a running Elasticsearch cluster. If this option is used in conjunction
+ * with Spring Boot then it's managed by the Spring Boot configuration (see: Disable Sniffer in Spring Boot).
+ */
+ public void setEnableSniffer(boolean enableSniffer) {
+ this.enableSniffer = enableSniffer;
+ }
+
+ /**
+ * The interval between consecutive ordinary sniff executions in milliseconds. Will be honoured when sniffOnFailure
+ * is disabled or when there are no failures between consecutive sniff executions
+ */
+ public int getSnifferInterval() {
+ return snifferInterval;
+ }
+
+ public void setSnifferInterval(int snifferInterval) {
+ this.snifferInterval = snifferInterval;
+ }
+
+ /**
+ * The delay of a sniff execution scheduled after a failure (in milliseconds)
+ */
+ public int getSniffAfterFailureDelay() {
+ return sniffAfterFailureDelay;
+ }
+
+ public void setSniffAfterFailureDelay(int sniffAfterFailureDelay) {
+ this.sniffAfterFailureDelay = sniffAfterFailureDelay;
+ }
+
+}
diff --git a/components/camel-elasticsearch/src/main/java/org/apache/camel/component/elasticsearch/ElasticsearchComponentVerifierExtension.java b/components/camel-elasticsearch/src/main/java/org/apache/camel/component/elasticsearch/ElasticsearchComponentVerifierExtension.java
new file mode 100644
index 00000000000..d79d363dc5d
--- /dev/null
+++ b/components/camel-elasticsearch/src/main/java/org/apache/camel/component/elasticsearch/ElasticsearchComponentVerifierExtension.java
@@ -0,0 +1,89 @@
+/*
+ * 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.camel.component.elasticsearch;
+
+import java.io.IOException;
+import java.util.Map;
+
+import co.elastic.clients.elasticsearch.ElasticsearchClient;
+import co.elastic.clients.json.jackson.JacksonJsonpMapper;
+import co.elastic.clients.transport.ElasticsearchTransport;
+import co.elastic.clients.transport.rest_client.RestClientTransport;
+import org.apache.camel.component.extension.verifier.DefaultComponentVerifierExtension;
+import org.apache.camel.component.extension.verifier.ResultBuilder;
+import org.apache.camel.component.extension.verifier.ResultErrorBuilder;
+import org.apache.camel.component.extension.verifier.ResultErrorHelper;
+import org.apache.http.HttpHost;
+import org.elasticsearch.client.RestClient;
+import org.elasticsearch.client.RestClientBuilder;
+
+public class ElasticsearchComponentVerifierExtension extends DefaultComponentVerifierExtension {
+
+ public ElasticsearchComponentVerifierExtension() {
+ this("elasticsearch");
+ }
+
+ public ElasticsearchComponentVerifierExtension(String scheme) {
+ super(scheme);
+ }
+
+ // *********************************
+ // Parameters validation
+ // *********************************
+
+ @Override
+ protected Result verifyParameters(Map<String, Object> parameters) {
+
+ ResultBuilder builder = ResultBuilder.withStatusAndScope(Result.Status.OK, Scope.PARAMETERS)
+ .error(ResultErrorHelper.requiresOption("clusterName", parameters))
+ .error(ResultErrorHelper.requiresOption("hostAddresses", parameters));
+ // Validate using the catalog
+
+ super.verifyParametersAgainstCatalog(builder, parameters);
+
+ return builder.build();
+ }
+
+ // *********************************
+ // Connectivity validation
+ // *********************************
+
+ @Override
+ protected Result verifyConnectivity(Map<String, Object> parameters) {
+ ResultBuilder builder = ResultBuilder.withStatusAndScope(Result.Status.OK, Scope.CONNECTIVITY);
+
+ try {
+ ElasticsearchConfiguration configuration = setProperties(new ElasticsearchConfiguration(), parameters);
+ RestClientBuilder clientBuilder = RestClient.builder(configuration.getHostAddressesList().toArray(new HttpHost[0]));
+ try (ElasticsearchTransport transport = new RestClientTransport(clientBuilder.build(), new JacksonJsonpMapper())) {
+ ElasticsearchClient esClient = new ElasticsearchClient(transport);
+ esClient.ping();
+ }
+ } catch (IOException e) {
+ ResultErrorBuilder errorBuilder
+ = ResultErrorBuilder.withCodeAndDescription(VerificationError.StandardCode.AUTHENTICATION, e.getMessage())
+ .detail("elasticsearch_exception_message", e.getMessage())
+ .detail(VerificationError.ExceptionAttribute.EXCEPTION_CLASS, e.getClass().getName())
+ .detail(VerificationError.ExceptionAttribute.EXCEPTION_INSTANCE, e);
+
+ builder.error(errorBuilder.build());
+ } catch (Exception e) {
+ builder.error(ResultErrorBuilder.withException(e).build());
+ }
+ return builder.build();
+ }
+}
diff --git a/components/camel-elasticsearch/src/main/java/org/apache/camel/component/elasticsearch/ElasticsearchConfiguration.java b/components/camel-elasticsearch/src/main/java/org/apache/camel/component/elasticsearch/ElasticsearchConfiguration.java
new file mode 100644
index 00000000000..7e8169aa8ed
--- /dev/null
+++ b/components/camel-elasticsearch/src/main/java/org/apache/camel/component/elasticsearch/ElasticsearchConfiguration.java
@@ -0,0 +1,315 @@
+/*
+ * 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.camel.component.elasticsearch;
+
+import java.util.List;
+
+import com.fasterxml.jackson.databind.node.ObjectNode;
+import org.apache.camel.spi.Metadata;
+import org.apache.camel.spi.UriParam;
+import org.apache.camel.spi.UriParams;
+import org.apache.camel.spi.UriPath;
+import org.apache.http.HttpHost;
+
+@UriParams
+public class ElasticsearchConfiguration {
+
+ private List<HttpHost> hostAddressesList;
+ private String user;
+ private String password;
+
+ @UriPath
+ @Metadata(required = true)
+ private String clusterName;
+ @UriParam
+ private ElasticsearchOperation operation;
+ @UriParam
+ private Integer size;
+ @UriParam
+ private Integer from;
+ @UriParam
+ private String indexName;
+ @UriParam(defaultValue = "" + ElasticsearchConstants.DEFAULT_FOR_WAIT_ACTIVE_SHARDS)
+ private int waitForActiveShards = ElasticsearchConstants.DEFAULT_FOR_WAIT_ACTIVE_SHARDS;
+ @UriParam
+ private String hostAddresses;
+ @UriParam(defaultValue = "" + ElasticsearchConstants.DEFAULT_SOCKET_TIMEOUT)
+ private int socketTimeout = ElasticsearchConstants.DEFAULT_SOCKET_TIMEOUT;
+ @UriParam(defaultValue = "" + ElasticsearchConstants.MAX_RETRY_TIMEOUT)
+ private int maxRetryTimeout = ElasticsearchConstants.MAX_RETRY_TIMEOUT;
+ @UriParam(defaultValue = "" + ElasticsearchConstants.DEFAULT_CONNECTION_TIMEOUT)
+ private int connectionTimeout = ElasticsearchConstants.DEFAULT_CONNECTION_TIMEOUT;
+ @UriParam
+ private boolean disconnect;
+ @UriParam(label = "security")
+ private boolean enableSSL;
+ @UriParam(label = "security")
+ private String certificatePath;
+ @UriParam
+ private boolean useScroll;
+ @UriParam(defaultValue = "" + ElasticsearchConstants.DEFAULT_SCROLL_KEEP_ALIVE_MS)
+ private int scrollKeepAliveMs = ElasticsearchConstants.DEFAULT_SCROLL_KEEP_ALIVE_MS;
+ @UriParam(label = "advanced")
+ private boolean enableSniffer;
+ @UriParam(label = "advanced", defaultValue = "" + ElasticsearchConstants.DEFAULT_SNIFFER_INTERVAL)
+ private int snifferInterval = ElasticsearchConstants.DEFAULT_SNIFFER_INTERVAL;
+ @UriParam(label = "advanced", defaultValue = "" + ElasticsearchConstants.DEFAULT_AFTER_FAILURE_DELAY)
+ private int sniffAfterFailureDelay = ElasticsearchConstants.DEFAULT_AFTER_FAILURE_DELAY;
+ @UriParam(label = "advanced", defaultValue = "ObjectNode")
+ private Class<?> documentClass = ObjectNode.class;
+
+ /**
+ * Starting index of the response.
+ */
+ public Integer getFrom() {
+ return from;
+ }
+
+ public void setFrom(Integer from) {
+ this.from = from;
+ }
+
+ /**
+ * Size of the response.
+ */
+ public Integer getSize() {
+ return size;
+ }
+
+ public void setSize(Integer size) {
+ this.size = size;
+ }
+
+ /**
+ * Name of the cluster
+ */
+ public String getClusterName() {
+ return clusterName;
+ }
+
+ public void setClusterName(String clusterName) {
+ this.clusterName = clusterName;
+ }
+
+ /**
+ * What operation to perform
+ */
+ public ElasticsearchOperation getOperation() {
+ return operation;
+ }
+
+ public void setOperation(ElasticsearchOperation operation) {
+ this.operation = operation;
+ }
+
+ /**
+ * The name of the index to act against
+ */
+ public String getIndexName() {
+ return indexName;
+ }
+
+ public void setIndexName(String indexName) {
+ this.indexName = indexName;
+ }
+
+ /**
+ * Comma separated list with ip:port formatted remote transport addresses to use.
+ */
+ public String getHostAddresses() {
+ return hostAddresses;
+ }
+
+ public void setHostAddresses(String hostAddresses) {
+ this.hostAddresses = hostAddresses;
+ }
+
+ /**
+ * Index creation waits for the write consistency number of shards to be available
+ */
+ public int getWaitForActiveShards() {
+ return waitForActiveShards;
+ }
+
+ public void setWaitForActiveShards(int waitForActiveShards) {
+ this.waitForActiveShards = waitForActiveShards;
+ }
+
+ public List<HttpHost> getHostAddressesList() {
+ return hostAddressesList;
+ }
+
+ public void setHostAddressesList(List<HttpHost> hostAddressesList) {
+ this.hostAddressesList = hostAddressesList;
+ }
+
+ /**
+ * The timeout in ms to wait before the socket will timeout.
+ */
+ public int getSocketTimeout() {
+ return socketTimeout;
+ }
+
+ public void setSocketTimeout(int socketTimeout) {
+ this.socketTimeout = socketTimeout;
+ }
+
+ /**
+ * The time in ms to wait before connection will timeout.
+ */
+ public int getConnectionTimeout() {
+ return connectionTimeout;
+ }
+
+ public void setConnectionTimeout(int connectionTimeout) {
+ this.connectionTimeout = connectionTimeout;
+ }
+
+ /**
+ * Basic authenticate user
+ */
+ public String getUser() {
+ return user;
+ }
+
+ public void setUser(String user) {
+ this.user = user;
+ }
+
+ /**
+ * Password for authenticate
+ */
+ public String getPassword() {
+ return password;
+ }
+
+ public void setPassword(String password) {
+ this.password = password;
+ }
+
+ /**
+ * Enable SSL
+ */
+ public boolean isEnableSSL() {
+ return enableSSL;
+ }
+
+ public void setEnableSSL(boolean enableSSL) {
+ this.enableSSL = enableSSL;
+ }
+
+ /**
+ * The path of the self-signed certificate to use to access to Elasticsearch.
+ */
+ public String getCertificatePath() {
+ return certificatePath;
+ }
+
+ public void setCertificatePath(String certificatePath) {
+ this.certificatePath = certificatePath;
+ }
+
+ /**
+ * The time in ms before retry
+ */
+ public int getMaxRetryTimeout() {
+ return maxRetryTimeout;
+ }
+
+ public void setMaxRetryTimeout(int maxRetryTimeout) {
+ this.maxRetryTimeout = maxRetryTimeout;
+ }
+
+ /**
+ * Disconnect after it finish calling the producer
+ */
+ public boolean isDisconnect() {
+ return disconnect;
+ }
+
+ public void setDisconnect(boolean disconnect) {
+ this.disconnect = disconnect;
+ }
+
+ /**
+ * Enable automatically discover nodes from a running Elasticsearch cluster. If this option is used in conjunction
+ * with Spring Boot then it's managed by the Spring Boot configuration (see: Disable Sniffer in Spring Boot).
+ */
+ public boolean isEnableSniffer() {
+ return enableSniffer;
+ }
+
+ public void setEnableSniffer(boolean enableSniffer) {
+ this.enableSniffer = enableSniffer;
+ }
+
+ /**
+ * The interval between consecutive ordinary sniff executions in milliseconds. Will be honoured when sniffOnFailure
+ * is disabled or when there are no failures between consecutive sniff executions
+ */
+ public int getSnifferInterval() {
+ return snifferInterval;
+ }
+
+ public void setSnifferInterval(int snifferInterval) {
+ this.snifferInterval = snifferInterval;
+ }
+
+ /**
+ * The delay of a sniff execution scheduled after a failure (in milliseconds)
+ */
+ public int getSniffAfterFailureDelay() {
+ return sniffAfterFailureDelay;
+ }
+
+ public void setSniffAfterFailureDelay(int sniffAfterFailureDelay) {
+ this.sniffAfterFailureDelay = sniffAfterFailureDelay;
+ }
+
+ /**
+ * Enable scroll usage
+ */
+ public boolean isUseScroll() {
+ return useScroll;
+ }
+
+ public void setUseScroll(boolean useScroll) {
+ this.useScroll = useScroll;
+ }
+
+ /**
+ * Time in ms during which elasticsearch will keep search context alive
+ */
+ public int getScrollKeepAliveMs() {
+ return scrollKeepAliveMs;
+ }
+
+ public void setScrollKeepAliveMs(int scrollKeepAliveMs) {
+ this.scrollKeepAliveMs = scrollKeepAliveMs;
+ }
+
+ /**
+ * The class to use when deserializing the documents.
+ */
+ public Class<?> getDocumentClass() {
+ return documentClass;
+ }
+
+ public void setDocumentClass(Class<?> documentClass) {
+ this.documentClass = documentClass;
+ }
+}
diff --git a/components/camel-elasticsearch/src/main/java/org/apache/camel/component/elasticsearch/ElasticsearchConstants.java b/components/camel-elasticsearch/src/main/java/org/apache/camel/component/elasticsearch/ElasticsearchConstants.java
new file mode 100644
index 00000000000..87ba4f71a20
--- /dev/null
+++ b/components/camel-elasticsearch/src/main/java/org/apache/camel/component/elasticsearch/ElasticsearchConstants.java
@@ -0,0 +1,55 @@
+/*
+ * 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.camel.component.elasticsearch;
+
+import org.apache.camel.spi.Metadata;
+
+public interface ElasticsearchConstants {
+
+ @Metadata(description = "The operation to perform",
+ javaType = "org.apache.camel.component.elasticsearch.ElasticsearchOperation")
+ String PARAM_OPERATION = "operation";
+ @Metadata(description = "The id of the indexed document.", javaType = "String")
+ String PARAM_INDEX_ID = "indexId";
+ @Metadata(description = "The name of the index to act against", javaType = "String")
+ String PARAM_INDEX_NAME = "indexName";
+ @Metadata(description = "The full qualified name of the class of the document to unmarshall", javaType = "Class",
+ defaultValue = "ObjectNode")
+ String PARAM_DOCUMENT_CLASS = "documentClass";
+ @Metadata(description = "The index creation waits for the write consistency number of shards to be available",
+ javaType = "Integer")
+ String PARAM_WAIT_FOR_ACTIVE_SHARDS = "waitForActiveShards";
+ @Metadata(description = "The starting index of the response.", javaType = "Integer")
+ String PARAM_SCROLL_KEEP_ALIVE_MS = "scrollKeepAliveMs";
+ @Metadata(description = "Set to true to enable scroll usage", javaType = "Boolean")
+ String PARAM_SCROLL = "useScroll";
+ @Metadata(description = "The size of the response.", javaType = "Integer")
+ String PARAM_SIZE = "size";
+ @Metadata(description = "The starting index of the response.", javaType = "Integer")
+ String PARAM_FROM = "from";
+
+ String PROPERTY_SCROLL_ES_QUERY_COUNT = "CamelElasticsearchScrollQueryCount";
+
+ int DEFAULT_PORT = 9200;
+ int DEFAULT_FOR_WAIT_ACTIVE_SHARDS = 1; // Meaning only wait for the primary shard
+ int DEFAULT_SOCKET_TIMEOUT = 30000; // Meaning how long time to wait before the socket timeout
+ int MAX_RETRY_TIMEOUT = 30000; // Meaning how long to wait before retry again
+ int DEFAULT_CONNECTION_TIMEOUT = 30000; // Meaning how many seconds before it timeout when establish connection
+ int DEFAULT_SNIFFER_INTERVAL = 60000 * 5; // Meaning how often it should search for elasticsearch nodes
+ int DEFAULT_AFTER_FAILURE_DELAY = 60000; // Meaning when should the sniff execution scheduled after a failure
+ int DEFAULT_SCROLL_KEEP_ALIVE_MS = 60000; // Meaning how many milliseconds elasticsearch will keep the search context
+}
diff --git a/components/camel-elasticsearch/src/main/java/org/apache/camel/component/elasticsearch/ElasticsearchEndpoint.java b/components/camel-elasticsearch/src/main/java/org/apache/camel/component/elasticsearch/ElasticsearchEndpoint.java
new file mode 100644
index 00000000000..971e40d2d38
--- /dev/null
+++ b/components/camel-elasticsearch/src/main/java/org/apache/camel/component/elasticsearch/ElasticsearchEndpoint.java
@@ -0,0 +1,65 @@
+/*
+ * 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.camel.component.elasticsearch;
+
+import org.apache.camel.Category;
+import org.apache.camel.Consumer;
+import org.apache.camel.Processor;
+import org.apache.camel.Producer;
+import org.apache.camel.spi.UriEndpoint;
+import org.apache.camel.spi.UriParam;
+import org.apache.camel.support.DefaultEndpoint;
+import org.elasticsearch.client.RestClient;
+
+/**
+ * Send requests to ElasticSearch via Java Client API.
+ */
+@UriEndpoint(firstVersion = "3.20.0", scheme = "elasticsearch", title = "Elasticsearch",
+ syntax = "elasticsearch:clusterName", producerOnly = true,
+ category = { Category.SEARCH, Category.MONITORING }, headersClass = ElasticsearchConstants.class)
+public class ElasticsearchEndpoint extends DefaultEndpoint {
+
+ @UriParam
+ private final ElasticsearchConfiguration configuration;
+
+ private final RestClient client;
+
+ public ElasticsearchEndpoint(String uri, ElasticsearchComponent component, ElasticsearchConfiguration config,
+ RestClient client) {
+ super(uri, component);
+ this.configuration = config;
+ this.client = client;
+ }
+
+ public ElasticsearchConfiguration getConfiguration() {
+ return configuration;
+ }
+
+ @Override
+ public Producer createProducer() {
+ return new ElasticsearchProducer(this, configuration);
+ }
+
+ @Override
+ public Consumer createConsumer(Processor processor) {
+ throw new UnsupportedOperationException("Cannot consume from an ElasticsearchEndpoint: " + getEndpointUri());
+ }
+
+ public RestClient getClient() {
+ return client;
+ }
+}
diff --git a/components/camel-elasticsearch/src/main/java/org/apache/camel/component/elasticsearch/ElasticsearchOperation.java b/components/camel-elasticsearch/src/main/java/org/apache/camel/component/elasticsearch/ElasticsearchOperation.java
new file mode 100644
index 00000000000..0e13aa4879e
--- /dev/null
+++ b/components/camel-elasticsearch/src/main/java/org/apache/camel/component/elasticsearch/ElasticsearchOperation.java
@@ -0,0 +1,64 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.camel.component.elasticsearch;
+
+/**
+ * The ElasticSearch server operations list which are implemented
+ * <p>
+ * <ul>
+ * <li>Index - Index a document associated with a given index</li>
+ * <li>Update - Updates a document based on a script</li>
+ * <li>Bulk - Executes a bulk of index / create/ update delete operations</li>
+ * <li>GetById - Get an indexed document from its id</li>
+ * <li>MultiGet - Multiple get documents</li>
+ * <li>Delete - Deletes a document from the index based on the index and id</li>
+ * <li>DeleteIndex - Deletes an index based on the index name</li>
+ * <li>MultiSearch - Multiple Search across one or more indices with a query</li>
+ * <li>Search - Search across one or more indices with a query</li>
+ * <li>Exists - Checks whether the index exists or not</li>
+ * <li>Ping - Pings the Elasticsearch cluster</li>
+ * </ul>
+ *
+ *
+ *
+ *
+ * (using search with size=0 and terminate_after=1 parameters)
+ */
+public enum ElasticsearchOperation {
+ Index("Index"),
+ Update("Update"),
+ Bulk("Bulk"),
+ GetById("GetById"),
+ MultiGet("MultiGet"),
+ MultiSearch("MultiSearch"),
+ Delete("Delete"),
+ DeleteIndex("DeleteIndex"),
+ Search("Search"),
+ Exists("Exists"),
+ Ping("Ping");
+
+ private final String text;
+
+ ElasticsearchOperation(final String text) {
+ this.text = text;
+ }
+
+ @Override
+ public String toString() {
+ return text;
+ }
+}
diff --git a/components/camel-elasticsearch/src/main/java/org/apache/camel/component/elasticsearch/ElasticsearchProducer.java b/components/camel-elasticsearch/src/main/java/org/apache/camel/component/elasticsearch/ElasticsearchProducer.java
new file mode 100644
index 00000000000..cacf78675f7
--- /dev/null
+++ b/components/camel-elasticsearch/src/main/java/org/apache/camel/component/elasticsearch/ElasticsearchProducer.java
@@ -0,0 +1,391 @@
+/*
+ * 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.camel.component.elasticsearch;
+
+import java.io.ByteArrayInputStream;
+import java.nio.file.Files;
+import java.nio.file.Paths;
+import java.security.KeyStore;
+import java.security.cert.Certificate;
+import java.security.cert.CertificateFactory;
+
+import javax.net.ssl.SSLContext;
+import javax.net.ssl.TrustManagerFactory;
+
+import co.elastic.clients.elasticsearch.ElasticsearchClient;
+import co.elastic.clients.elasticsearch.core.BulkRequest;
+import co.elastic.clients.elasticsearch.core.DeleteRequest;
+import co.elastic.clients.elasticsearch.core.GetRequest;
+import co.elastic.clients.elasticsearch.core.IndexRequest;
+import co.elastic.clients.elasticsearch.core.MgetRequest;
+import co.elastic.clients.elasticsearch.core.MsearchRequest;
+import co.elastic.clients.elasticsearch.core.SearchRequest;
+import co.elastic.clients.elasticsearch.core.UpdateRequest;
+import co.elastic.clients.elasticsearch.indices.DeleteIndexRequest;
+import co.elastic.clients.elasticsearch.indices.ExistsRequest;
+import co.elastic.clients.json.jackson.JacksonJsonpMapper;
+import co.elastic.clients.transport.ElasticsearchTransport;
+import co.elastic.clients.transport.rest_client.RestClientTransport;
+import com.fasterxml.jackson.databind.ObjectMapper;
+import com.fasterxml.jackson.databind.SerializationFeature;
+import org.apache.camel.Exchange;
+import org.apache.camel.Message;
+import org.apache.camel.support.DefaultProducer;
+import org.apache.camel.util.IOHelper;
+import org.apache.http.HttpHost;
+import org.apache.http.auth.AuthScope;
+import org.apache.http.auth.UsernamePasswordCredentials;
+import org.apache.http.client.CredentialsProvider;
+import org.apache.http.impl.client.BasicCredentialsProvider;
+import org.elasticsearch.client.RestClient;
+import org.elasticsearch.client.RestClientBuilder;
+import org.elasticsearch.client.sniff.Sniffer;
+import org.elasticsearch.client.sniff.SnifferBuilder;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import static org.apache.camel.component.elasticsearch.ElasticsearchConstants.PARAM_SCROLL;
+import static org.apache.camel.component.elasticsearch.ElasticsearchConstants.PARAM_SCROLL_KEEP_ALIVE_MS;
+
+/**
+ * Represents an Elasticsearch producer.
+ */
+public class ElasticsearchProducer extends DefaultProducer {
+
+ private static final Logger LOG = LoggerFactory.getLogger(ElasticsearchProducer.class);
+
+ protected final ElasticsearchConfiguration configuration;
+ private volatile RestClient client;
+ private volatile Sniffer sniffer;
+
+ public ElasticsearchProducer(ElasticsearchEndpoint endpoint, ElasticsearchConfiguration configuration) {
+ super(endpoint);
+ this.configuration = configuration;
+ this.client = endpoint.getClient();
+ }
+
+ private ElasticsearchOperation resolveOperation(Exchange exchange) {
+ // 1. Operation can be driven by either (in order of preference):
+ // a. If the body is an ActionRequest the operation is set by the type
+ // of request.
+ // b. If the body is not an ActionRequest, the operation is set by the
+ // header if it exists.
+ // c. If neither the operation can not be derived from the body or
+ // header, the configuration is used.
+ // In the event we can't discover the operation from a, b or c we throw
+ // an error.
+ Object request = exchange.getIn().getBody();
+ if (request instanceof IndexRequest) {
+ return ElasticsearchOperation.Index;
+ } else if (request instanceof GetRequest) {
+ return ElasticsearchOperation.GetById;
+ } else if (request instanceof MgetRequest) {
+ return ElasticsearchOperation.MultiGet;
+ } else if (request instanceof UpdateRequest) {
+ return ElasticsearchOperation.Update;
+ } else if (request instanceof BulkRequest) {
+ return ElasticsearchOperation.Bulk;
+ } else if (request instanceof DeleteRequest) {
+ return ElasticsearchOperation.Delete;
+ } else if (request instanceof SearchRequest) {
+ return ElasticsearchOperation.Search;
+ } else if (request instanceof MsearchRequest) {
+ return ElasticsearchOperation.MultiSearch;
+ } else if (request instanceof DeleteIndexRequest) {
+ return ElasticsearchOperation.DeleteIndex;
+ }
+
+ ElasticsearchOperation operationConfig
+ = exchange.getIn().getHeader(ElasticsearchConstants.PARAM_OPERATION, ElasticsearchOperation.class);
+ if (operationConfig == null) {
+ operationConfig = configuration.getOperation();
+ }
+ if (operationConfig == null) {
+ throw new IllegalArgumentException(
+ ElasticsearchConstants.PARAM_OPERATION + " value '" + operationConfig + "' is not supported");
+ }
+ return operationConfig;
+ }
+
+ @Override
+ public void process(Exchange exchange) throws Exception {
+ if (configuration.isDisconnect() && client == null) {
+ startClient();
+ }
+ final ObjectMapper mapper = new ObjectMapper();
+ mapper.configure(SerializationFeature.FAIL_ON_EMPTY_BEANS, false);
+ ElasticsearchTransport transport = new RestClientTransport(client, new JacksonJsonpMapper(mapper));
+ ElasticsearchClient esClient = new ElasticsearchClient(transport);
+ // 2. Index and type will be set by:
+ // a. If the incoming body is already an action request
+ // b. If the body is not an action request we will use headers if they
+ // are set.
+ // c. If the body is not an action request and the headers aren't set we
+ // will use the configuration.
+ // No error is thrown by the component in the event none of the above
+ // conditions are met. The java es client
+ // will throw.
+
+ Message message = exchange.getIn();
+ final ElasticsearchOperation operation = resolveOperation(exchange);
+
+ // Set the index/type headers on the exchange if necessary. This is used
+ // for type conversion.
+ boolean configIndexName = false;
+ String indexName = message.getHeader(ElasticsearchConstants.PARAM_INDEX_NAME, String.class);
+ if (indexName == null) {
+ message.setHeader(ElasticsearchConstants.PARAM_INDEX_NAME, configuration.getIndexName());
+ configIndexName = true;
+ }
+
+ Integer size = message.getHeader(ElasticsearchConstants.PARAM_SIZE, Integer.class);
+ if (size == null) {
+ message.setHeader(ElasticsearchConstants.PARAM_SIZE, configuration.getSize());
+ }
+
+ Integer from = message.getHeader(ElasticsearchConstants.PARAM_FROM, Integer.class);
+ if (from == null) {
+ message.setHeader(ElasticsearchConstants.PARAM_FROM, configuration.getFrom());
+ }
+
+ boolean configWaitForActiveShards = false;
+ Integer waitForActiveShards = message.getHeader(ElasticsearchConstants.PARAM_WAIT_FOR_ACTIVE_SHARDS, Integer.class);
+ if (waitForActiveShards == null) {
+ message.setHeader(ElasticsearchConstants.PARAM_WAIT_FOR_ACTIVE_SHARDS, configuration.getWaitForActiveShards());
+ configWaitForActiveShards = true;
+ }
+
+ Class<?> documentClass = message.getHeader(ElasticsearchConstants.PARAM_DOCUMENT_CLASS, Class.class);
+ if (documentClass == null) {
+ documentClass = configuration.getDocumentClass();
+ }
+
+ switch (operation) {
+ case Index: {
+ IndexRequest.Builder<?> indexRequestBuilder = message.getBody(IndexRequest.Builder.class);
+ message.setBody(esClient.index(indexRequestBuilder.build()).id());
+ break;
+ }
+ case Update: {
+ UpdateRequest.Builder updateRequestBuilder = message.getBody(UpdateRequest.Builder.class);
+ message.setBody(esClient.update(updateRequestBuilder.build(), documentClass).id());
+ break;
+ }
+ case GetById: {
+ GetRequest.Builder getRequestBuilder = message.getBody(GetRequest.Builder.class);
+ if (getRequestBuilder == null) {
+ throw new IllegalArgumentException(
+ "Wrong body type. Only String or GetRequest.Builder is allowed as a type");
+ }
+ message.setBody(esClient.get(getRequestBuilder.build(), documentClass));
+ break;
+ }
+ case Bulk: {
+ BulkRequest.Builder bulkRequestBuilder = message.getBody(BulkRequest.Builder.class);
+ if (bulkRequestBuilder == null) {
+ throw new IllegalArgumentException(
+ "Wrong body type. Only Iterable or BulkRequest.Builder is allowed as a type");
+ }
+ message.setBody(esClient.bulk(bulkRequestBuilder.build()).items());
+ break;
+ }
+ case Delete: {
+ DeleteRequest.Builder deleteRequestBuilder = message.getBody(DeleteRequest.Builder.class);
+ if (deleteRequestBuilder == null) {
+ throw new IllegalArgumentException(
+ "Wrong body type. Only String or DeleteRequest.Builder is allowed as a type");
+ }
+ message.setBody(esClient.delete(deleteRequestBuilder.build()).result());
+ break;
+ }
+ case DeleteIndex: {
+ DeleteIndexRequest.Builder deleteIndexRequestBuilder = message.getBody(DeleteIndexRequest.Builder.class);
+ if (deleteIndexRequestBuilder == null) {
+ throw new IllegalArgumentException(
+ "Wrong body type. Only String or DeleteIndexRequest.Builder is allowed as a type");
+ }
+ message.setBody(esClient.indices().delete(deleteIndexRequestBuilder.build()).acknowledged());
+ break;
+ }
+ case Exists: {
+ ExistsRequest.Builder builder = new ExistsRequest.Builder();
+ builder.index(exchange.getIn().getHeader(ElasticsearchConstants.PARAM_INDEX_NAME, String.class));
+ message.setBody(esClient.indices().exists(builder.build()).value());
+ break;
+ }
+ case Search: {
+ SearchRequest.Builder searchRequestBuilder = message.getBody(SearchRequest.Builder.class);
+ if (searchRequestBuilder == null) {
+ throw new IllegalArgumentException(
+ "Wrong body type. Only Map, String or SearchRequest.Builder is allowed as a type");
+ }
+ // is it a scroll request ?
+ boolean useScroll = message.getHeader(PARAM_SCROLL, configuration.isUseScroll(), Boolean.class);
+ if (useScroll) {
+ int scrollKeepAliveMs
+ = message.getHeader(PARAM_SCROLL_KEEP_ALIVE_MS, configuration.getScrollKeepAliveMs(),
+ Integer.class);
+ ElasticsearchScrollRequestIterator<?> scrollRequestIterator = new ElasticsearchScrollRequestIterator<>(
+ searchRequestBuilder, esClient, scrollKeepAliveMs, exchange, documentClass);
+ exchange.getIn().setBody(scrollRequestIterator);
+ } else {
+ message.setBody(esClient.search(searchRequestBuilder.build(), documentClass).hits());
+ }
+ break;
+ }
+ case MultiSearch: {
+ MsearchRequest.Builder msearchRequestBuilder = message.getBody(MsearchRequest.Builder.class);
+ if (msearchRequestBuilder == null) {
+ throw new IllegalArgumentException("Wrong body type. Only MsearchRequest.Builder is allowed as a type");
+ }
+ message.setBody(esClient.msearch(msearchRequestBuilder.build(), documentClass).responses());
+ break;
+ }
+ case MultiGet: {
+ MgetRequest.Builder mgetRequestBuilder = message.getBody(MgetRequest.Builder.class);
+ if (mgetRequestBuilder == null) {
+ throw new IllegalArgumentException("Wrong body type. Only MgetRequest.Builder is allowed as a type");
+ }
+ message.setBody(esClient.mget(mgetRequestBuilder.build(), documentClass).docs());
+ break;
+ }
+ case Ping: {
+ message.setBody(esClient.ping().value());
+ break;
+ }
+ default: {
+ throw new IllegalArgumentException(
+ ElasticsearchConstants.PARAM_OPERATION + " value '" + operation + "' is not supported");
+ }
+ }
+
+ // If we set params via the configuration on this exchange, remove them
+ // now. This preserves legacy behavior for this component and enables a
+ // use case where one message can be sent to multiple elasticsearch
+ // endpoints where the user is relying on the endpoint configuration
+ // (index/type) rather than header values. If we do not clear this out
+ // sending the same message (index request, for example) to multiple
+ // elasticsearch endpoints would have the effect overriding any
+ // subsequent endpoint index/type with the first endpoint index/type.
+ if (configIndexName) {
+ message.removeHeader(ElasticsearchConstants.PARAM_INDEX_NAME);
+ }
+
+ if (configWaitForActiveShards) {
+ message.removeHeader(ElasticsearchConstants.PARAM_WAIT_FOR_ACTIVE_SHARDS);
+ }
+ if (configuration.isDisconnect()) {
+ IOHelper.close(transport);
+ IOHelper.close(client);
+ client = null;
+ if (configuration.isEnableSniffer()) {
+ IOHelper.close(sniffer);
+ sniffer = null;
+ }
+ }
+
+ }
+
+ @Override
+ protected void doStart() throws Exception {
+ super.doStart();
+ if (!configuration.isDisconnect()) {
+ startClient();
+ }
+ }
+
+ private void startClient() {
+ if (client == null) {
+ LOG.info("Connecting to the ElasticSearch cluster: {}", configuration.getClusterName());
+ if (configuration.getHostAddressesList() != null
+ && !configuration.getHostAddressesList().isEmpty()) {
+ client = createClient();
+ } else {
+ LOG.warn("Incorrect ip address and port parameters settings for ElasticSearch cluster");
+ }
+ }
+ }
+
+ private RestClient createClient() {
+ final RestClientBuilder builder = RestClient.builder(configuration.getHostAddressesList().toArray(new HttpHost[0]));
+
+ builder.setRequestConfigCallback(requestConfigBuilder -> requestConfigBuilder
+ .setConnectTimeout(configuration.getConnectionTimeout()).setSocketTimeout(configuration.getSocketTimeout()));
+ if (configuration.getUser() != null && configuration.getPassword() != null) {
+ final CredentialsProvider credentialsProvider = new BasicCredentialsProvider();
+ credentialsProvider.setCredentials(AuthScope.ANY,
+ new UsernamePasswordCredentials(configuration.getUser(), configuration.getPassword()));
+ builder.setHttpClientConfigCallback(httpClientBuilder -> {
+ httpClientBuilder.setDefaultCredentialsProvider(credentialsProvider);
+ if (configuration.getCertificatePath() != null) {
+ httpClientBuilder.setSSLContext(createSslContextFromCa());
+ }
+ return httpClientBuilder;
+ });
+ }
+ final RestClient restClient = builder.build();
+ if (configuration.isEnableSniffer()) {
+ SnifferBuilder snifferBuilder = Sniffer.builder(restClient);
+ snifferBuilder.setSniffIntervalMillis(configuration.getSnifferInterval());
+ snifferBuilder.setSniffAfterFailureDelayMillis(configuration.getSniffAfterFailureDelay());
+ sniffer = snifferBuilder.build();
+ }
+ return restClient;
+ }
+
+ @Override
+ protected void doStop() throws Exception {
+ if (client != null) {
+ LOG.info("Disconnecting from ElasticSearch cluster: {}", configuration.getClusterName());
+ client.close();
+ if (sniffer != null) {
+ sniffer.close();
+ }
+ }
+ super.doStop();
+ }
+
+ public RestClient getClient() {
+ return client;
+ }
+
+ /**
+ * A SSL context based on the self-signed CA, so that using this SSL Context allows to connect to the Elasticsearch
+ * service
+ *
+ * @return a customized SSL Context
+ */
+ private SSLContext createSslContextFromCa() {
+ try {
+ CertificateFactory factory = CertificateFactory.getInstance("X.509");
+ Certificate trustedCa = factory.generateCertificate(
+ new ByteArrayInputStream(Files.readAllBytes(Paths.get(configuration.getCertificatePath()))));
+ KeyStore trustStore = KeyStore.getInstance("pkcs12");
+ trustStore.load(null, null);
+ trustStore.setCertificateEntry("ca", trustedCa);
+
+ final SSLContext sslContext = SSLContext.getInstance("TLSv1.3");
+ TrustManagerFactory trustManagerFactory
+ = TrustManagerFactory.getInstance(TrustManagerFactory.getDefaultAlgorithm());
+ trustManagerFactory.init(trustStore);
+ sslContext.init(null, trustManagerFactory.getTrustManagers(), null);
+ return sslContext;
+ } catch (Exception e) {
+ throw new RuntimeException(e);
+ }
+ }
+}
diff --git a/components/camel-elasticsearch/src/main/java/org/apache/camel/component/elasticsearch/ElasticsearchScrollRequestIterator.java b/components/camel-elasticsearch/src/main/java/org/apache/camel/component/elasticsearch/ElasticsearchScrollRequestIterator.java
new file mode 100644
index 00000000000..e55190d1ae5
--- /dev/null
+++ b/components/camel-elasticsearch/src/main/java/org/apache/camel/component/elasticsearch/ElasticsearchScrollRequestIterator.java
@@ -0,0 +1,150 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.camel.component.elasticsearch;
+
+import java.io.Closeable;
+import java.io.IOException;
+import java.util.Iterator;
+import java.util.List;
+
+import co.elastic.clients.elasticsearch.ElasticsearchClient;
+import co.elastic.clients.elasticsearch._types.Time;
+import co.elastic.clients.elasticsearch.core.ClearScrollRequest;
+import co.elastic.clients.elasticsearch.core.ScrollRequest;
+import co.elastic.clients.elasticsearch.core.ScrollResponse;
+import co.elastic.clients.elasticsearch.core.SearchRequest;
+import co.elastic.clients.elasticsearch.core.SearchResponse;
+import co.elastic.clients.elasticsearch.core.search.Hit;
+import org.apache.camel.Exchange;
+
+import static org.apache.camel.component.elasticsearch.ElasticsearchConstants.PROPERTY_SCROLL_ES_QUERY_COUNT;
+
+public class ElasticsearchScrollRequestIterator<TDocument> implements Iterator<Hit<TDocument>>, Closeable {
+
+ private final SearchRequest searchRequest;
+ private final ElasticsearchClient esClient;
+ private final Class<TDocument> documentClass;
+ private Iterator<? extends Hit<TDocument>> currentSearchHits;
+ private final int scrollKeepAliveMs;
+ private final Exchange exchange;
+ private String scrollId;
+ private boolean closed;
+ private int requestCount;
+
+ public ElasticsearchScrollRequestIterator(SearchRequest.Builder searchRequestBuilder, ElasticsearchClient esClient,
+ int scrollKeepAliveMs, Exchange exchange, Class<TDocument> documentClass) {
+ // add scroll option on the first query
+ this.searchRequest = searchRequestBuilder
+ .scroll(Time.of(b -> b.time(String.format("%sms", scrollKeepAliveMs))))
+ .build();
+ this.esClient = esClient;
+ this.scrollKeepAliveMs = scrollKeepAliveMs;
+ this.exchange = exchange;
+ this.closed = false;
+ this.requestCount = 0;
+ this.documentClass = documentClass;
+
+ setFirstCurrentSearchHits();
+ }
+
+ @Override
+ public boolean hasNext() {
+ if (closed) {
+ return false;
+ }
+
+ boolean hasNext = currentSearchHits.hasNext();
+ if (!hasNext) {
+ updateCurrentSearchHits();
+
+ hasNext = currentSearchHits.hasNext();
+ }
+
+ return hasNext;
+ }
+
+ @Override
+ public Hit<TDocument> next() {
+ return closed ? null : currentSearchHits.next();
+ }
+
+ /**
+ * Execute next Elasticsearch scroll request and update the current scroll result.
+ */
+ private void updateCurrentSearchHits() {
+ ScrollResponse<TDocument> scrollResponse = scrollSearch();
+ this.currentSearchHits = scrollResponse.hits().hits().iterator();
+ }
+
+ private void setFirstCurrentSearchHits() {
+ SearchResponse<TDocument> searchResponse = firstSearch();
+ this.currentSearchHits = searchResponse.hits().hits().iterator();
+ this.scrollId = searchResponse.scrollId();
+ }
+
+ private SearchResponse<TDocument> firstSearch() {
+ SearchResponse<TDocument> searchResponse;
+ try {
+ searchResponse = esClient.search(searchRequest, documentClass);
+ requestCount++;
+ } catch (IOException e) {
+ throw new IllegalStateException(e);
+ }
+ return searchResponse;
+ }
+
+ private ScrollResponse<TDocument> scrollSearch() {
+ ScrollResponse<TDocument> scrollResponse;
+ try {
+ ScrollRequest searchScrollRequest = new ScrollRequest.Builder()
+ .scroll(Time.of(b -> b.time(String.format("%sms", scrollKeepAliveMs))))
+ .scrollId(scrollId)
+ .build();
+
+ scrollResponse = esClient.scroll(searchScrollRequest, documentClass);
+ requestCount++;
+ } catch (IOException e) {
+ throw new IllegalStateException(e);
+ }
+ return scrollResponse;
+ }
+
+ @Override
+ public void close() {
+ if (!closed) {
+ try {
+ ClearScrollRequest clearScrollRequest = new ClearScrollRequest.Builder()
+ .scrollId(List.of(scrollId))
+ .build();
+
+ esClient.clearScroll(clearScrollRequest);
+ closed = true;
+ exchange.setProperty(PROPERTY_SCROLL_ES_QUERY_COUNT, requestCount);
+ } catch (IOException e) {
+ throw new IllegalStateException(e);
+ }
+ }
+ }
+
+ public int getRequestCount() {
+ return requestCount;
+ }
+
+ public boolean isClosed() {
+ return closed;
+ }
+}
diff --git a/components/camel-elasticsearch/src/main/java/org/apache/camel/component/elasticsearch/aggregation/BulkRequestAggregationStrategy.java b/components/camel-elasticsearch/src/main/java/org/apache/camel/component/elasticsearch/aggregation/BulkRequestAggregationStrategy.java
new file mode 100644
index 00000000000..34706104ad5
--- /dev/null
+++ b/components/camel-elasticsearch/src/main/java/org/apache/camel/component/elasticsearch/aggregation/BulkRequestAggregationStrategy.java
@@ -0,0 +1,50 @@
+/*
+ * 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.camel.component.elasticsearch.aggregation;
+
+import java.util.List;
+
+import co.elastic.clients.elasticsearch.core.BulkRequest;
+import co.elastic.clients.elasticsearch.core.bulk.BulkOperation;
+import org.apache.camel.AggregationStrategy;
+import org.apache.camel.Exchange;
+import org.apache.camel.InvalidPayloadRuntimeException;
+
+/**
+ * Aggregates two {@link BulkOperation}s into a single {@link BulkRequest}.
+ */
+public class BulkRequestAggregationStrategy implements AggregationStrategy {
+
+ @Override
+ public Exchange aggregate(Exchange oldExchange, Exchange newExchange) {
+ // Don't use getBody(Class<T>) here as we don't want to coerce the body type using a type converter.
+ Object objBody = newExchange.getIn().getBody();
+ if (!(objBody instanceof BulkOperation[])) {
+ throw new InvalidPayloadRuntimeException(newExchange, BulkOperation[].class);
+ }
+
+ BulkOperation[] newBody = (BulkOperation[]) objBody;
+ BulkRequest.Builder builder = new BulkRequest.Builder();
+ builder.operations(List.of(newBody));
+ if (oldExchange != null) {
+ BulkRequest request = oldExchange.getIn().getBody(BulkRequest.class);
+ builder.operations(request.operations());
+ }
+ newExchange.getIn().setBody(builder.build());
+ return oldExchange;
+ }
+}
diff --git a/components/camel-elasticsearch/src/main/java/org/apache/camel/component/elasticsearch/converter/ElasticsearchActionRequestConverter.java b/components/camel-elasticsearch/src/main/java/org/apache/camel/component/elasticsearch/converter/ElasticsearchActionRequestConverter.java
new file mode 100644
index 00000000000..ee08e1f1a35
--- /dev/null
+++ b/components/camel-elasticsearch/src/main/java/org/apache/camel/component/elasticsearch/converter/ElasticsearchActionRequestConverter.java
@@ -0,0 +1,296 @@
+/*
+ * 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.camel.component.elasticsearch.converter;
+
+import java.io.ByteArrayInputStream;
+import java.io.IOException;
+import java.io.InputStream;
+import java.io.Reader;
+import java.io.StringReader;
+import java.util.List;
+import java.util.Map;
+
+import co.elastic.clients.elasticsearch._types.WaitForActiveShards;
+import co.elastic.clients.elasticsearch._types.query_dsl.Query;
+import co.elastic.clients.elasticsearch.core.BulkRequest;
+import co.elastic.clients.elasticsearch.core.DeleteRequest;
+import co.elastic.clients.elasticsearch.core.GetRequest;
+import co.elastic.clients.elasticsearch.core.IndexRequest;
+import co.elastic.clients.elasticsearch.core.MgetRequest;
+import co.elastic.clients.elasticsearch.core.SearchRequest;
+import co.elastic.clients.elasticsearch.core.UpdateRequest;
+import co.elastic.clients.elasticsearch.core.bulk.BulkOperation;
+import co.elastic.clients.elasticsearch.core.bulk.CreateOperation;
+import co.elastic.clients.elasticsearch.core.bulk.DeleteOperation;
+import co.elastic.clients.elasticsearch.core.bulk.IndexOperation;
+import co.elastic.clients.elasticsearch.core.bulk.UpdateOperation;
+import co.elastic.clients.elasticsearch.indices.DeleteIndexRequest;
+import co.elastic.clients.json.JsonData;
+import com.fasterxml.jackson.databind.JsonNode;
+import com.fasterxml.jackson.databind.ObjectMapper;
+import org.apache.camel.Converter;
+import org.apache.camel.Exchange;
+import org.apache.camel.component.elasticsearch.ElasticsearchConstants;
+import org.apache.camel.util.ObjectHelper;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+@Converter(generateLoader = true)
+public final class ElasticsearchActionRequestConverter {
+ private static final Logger LOG = LoggerFactory.getLogger(ElasticsearchActionRequestConverter.class);
+
+ private static final String ES_QUERY_DSL_PREFIX = "query";
+
+ private ElasticsearchActionRequestConverter() {
+ }
+
+ // Index requests
+ private static IndexOperation.Builder<?> createIndexOperationBuilder(Object document, Exchange exchange)
+ throws IOException {
+ if (document instanceof IndexOperation.Builder) {
+ return (IndexOperation.Builder<?>) document;
+ }
+ IndexOperation.Builder<Object> builder = new IndexOperation.Builder<>();
+ if (document instanceof byte[]) {
+ builder.document(JsonData.from(new ByteArrayInputStream((byte[]) document)));
+ } else if (document instanceof InputStream) {
+ builder.document(JsonData.from((InputStream) document));
+ } else if (document instanceof String) {
+ builder.document(JsonData.from(new StringReader((String) document)));
+ } else if (document instanceof Reader) {
+ builder.document(JsonData.from((Reader) document));
+ } else if (document instanceof Map) {
+ ObjectMapper objectMapper = new ObjectMapper();
+ builder.document(JsonData.from(new StringReader(objectMapper.writeValueAsString(document))));
+ } else {
+ builder.document(document);
+ }
+ return builder
+ .index(exchange.getIn().getHeader(ElasticsearchConstants.PARAM_INDEX_NAME, String.class));
+ }
+
+ @Converter
+ public static IndexRequest.Builder<?> toIndexRequestBuilder(Object document, Exchange exchange) throws IOException {
+ if (document instanceof IndexRequest.Builder) {
+ IndexRequest.Builder<?> builder = (IndexRequest.Builder<?>) document;
+ return builder.id(exchange.getIn().getHeader(ElasticsearchConstants.PARAM_INDEX_ID, String.class));
+ }
+ IndexRequest.Builder<Object> builder = new IndexRequest.Builder<>();
+ if (document instanceof byte[]) {
+ builder.withJson(new ByteArrayInputStream((byte[]) document));
+ } else if (document instanceof InputStream) {
+ builder.withJson((InputStream) document);
+ } else if (document instanceof String) {
+ builder.withJson(new StringReader((String) document));
+ } else if (document instanceof Reader) {
+ builder.withJson((Reader) document);
+ } else if (document instanceof Map) {
+ ObjectMapper objectMapper = new ObjectMapper();
+ builder.withJson(new StringReader(objectMapper.writeValueAsString(document)));
+ } else {
+ builder.document(document);
+ }
+ return builder
+ .waitForActiveShards(
+ new WaitForActiveShards.Builder()
+ .count(exchange.getIn().getHeader(ElasticsearchConstants.PARAM_WAIT_FOR_ACTIVE_SHARDS,
+ Integer.class))
+ .build())
+ .id(exchange.getIn().getHeader(ElasticsearchConstants.PARAM_INDEX_ID, String.class))
+ .index(exchange.getIn().getHeader(ElasticsearchConstants.PARAM_INDEX_NAME, String.class));
+ }
+
+ @Converter
+ public static UpdateRequest.Builder<?, ?> toUpdateRequestBuilder(Object document, Exchange exchange) throws IOException {
+ if (document instanceof UpdateRequest.Builder) {
+ UpdateRequest.Builder<?, ?> builder = (UpdateRequest.Builder<?, ?>) document;
+ return builder.id(exchange.getIn().getHeader(ElasticsearchConstants.PARAM_INDEX_ID, String.class));
+ }
+ UpdateRequest.Builder<?, Object> builder = new UpdateRequest.Builder<>();
+ if (document instanceof byte[]) {
+ builder.withJson(new ByteArrayInputStream((byte[]) document));
+ } else if (document instanceof InputStream) {
+ builder.withJson((InputStream) document);
+ } else if (document instanceof String) {
+ builder.withJson(new StringReader((String) document));
+ } else if (document instanceof Reader) {
+ builder.withJson((Reader) document);
+ } else if (document instanceof Map) {
+ ObjectMapper objectMapper = new ObjectMapper();
+ builder.withJson(new StringReader(objectMapper.writeValueAsString(document)));
+ } else {
+ builder.doc(document);
+ }
+
+ return builder
+ .waitForActiveShards(
+ new WaitForActiveShards.Builder()
+ .count(exchange.getIn().getHeader(ElasticsearchConstants.PARAM_WAIT_FOR_ACTIVE_SHARDS,
+ Integer.class))
+ .build())
+ .index(exchange.getIn().getHeader(ElasticsearchConstants.PARAM_INDEX_NAME, String.class))
+ .id(exchange.getIn().getHeader(ElasticsearchConstants.PARAM_INDEX_ID, String.class));
+ }
+
+ @Converter
+ public static GetRequest.Builder toGetRequestBuilder(Object document, Exchange exchange) {
+ if (document instanceof GetRequest.Builder) {
+ return (GetRequest.Builder) document;
+ }
+ if (document instanceof String) {
+ return new GetRequest.Builder()
+ .index(exchange.getIn().getHeader(ElasticsearchConstants.PARAM_INDEX_NAME, String.class))
+ .id((String) document);
+ }
+ return null;
+ }
+
+ @Converter
+ public static DeleteRequest.Builder toDeleteRequestBuilder(Object document, Exchange exchange) {
+ if (document instanceof DeleteRequest.Builder) {
+ return (DeleteRequest.Builder) document;
+ }
+ if (document instanceof String) {
+ return new DeleteRequest.Builder()
+ .index(exchange.getIn().getHeader(ElasticsearchConstants.PARAM_INDEX_NAME, String.class))
+ .id((String) document);
+ }
+ return null;
+ }
+
+ @Converter
+ public static DeleteIndexRequest.Builder toDeleteIndexRequestBuilder(Object document, Exchange exchange) {
+ if (document instanceof DeleteIndexRequest.Builder) {
+ return (DeleteIndexRequest.Builder) document;
+ }
+ if (document instanceof String) {
+ return new DeleteIndexRequest.Builder()
+ .index(exchange.getIn().getHeader(ElasticsearchConstants.PARAM_INDEX_NAME, String.class));
+ }
+ return null;
+ }
+
+ @Converter
+ public static MgetRequest.Builder toMgetRequestBuilder(Object documents, Exchange exchange) {
+ if (documents instanceof MgetRequest.Builder) {
+ return (MgetRequest.Builder) documents;
+ }
+ if (documents instanceof Iterable) {
+ MgetRequest.Builder builder = new MgetRequest.Builder();
+ builder.index(exchange.getIn().getHeader(ElasticsearchConstants.PARAM_INDEX_NAME, String.class));
+ for (Object document : (List<?>) documents) {
+ if (document instanceof String) {
+ builder.ids((String) document);
+ } else {
+ LOG.warn(
+ "Cannot convert document id of type {} into a String",
+ document == null ? "null" : document.getClass().getName());
+ return null;
+ }
+ }
+ return builder;
+ }
+ return null;
+ }
+
+ @Converter
+ public static SearchRequest.Builder toSearchRequestBuilder(Object queryObject, Exchange exchange) throws IOException {
+ String indexName = exchange.getIn().getHeader(ElasticsearchConstants.PARAM_INDEX_NAME, String.class);
+
+ if (queryObject instanceof SearchRequest.Builder) {
+ SearchRequest.Builder builder = (SearchRequest.Builder) queryObject;
+ if (builder.build().index().isEmpty()) {
+ builder.index(indexName);
+ }
+ return builder;
+ }
+ SearchRequest.Builder builder = new SearchRequest.Builder();
+
+ // Only setup the indexName if the message header has the
+ // setting
+
+ Integer size = exchange.getIn().getHeader(ElasticsearchConstants.PARAM_SIZE, Integer.class);
+ Integer from = exchange.getIn().getHeader(ElasticsearchConstants.PARAM_FROM, Integer.class);
+ if (ObjectHelper.isNotEmpty(indexName)) {
+ builder.index(indexName);
+ }
+
+ String queryText;
+
+ if (queryObject instanceof Map<?, ?>) {
+ Map<?, ?> mapQuery = (Map<?, ?>) queryObject;
+ // Remove 'query' prefix from the query object for backward
+ // compatibility
+ if (mapQuery.containsKey(ES_QUERY_DSL_PREFIX)) {
+ mapQuery = (Map<?, ?>) mapQuery.get(ES_QUERY_DSL_PREFIX);
+ }
+ ObjectMapper objectMapper = new ObjectMapper();
+ queryText = objectMapper.writeValueAsString(mapQuery);
+ } else if (queryObject instanceof String) {
+ queryText = (String) queryObject;
+ ObjectMapper mapper = new ObjectMapper();
+ JsonNode jsonTextObject = mapper.readValue(queryText, JsonNode.class);
+ JsonNode parentJsonNode = jsonTextObject.get(ES_QUERY_DSL_PREFIX);
+ if (parentJsonNode != null) {
+ queryText = parentJsonNode.toString();
+ }
+ } else {
+ // Cannot convert the queryObject into SearchRequest
+ LOG.warn(
+ "Cannot convert queryObject of type {} into SearchRequest object",
+ queryObject == null ? "null" : queryObject.getClass().getName());
+ return null;
+ }
+ if (size != null) {
+ builder.size(size);
+ }
+ if (from != null) {
+ builder.from(from);
+ }
+ builder.query(new Query.Builder().withJson(new StringReader(queryText)).build());
+
+ return builder;
+ }
+
+ @Converter
+ public static BulkRequest.Builder toBulkRequestBuilder(Object documents, Exchange exchange) throws IOException {
+ if (documents instanceof BulkRequest.Builder) {
+ return (BulkRequest.Builder) documents;
+ }
+ if (documents instanceof Iterable) {
+ BulkRequest.Builder builder = new BulkRequest.Builder();
+ builder.index(exchange.getIn().getHeader(ElasticsearchConstants.PARAM_INDEX_NAME, String.class));
+ for (Object document : (List<?>) documents) {
+ if (document instanceof DeleteOperation.Builder) {
+ builder.operations(
+ new BulkOperation.Builder().delete(((DeleteOperation.Builder) document).build()).build());
+ } else if (document instanceof UpdateOperation.Builder) {
+ builder.operations(
+ new BulkOperation.Builder().update(((UpdateOperation.Builder<?, ?>) document).build()).build());
+ } else if (document instanceof CreateOperation.Builder) {
+ builder.operations(
+ new BulkOperation.Builder().create(((CreateOperation.Builder<?>) document).build()).build());
+ } else {
+ builder.operations(
+ new BulkOperation.Builder().index(createIndexOperationBuilder(document, exchange).build()).build());
+ }
+ }
+ return builder;
+ }
+ return null;
+ }
+}
diff --git a/components/camel-elasticsearch/src/test/java/org/apache/camel/component/elasticsearch/ElasticsearchRestComponentVerifierExtensionTest.java b/components/camel-elasticsearch/src/test/java/org/apache/camel/component/elasticsearch/ElasticsearchRestComponentVerifierExtensionTest.java
new file mode 100644
index 00000000000..28d407deda9
--- /dev/null
+++ b/components/camel-elasticsearch/src/test/java/org/apache/camel/component/elasticsearch/ElasticsearchRestComponentVerifierExtensionTest.java
@@ -0,0 +1,65 @@
+/*
+ * 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.camel.component.elasticsearch;
+
+import java.util.HashMap;
+import java.util.Map;
+
+import org.apache.camel.Component;
+import org.apache.camel.component.extension.ComponentVerifierExtension;
+import org.apache.camel.test.junit5.CamelTestSupport;
+import org.junit.jupiter.api.Test;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+
+class ElasticsearchRestComponentVerifierExtensionTest extends CamelTestSupport {
+ @Override
+ public boolean isUseRouteBuilder() {
+ return false;
+ }
+
+ @Test
+ void testParameters() {
+ Component component = context().getComponent("elasticsearch");
+
+ ComponentVerifierExtension verifier
+ = component.getExtension(ComponentVerifierExtension.class).orElseThrow(IllegalStateException::new);
+
+ Map<String, Object> parameters = new HashMap<>();
+ parameters.put("hostAddresses", "http://localhost:9000");
+ parameters.put("clusterName", "es-test");
+
+ ComponentVerifierExtension.Result result = verifier.verify(ComponentVerifierExtension.Scope.PARAMETERS, parameters);
+
+ assertEquals(ComponentVerifierExtension.Result.Status.OK, result.getStatus());
+ }
+
+ @Test
+ void testConnectivity() {
+ Component component = context().getComponent("elasticsearch");
+ ComponentVerifierExtension verifier
+ = component.getExtension(ComponentVerifierExtension.class).orElseThrow(IllegalStateException::new);
+
+ Map<String, Object> parameters = new HashMap<>();
+ parameters.put("hostAddresses", "http://localhost:9000");
+
+ ComponentVerifierExtension.Result result = verifier.verify(ComponentVerifierExtension.Scope.CONNECTIVITY, parameters);
+
+ assertEquals(ComponentVerifierExtension.Result.Status.ERROR, result.getStatus());
+ }
+
+}
diff --git a/components/camel-elasticsearch/src/test/java/org/apache/camel/component/elasticsearch/integration/ElasticsearchBulkIT.java b/components/camel-elasticsearch/src/test/java/org/apache/camel/component/elasticsearch/integration/ElasticsearchBulkIT.java
new file mode 100644
index 00000000000..507b07dd426
--- /dev/null
+++ b/components/camel-elasticsearch/src/test/java/org/apache/camel/component/elasticsearch/integration/ElasticsearchBulkIT.java
@@ -0,0 +1,255 @@
+/*
+ * 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.camel.component.elasticsearch.integration;
+
+import java.io.ByteArrayInputStream;
+import java.io.InputStream;
+import java.io.Reader;
+import java.io.StringReader;
+import java.nio.charset.StandardCharsets;
+import java.util.ArrayList;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+
+import co.elastic.clients.elasticsearch.core.BulkRequest;
+import co.elastic.clients.elasticsearch.core.GetResponse;
+import co.elastic.clients.elasticsearch.core.bulk.BulkOperation;
+import co.elastic.clients.elasticsearch.core.bulk.BulkResponseItem;
+import co.elastic.clients.elasticsearch.core.bulk.CreateOperation;
+import co.elastic.clients.elasticsearch.core.bulk.DeleteOperation;
+import co.elastic.clients.elasticsearch.core.bulk.IndexOperation;
+import co.elastic.clients.elasticsearch.core.bulk.UpdateAction;
+import co.elastic.clients.elasticsearch.core.bulk.UpdateOperation;
+import com.fasterxml.jackson.databind.node.ObjectNode;
+import org.apache.camel.builder.RouteBuilder;
+import org.junit.jupiter.api.Test;
+
+import static org.apache.camel.test.junit5.TestSupport.assertCollectionSize;
+import static org.hamcrest.CoreMatchers.equalTo;
+import static org.hamcrest.CoreMatchers.notNullValue;
+import static org.hamcrest.CoreMatchers.nullValue;
+import static org.hamcrest.MatcherAssert.assertThat;
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertInstanceOf;
+import static org.junit.jupiter.api.Assertions.assertNotNull;
+import static org.junit.jupiter.api.Assertions.assertNull;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
+class ElasticsearchBulkIT extends ElasticsearchTestSupport {
+
+ @Test
+ void testBulkWithMap() {
+ List<Map<String, String>> documents = new ArrayList<>();
+ Map<String, String> document1 = createIndexedData("1");
+ Map<String, String> document2 = createIndexedData("2");
+
+ documents.add(document1);
+ documents.add(document2);
+
+ List<?> indexIds = template.requestBody("direct:bulk", documents, List.class);
+ assertNotNull(indexIds, "indexIds should be set");
+ assertCollectionSize("Indexed documents should match the size of documents", indexIds, documents.size());
+ }
+
+ @Test
+ void testBulkWithString() {
+ List<String> documents = List.of(
+ "{\"testBulkWithString1\": \"some-value\"}", "{\"testBulkWithString2\": \"some-value\"}");
+
+ List<?> indexIds = template.requestBody("direct:bulk", documents, List.class);
+ assertNotNull(indexIds, "indexIds should be set");
+ assertCollectionSize("Indexed documents should match the size of documents", indexIds, documents.size());
+ }
+
+ @Test
+ void testBulkWithBytes() {
+ List<byte[]> documents = List.of(
+ "{\"testBulkWithBytes1\": \"some-value\"}".getBytes(StandardCharsets.UTF_8),
+ "{\"testBulkWithBytes2\": \"some-value\"}".getBytes(StandardCharsets.UTF_8));
+
+ List<?> indexIds = template.requestBody("direct:bulk", documents, List.class);
+ assertNotNull(indexIds, "indexIds should be set");
+ assertCollectionSize("Indexed documents should match the size of documents", indexIds, documents.size());
+ }
+
+ @Test
+ void testBulkWithReader() {
+ List<Reader> documents = List.of(
+ new StringReader("{\"testBulkWithReader1\": \"some-value\"}"),
+ new StringReader("{\"testBulkWithReader2\": \"some-value\"}"));
+
+ List<?> indexIds = template.requestBody("direct:bulk", documents, List.class);
+ assertNotNull(indexIds, "indexIds should be set");
+ assertCollectionSize("Indexed documents should match the size of documents", indexIds, documents.size());
+ }
+
+ @Test
+ void testBulkWithInputStream() {
+ List<InputStream> documents = List.of(
+ new ByteArrayInputStream(
+ "{\"testBulkWithInputStream1\": \"some-value\"}".getBytes(StandardCharsets.UTF_8)),
+ new ByteArrayInputStream(
+ "{\"testBulkWithInputStream2\": \"some-value\"}".getBytes(StandardCharsets.UTF_8)));
+
+ List<?> indexIds = template.requestBody("direct:bulk", documents, List.class);
+ assertNotNull(indexIds, "indexIds should be set");
+ assertCollectionSize("Indexed documents should match the size of documents", indexIds, documents.size());
+ }
+
+ @Test
+ void testBulkListRequestBody() {
+ String prefix = createPrefix();
+
+ // given
+ List<Map<String, String>> request = new ArrayList<>();
+ final HashMap<String, String> valueMap = new HashMap<>();
+ valueMap.put("id", prefix + "baz");
+ valueMap.put("content", prefix + "hello");
+ request.add(valueMap);
+ // when
+ List<?> indexedDocumentIds = template.requestBody("direct:bulk", request, List.class);
+
+ // then
+ assertThat(indexedDocumentIds, notNullValue());
+ assertThat(indexedDocumentIds.size(), equalTo(1));
+ }
+
+ @Test
+ void testBulkRequestBody() {
+ String prefix = createPrefix();
+
+ // given
+ BulkRequest.Builder builder = new BulkRequest.Builder();
+ builder.operations(
+ new BulkOperation.Builder()
+ .index(new IndexOperation.Builder<>().index(prefix + "foo").id(prefix + "baz")
+ .document(Map.of(prefix + "content", prefix + "hello")).build())
+ .build());
+
+ // when
+ @SuppressWarnings("unchecked")
+ List<BulkResponseItem> response = template.requestBody("direct:bulk", builder, List.class);
+
+ // then
+ assertThat(response, notNullValue());
+ assertThat(response.size(), equalTo(1));
+ assertThat(response.get(0).error(), nullValue());
+ assertThat(response.get(0).id(), equalTo(prefix + "baz"));
+ }
+
+ @Test
+ void bulkRequestBody() {
+ String prefix = createPrefix();
+
+ // given
+ BulkRequest.Builder builder = new BulkRequest.Builder();
+ builder.operations(
+ new BulkOperation.Builder()
+ .index(new IndexOperation.Builder<>().index(prefix + "foo").id(prefix + "baz")
+ .document(Map.of(prefix + "content", prefix + "hello")).build())
+ .build());
+ // when
+ @SuppressWarnings("unchecked")
+ List<BulkResponseItem> response = template.requestBody("direct:bulk", builder, List.class);
+
+ // then
+ assertThat(response, notNullValue());
+ assertEquals(prefix + "baz", response.get(0).id());
+ }
+
+ @Test
+ void bulkDeleteOperation() {
+ // given
+ Map<String, String> map = createIndexedData();
+ String indexId = template.requestBody("direct:index", map, String.class);
+ assertNotNull(indexId, "indexId should be set");
+
+ DeleteOperation.Builder builder = new DeleteOperation.Builder().index("twitter").id(indexId);
+ // when
+ @SuppressWarnings("unchecked")
+ List<BulkResponseItem> response = template.requestBody("direct:bulk", List.of(builder), List.class);
+
+ // then
+ assertThat(response, notNullValue());
+ assertEquals(indexId, response.get(0).id());
+ GetResponse<?> resp = template.requestBody("direct:get", indexId, GetResponse.class);
+ assertNotNull(resp, "response should not be null");
+ assertNull(resp.source(), "response source should be null");
+ }
+
+ @Test
+ void bulkCreateOperation() {
+ // given
+ String prefix = createPrefix();
+
+ CreateOperation.Builder<?> builder
+ = new CreateOperation.Builder<>().index("twitter").document(Map.of(prefix + "content", prefix + "hello"));
+ // when
+ @SuppressWarnings("unchecked")
+ List<BulkResponseItem> response = template.requestBody("direct:bulk", List.of(builder), List.class);
+
+ // then
+ assertThat(response, notNullValue());
+ GetResponse<?> resp = template.requestBody("direct:get", response.get(0).id(), GetResponse.class);
+ assertNotNull(resp, "response should not be null");
+ assertNotNull(resp.source(), "response source should not be null");
+ }
+
+ @Test
+ void bulkUpdateOperation() {
+ Map<String, String> map = createIndexedData();
+ String indexId = template.requestBody("direct:index", map, String.class);
+ assertNotNull(indexId, "indexId should be set");
+
+ UpdateOperation.Builder<?, ?> builder = new UpdateOperation.Builder<>()
+ .index("twitter").id(indexId)
+ .action(
+ new UpdateAction.Builder<>()
+ .withJson(
+ new StringReader(
+ String.format("{ \"doc\": {\"%skey2\": \"%svalue2\"}}",
+ createPrefix(), createPrefix())))
+ .build());
+ @SuppressWarnings("unchecked")
+ List<BulkResponseItem> response = template.requestBody("direct:bulk", List.of(builder), List.class);
+
+ //now, verify GET succeeded
+ assertThat(response, notNullValue());
+ GetResponse<?> resp = template.requestBody("direct:get", indexId, GetResponse.class);
+ assertNotNull(resp, "response should not be null");
+ assertNotNull(resp.source(), "response source should not be null");
+ assertInstanceOf(ObjectNode.class, resp.source(), "response source should be a ObjectNode");
+ assertTrue(((ObjectNode) resp.source()).has(createPrefix() + "key2"));
+ assertEquals(createPrefix() + "value2", ((ObjectNode) resp.source()).get(createPrefix() + "key2").asText());
+ }
+
+ @Override
+ protected RouteBuilder createRouteBuilder() {
+ return new RouteBuilder() {
+ @Override
+ public void configure() {
+ from("direct:index")
+ .to("elasticsearch://elasticsearch?operation=Index&indexName=twitter");
+ from("direct:get")
+ .to("elasticsearch://elasticsearch?operation=GetById&indexName=twitter");
+ from("direct:bulk")
+ .to("elasticsearch://elasticsearch?operation=Bulk&indexName=twitter");
+ }
+ };
+ }
+}
diff --git a/components/camel-elasticsearch/src/test/java/org/apache/camel/component/elasticsearch/integration/ElasticsearchClusterIndexIT.java b/components/camel-elasticsearch/src/test/java/org/apache/camel/component/elasticsearch/integration/ElasticsearchClusterIndexIT.java
new file mode 100644
index 00000000000..cfd326e5b2b
--- /dev/null
+++ b/components/camel-elasticsearch/src/test/java/org/apache/camel/component/elasticsearch/integration/ElasticsearchClusterIndexIT.java
@@ -0,0 +1,86 @@
+/*
+ * 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.camel.component.elasticsearch.integration;
+
+import java.util.HashMap;
+import java.util.Map;
+
+import co.elastic.clients.elasticsearch.core.GetRequest;
+import com.fasterxml.jackson.databind.node.ObjectNode;
+import org.apache.camel.builder.RouteBuilder;
+import org.apache.camel.component.elasticsearch.ElasticsearchConstants;
+import org.apache.camel.component.elasticsearch.ElasticsearchOperation;
+import org.apache.http.impl.client.BasicResponseHandler;
+import org.elasticsearch.client.Request;
+import org.junit.jupiter.api.Test;
+
+import static org.apache.camel.test.junit5.TestSupport.assertStringContains;
+import static org.junit.jupiter.api.Assertions.assertNotNull;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
+class ElasticsearchClusterIndexIT extends ElasticsearchTestSupport {
+ @Test
+ void indexWithIpAndPort() throws Exception {
+ Map<String, String> map = createIndexedData();
+ Map<String, Object> headers = new HashMap<>();
+ headers.put(ElasticsearchConstants.PARAM_OPERATION, ElasticsearchOperation.Index);
+ headers.put(ElasticsearchConstants.PARAM_INDEX_NAME, "twitter");
+ headers.put(ElasticsearchConstants.PARAM_INDEX_ID, "1");
+
+ String indexId = template.requestBodyAndHeaders("direct:indexWithIpAndPort", map, headers, String.class);
+ assertNotNull(indexId, "indexId should be set");
+
+ indexId = template.requestBodyAndHeaders("direct:indexWithIpAndPort", map, headers, String.class);
+ assertNotNull(indexId, "indexId should be set");
+
+ assertTrue(client.get(new GetRequest.Builder().index("twitter").id("1").build(), ObjectNode.class).found(),
+ "Index id 1 must exists");
+ }
+
+ @Test
+ void indexWithSnifferEnable() throws Exception {
+ Map<String, String> map = createIndexedData();
+ Map<String, Object> headers = new HashMap<>();
+ headers.put(ElasticsearchConstants.PARAM_OPERATION, ElasticsearchOperation.Index);
+ headers.put(ElasticsearchConstants.PARAM_INDEX_NAME, "facebook");
+ headers.put(ElasticsearchConstants.PARAM_INDEX_ID, "4");
+
+ String indexId = template.requestBodyAndHeaders("direct:indexWithSniffer", map, headers, String.class);
+ assertNotNull(indexId, "indexId should be set");
+
+ assertTrue(client.get(new GetRequest.Builder().index("facebook").id("4").build(), ObjectNode.class).found(),
+ "Index id 4 must exists");
+
+ final BasicResponseHandler responseHandler = new BasicResponseHandler();
+ Request request = new Request("GET", "/_cluster/health?pretty");
+ String body = responseHandler.handleEntity(restClient.performRequest(request).getEntity());
+ assertStringContains(body, "\"number_of_data_nodes\" : 1");
+ }
+
+ @Override
+ protected RouteBuilder createRouteBuilder() {
+ return new RouteBuilder() {
+ @Override
+ public void configure() {
+ from("direct:indexWithIpAndPort")
+ .to("elasticsearch://" + clusterName + "?operation=Index&indexName=twitter");
+ from("direct:indexWithSniffer")
+ .to("elasticsearch://" + clusterName + "?operation=Index&indexName=twitter&enableSniffer=true");
+ }
+ };
+ }
+}
diff --git a/components/camel-elasticsearch/src/test/java/org/apache/camel/component/elasticsearch/integration/ElasticsearchGetSearchDeleteExistsUpdateIT.java b/components/camel-elasticsearch/src/test/java/org/apache/camel/component/elasticsearch/integration/ElasticsearchGetSearchDeleteExistsUpdateIT.java
new file mode 100644
index 00000000000..16c32156893
--- /dev/null
+++ b/components/camel-elasticsearch/src/test/java/org/apache/camel/component/elasticsearch/integration/ElasticsearchGetSearchDeleteExistsUpdateIT.java
@@ -0,0 +1,901 @@
+/*
+ * 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.camel.component.elasticsearch.integration;
+
+import java.io.ByteArrayInputStream;
+import java.io.StringReader;
+import java.nio.charset.StandardCharsets;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.Objects;
+import java.util.concurrent.TimeUnit;
+
+import co.elastic.clients.elasticsearch._types.Result;
+import co.elastic.clients.elasticsearch._types.query_dsl.MatchQuery;
+import co.elastic.clients.elasticsearch._types.query_dsl.Query;
+import co.elastic.clients.elasticsearch.core.DeleteRequest;
+import co.elastic.clients.elasticsearch.core.GetRequest;
+import co.elastic.clients.elasticsearch.core.GetResponse;
+import co.elastic.clients.elasticsearch.core.IndexRequest;
+import co.elastic.clients.elasticsearch.core.MsearchRequest;
+import co.elastic.clients.elasticsearch.core.SearchRequest;
+import co.elastic.clients.elasticsearch.core.mget.MultiGetResponseItem;
+import co.elastic.clients.elasticsearch.core.msearch.MultiSearchResponseItem;
+import co.elastic.clients.elasticsearch.core.msearch.MultisearchBody;
+import co.elastic.clients.elasticsearch.core.msearch.MultisearchHeader;
+import co.elastic.clients.elasticsearch.core.msearch.RequestItem;
+import co.elastic.clients.elasticsearch.core.search.HitsMetadata;
+import com.fasterxml.jackson.databind.JsonNode;
+import com.fasterxml.jackson.databind.node.ObjectNode;
+import org.apache.camel.builder.RouteBuilder;
+import org.apache.camel.component.elasticsearch.ElasticsearchConstants;
+import org.apache.camel.component.elasticsearch.ElasticsearchOperation;
+import org.awaitility.Awaitility;
+import org.junit.jupiter.api.Test;
+
+import static org.hamcrest.CoreMatchers.equalTo;
+import static org.hamcrest.CoreMatchers.notNullValue;
+import static org.hamcrest.MatcherAssert.assertThat;
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertFalse;
+import static org.junit.jupiter.api.Assertions.assertInstanceOf;
+import static org.junit.jupiter.api.Assertions.assertNotNull;
+import static org.junit.jupiter.api.Assertions.assertNotSame;
+import static org.junit.jupiter.api.Assertions.assertNull;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
+class ElasticsearchGetSearchDeleteExistsUpdateIT extends ElasticsearchTestSupport {
+
+ @Test
+ void testIndexWithMap() {
+ //first, Index a value
+ Map<String, String> map = createIndexedData();
+ String indexId = template.requestBody("direct:index", map, String.class);
+ assertNotNull(indexId, "indexId should be set");
+
+ //now, verify GET succeeded
+ GetResponse<?> response = template.requestBody("direct:get", indexId, GetResponse.class);
+ assertNotNull(response, "response should not be null");
+ assertNotNull(response.source(), "response source should not be null");
+ assertInstanceOf(ObjectNode.class, response.source(), "response source should be a ObjectNode");
+ String key = map.keySet().iterator().next();
+ assertTrue(((ObjectNode) response.source()).has(key));
+ assertEquals(map.get(key), ((ObjectNode) response.source()).get(key).asText());
+ }
+
+ @Test
+ void testIndexWithString() {
+ //first, Index a value
+ String indexId = template.requestBody("direct:index", "{\"testIndexWithString\": \"some-value\"}", String.class);
+ assertNotNull(indexId, "indexId should be set");
+
+ //now, verify GET succeeded
+ GetResponse<?> response = template.requestBody("direct:get", indexId, GetResponse.class);
+ assertNotNull(response, "response should not be null");
+ assertNotNull(response.source(), "response source should not be null");
+ assertInstanceOf(ObjectNode.class, response.source(), "response source should be a ObjectNode");
+ assertTrue(((ObjectNode) response.source()).has("testIndexWithString"));
+ assertEquals("some-value", ((ObjectNode) response.source()).get("testIndexWithString").asText());
+ }
+
+ @Test
+ void testIndexWithReader() {
+ //first, Index a value
+ String indexId = template.requestBody("direct:index", new StringReader("{\"testIndexWithReader\": \"some-value\"}"),
+ String.class);
+ assertNotNull(indexId, "indexId should be set");
+
+ //now, verify GET succeeded
+ GetResponse<?> response = template.requestBody("direct:get", indexId, GetResponse.class);
+ assertNotNull(response, "response should not be null");
+ assertNotNull(response.source(), "response source should not be null");
+ assertInstanceOf(ObjectNode.class, response.source(), "response source should be a ObjectNode");
+ assertTrue(((ObjectNode) response.source()).has("testIndexWithReader"));
+ assertEquals("some-value", ((ObjectNode) response.source()).get("testIndexWithReader").asText());
+ }
+
+ @Test
+ void testIndexWithBytes() {
+ //first, Index a value
+ String indexId = template.requestBody("direct:index",
+ "{\"testIndexWithBytes\": \"some-value\"}".getBytes(StandardCharsets.UTF_8), String.class);
+ assertNotNull(indexId, "indexId should be set");
+
+ //now, verify GET succeeded
+ GetResponse<?> response = template.requestBody("direct:get", indexId, GetResponse.class);
+ assertNotNull(response, "response should not be null");
+ assertNotNull(response.source(), "response source should not be null");
+ assertInstanceOf(ObjectNode.class, response.source(), "response source should be a ObjectNode");
+ assertTrue(((ObjectNode) response.source()).has("testIndexWithBytes"));
+ assertEquals("some-value", ((ObjectNode) response.source()).get("testIndexWithBytes").asText());
+ }
+
+ @Test
+ void testIndexWithInputStream() {
+ //first, Index a value
+ String indexId = template.requestBody("direct:index",
+ new ByteArrayInputStream("{\"testIndexWithInputStream\": \"some-value\"}".getBytes(StandardCharsets.UTF_8)),
+ String.class);
+ assertNotNull(indexId, "indexId should be set");
+
+ //now, verify GET succeeded
+ GetResponse<?> response = template.requestBody("direct:get", indexId, GetResponse.class);
+ assertNotNull(response, "response should not be null");
+ assertNotNull(response.source(), "response source should not be null");
+ assertInstanceOf(ObjectNode.class, response.source(), "response source should be a ObjectNode");
+ assertTrue(((ObjectNode) response.source()).has("testIndexWithInputStream"));
+ assertEquals("some-value", ((ObjectNode) response.source()).get("testIndexWithInputStream").asText());
+ }
+
+ @Test
+ void testIndexWithDocumentType() {
+ Product product = new Product();
+ product.setId("book-world-records-2021");
+ product.setStockAvailable(1);
+ product.setPrice(100);
+ product.setDescription("The book of the year!");
+ product.setName("Guinness book of records 2021");
+
+ //first, Index a value
+ String indexId = template.requestBody("direct:index-product", product, String.class);
+ assertNotNull(indexId, "indexId should be set");
+
+ //now, verify GET succeeded
+ GetResponse<?> response = template.requestBodyAndHeader("direct:get", indexId,
+ ElasticsearchConstants.PARAM_DOCUMENT_CLASS, Product.class, GetResponse.class);
+ assertNotNull(response, "response should not be null");
+ assertNotNull(response.source(), "response source should not be null");
+ assertInstanceOf(Product.class, response.source(), "response source should be a Product");
+ Product actual = (Product) response.source();
+ assertNotSame(product, actual);
+ assertEquals(product, actual);
+ }
+
+ @Test
+ void testGetWithString() {
+ //first, Index a value
+ Map<String, String> map = createIndexedData();
+ String indexId = template.requestBody("direct:index", map, String.class);
+ assertNotNull(indexId, "indexId should be set");
+
+ //now, verify GET succeeded
+ GetResponse<?> response = template.requestBody("direct:get", indexId, GetResponse.class);
+ assertNotNull(response, "response should not be null");
+ assertNotNull(response.source(), "response source should not be null");
+ assertInstanceOf(ObjectNode.class, response.source());
+ }
+
+ @Test
+ void testGetWithDocumentType() {
+ //first, Index a value
+ Product product = new Product();
+ product.setId("book-world-records-1890");
+ product.setStockAvailable(0);
+ product.setPrice(200);
+ product.setDescription("The book of the year!");
+ product.setName("Guinness book of records 1890");
+
+ String indexId = template.requestBody("direct:index", product, String.class);
+ assertNotNull(indexId, "indexId should be set");
+
+ //now, verify GET succeeded
+ GetResponse<?> response = template.requestBodyAndHeader(
+ "direct:get", indexId, ElasticsearchConstants.PARAM_DOCUMENT_CLASS, Product.class, GetResponse.class);
+ assertNotNull(response, "response should not be null");
+ assertNotNull(response.source(), "response source should not be null");
+ assertInstanceOf(Product.class, response.source());
+ Product p = (Product) response.source();
+ assertEquals(product, p);
+ }
+
+ @Test
+ void testMGetWithString() {
+ //first, Index a value
+ Map<String, String> map = createIndexedData();
+ String indexId = template.requestBody("direct:index", map, String.class);
+ assertNotNull(indexId, "indexId should be set");
+
+ //now, verify GET succeeded
+ @SuppressWarnings("unchecked")
+ List<MultiGetResponseItem<?>> response = template.requestBody("direct:multiget", List.of(indexId), List.class);
+ assertNotNull(response, "response should not be null");
+ assertEquals(1, response.size(), "response should contain one result");
+ assertTrue(response.get(0).isResult());
+ assertNotNull(response.get(0).result().source(), "response source should not be null");
+ assertInstanceOf(ObjectNode.class, response.get(0).result().source());
+ }
+
+ @Test
+ void testMGetWithDocumentType() {
+ //first, Index a value
+ Product product = new Product();
+ product.setId("book-world-records-1890");
+ product.setStockAvailable(0);
+ product.setPrice(200);
+ product.setDescription("The book of the year!");
+ product.setName("Guinness book of records 1890");
+
+ String indexId = template.requestBody("direct:index", product, String.class);
+ assertNotNull(indexId, "indexId should be set");
+
+ //now, verify GET succeeded
+ @SuppressWarnings("unchecked")
+ List<MultiGetResponseItem<?>> response = template.requestBodyAndHeader(
+ "direct:multiget", List.of(indexId), ElasticsearchConstants.PARAM_DOCUMENT_CLASS, Product.class, List.class);
+ assertNotNull(response, "response should not be null");
+ assertEquals(1, response.size(), "response should contain one result");
+ assertTrue(response.get(0).isResult());
+ assertNotNull(response.get(0).result().source(), "response source should not be null");
+ assertInstanceOf(Product.class, response.get(0).result().source());
+ Product p = (Product) response.get(0).result().source();
+ assertEquals(product, p);
+ }
+
+ @Test
+ void testDeleteWithString() {
+ //first, Index a value
+ Map<String, String> map = createIndexedData();
+ String indexId = template.requestBody("direct:index", map, String.class);
+ assertNotNull(indexId, "indexId should be set");
+
+ //now, verify GET succeeded
+ GetResponse<?> response = template.requestBody("direct:get", indexId, GetResponse.class);
+ assertNotNull(response, "response should not be null");
+ assertNotNull(response.source(), "response source should not be null");
+
+ //now, perform Delete
+ Result deleteResponse = template.requestBody("direct:delete", indexId, Result.class);
+ assertNotNull(deleteResponse, "response should not be null");
+
+ //now, verify GET fails to find the indexed value
+ response = template.requestBody("direct:get", indexId, GetResponse.class);
+ assertNotNull(response, "response should not be null");
+ assertNull(response.source(), "response source should be null");
+ }
+
+ @Test
+ void testSearchWithMapQuery() {
+ //first, Index a value
+ Map<String, String> map1 = Map.of("testSearchWithMapQuery1", "foo");
+ Map<String, String> map2 = Map.of("testSearchWithMapQuery2", "bar");
+ Map<String, Object> headers = new HashMap<>();
+ headers.put(ElasticsearchConstants.PARAM_OPERATION, ElasticsearchOperation.Bulk);
+ headers.put(ElasticsearchConstants.PARAM_INDEX_NAME, "twitter");
+ template.requestBodyAndHeaders("direct:start", List.of(Map.of("doc", map1), Map.of("doc", map2)), headers,
+ String.class);
+
+ // No match
+ Map<String, Object> actualQuery = new HashMap<>();
+ actualQuery.put("doc.testSearchWithMapQuery1", "bar");
+ Map<String, Object> match = new HashMap<>();
+ match.put("match", actualQuery);
+ Map<String, Object> query = new HashMap<>();
+ query.put("query", match);
+ HitsMetadata<?> response = template.requestBody("direct:search", query, HitsMetadata.class);
+ assertNotNull(response, "response should not be null");
+ assertNotNull(response.total());
+ assertEquals(0, response.total().value(), "response hits should be == 0");
+
+ // Match
+ actualQuery.put("doc.testSearchWithMapQuery1", "foo");
+ // the result may see stale data so use Awaitility
+ Awaitility.await().atMost(10, TimeUnit.SECONDS).untilAsserted(() -> {
+ HitsMetadata<?> resp = template.requestBody("direct:search", query, HitsMetadata.class);
+ assertNotNull(resp, "response should not be null");
+ assertNotNull(resp.total());
+ assertEquals(1, resp.total().value(), "response hits should be == 1");
+ assertEquals(1, resp.hits().size(), "response hits should be == 1");
+ Object result = resp.hits().get(0).source();
+ assertInstanceOf(ObjectNode.class, result);
+ assertTrue(((ObjectNode) result).has("doc"));
+ JsonNode node = ((ObjectNode) result).get("doc");
+ assertTrue(node.has("testSearchWithMapQuery1"));
+ assertEquals("foo", node.get("testSearchWithMapQuery1").asText());
+ });
+ }
+
+ @Test
+ void testSearchWithStringQuery() {
+ //first, Index a value
+ Map<String, String> map1 = Map.of("testSearchWithStringQuery1", "foo");
+ Map<String, String> map2 = Map.of("testSearchWithStringQuery2", "bar");
+ Map<String, Object> headers = new HashMap<>();
+ headers.put(ElasticsearchConstants.PARAM_OPERATION, ElasticsearchOperation.Bulk);
+ headers.put(ElasticsearchConstants.PARAM_INDEX_NAME, "twitter");
+ template.requestBodyAndHeaders("direct:start", List.of(Map.of("doc", map1), Map.of("doc", map2)), headers,
+ String.class);
+
+ // No match
+ String query = "{\n"
+ + " \"query\" : { \"match\" : { \"doc.testSearchWithStringQuery1\" : \"bar\" }}\n"
+ + "}\n";
+
+ HitsMetadata<?> response = template.requestBody("direct:search", query, HitsMetadata.class);
+ assertNotNull(response, "response should not be null");
+ assertNotNull(response.total());
+ assertEquals(0, response.total().value(), "response hits should be == 0");
+
+ // Match
+ String q = "{\n"
+ + " \"query\" : { \"match\" : { \"doc.testSearchWithStringQuery1\" : \"foo\" }}\n"
+ + "}\n";
+ // the result may see stale data so use Awaitility
+ Awaitility.await().atMost(10, TimeUnit.SECONDS).untilAsserted(() -> {
+ HitsMetadata<?> resp = template.requestBody("direct:search", q, HitsMetadata.class);
+ assertNotNull(resp, "response should not be null");
+ assertNotNull(resp.total());
+ assertEquals(1, resp.total().value(), "response hits should be == 1");
+ assertEquals(1, resp.hits().size(), "response hits should be == 1");
+ Object result = resp.hits().get(0).source();
+ assertInstanceOf(ObjectNode.class, result);
+ assertTrue(((ObjectNode) result).has("doc"));
+ JsonNode node = ((ObjectNode) result).get("doc");
+ assertTrue(node.has("testSearchWithStringQuery1"));
+ assertEquals("foo", node.get("testSearchWithStringQuery1").asText());
+ });
+ }
+
+ @Test
+ void testSearchWithBuilder() {
+ //first, Index a value
+ Map<String, String> map1 = Map.of("testSearchWithBuilder1", "foo");
+ Map<String, String> map2 = Map.of("testSearchWithBuilder2", "bar");
+ Map<String, Object> headers = new HashMap<>();
+ headers.put(ElasticsearchConstants.PARAM_OPERATION, ElasticsearchOperation.Bulk);
+ headers.put(ElasticsearchConstants.PARAM_INDEX_NAME, "twitter");
+ template.requestBodyAndHeaders("direct:start", List.of(Map.of("doc", map1), Map.of("doc", map2)), headers,
+ String.class);
+
+ // No match
+ SearchRequest.Builder builder = new SearchRequest.Builder()
+ .query(new Query.Builder()
+ .match(new MatchQuery.Builder().field("doc.testSearchWithBuilder1").query("bar").build()).build());
+ HitsMetadata<?> response = template.requestBody("direct:search", builder, HitsMetadata.class);
+ assertNotNull(response, "response should not be null");
+ assertNotNull(response.total());
+ assertEquals(0, response.total().value(), "response hits should be == 0");
+
+ // Match
+ // the result may see stale data so use Awaitility
+ Awaitility.await().atMost(10, TimeUnit.SECONDS).untilAsserted(() -> {
+ SearchRequest.Builder b = new SearchRequest.Builder()
+ .query(new Query.Builder()
+ .match(new MatchQuery.Builder().field("doc.testSearchWithBuilder1").query("foo").build()).build());
+
+ HitsMetadata<?> resp = template.requestBody("direct:search", b, HitsMetadata.class);
+ assertNotNull(resp, "response should not be null");
+ assertNotNull(resp.total());
+ assertEquals(1, resp.total().value(), "response hits should be == 1");
+ assertEquals(1, resp.hits().size(), "response hits should be == 1");
+ Object result = resp.hits().get(0).source();
+ assertInstanceOf(ObjectNode.class, result);
+ assertTrue(((ObjectNode) result).has("doc"));
+ JsonNode node = ((ObjectNode) result).get("doc");
+ assertTrue(node.has("testSearchWithBuilder1"));
+ assertEquals("foo", node.get("testSearchWithBuilder1").asText());
+ });
+ }
+
+ @Test
+ void testSearchWithDocumentType() {
+ //first, Index a value
+ Product product1 = new Product();
+ product1.setId("book-world-records-2020");
+ product1.setStockAvailable(1);
+ product1.setPrice(100);
+ product1.setDescription("The book of the year!");
+ product1.setName("Guinness book of records 2020");
+
+ Product product2 = new Product();
+ product2.setId("book-world-records-2010");
+ product2.setStockAvailable(200);
+ product2.setPrice(80);
+ product2.setDescription("The book of the year!");
+ product2.setName("Guinness book of records 2010");
+ Map<String, Object> headers = new HashMap<>();
+ headers.put(ElasticsearchConstants.PARAM_OPERATION, ElasticsearchOperation.Bulk);
+ headers.put(ElasticsearchConstants.PARAM_INDEX_NAME, "twitter");
+ template.requestBodyAndHeaders("direct:start", List.of(product1, product2), headers, String.class);
+
+ // No match
+ SearchRequest.Builder builder = new SearchRequest.Builder()
+ .query(new Query.Builder().match(new MatchQuery.Builder().field("doc.id").query("bar").build()).build());
+ HitsMetadata<?> response = template.requestBodyAndHeader(
+ "direct:search", builder, ElasticsearchConstants.PARAM_DOCUMENT_CLASS, Product.class, HitsMetadata.class);
+ assertNotNull(response, "response should not be null");
+ assertNotNull(response.total());
+ assertEquals(0, response.total().value(), "response hits should be == 0");
+
+ // Match
+ // the result may see stale data so use Awaitility
+ Awaitility.await().atMost(10, TimeUnit.SECONDS).untilAsserted(() -> {
+ SearchRequest.Builder b = new SearchRequest.Builder()
+ .query(new Query.Builder().match(new MatchQuery.Builder().field("id").query("2020").build()).build());
+
+ HitsMetadata<?> resp = template.requestBodyAndHeader(
+ "direct:search", b, ElasticsearchConstants.PARAM_DOCUMENT_CLASS, Product.class, HitsMetadata.class);
+ assertNotNull(resp, "response should not be null");
+ assertNotNull(resp.total());
+ assertEquals(1, resp.total().value(), "response hits should be == 1");
+ assertEquals(1, resp.hits().size(), "response hits should be == 1");
+ Object result = resp.hits().get(0).source();
+ assertInstanceOf(Product.class, result);
+ Product p = (Product) result;
+ assertEquals(product1, p);
+ });
+ }
+
+ @Test
+ void testMultiSearch() {
+ //first, Index a value
+ Map<String, String> map = createIndexedData();
+ String indexId = template.requestBody("direct:index", map, String.class);
+ assertNotNull(indexId, "indexId should be set");
+
+ // the result may see stale data so use Awaitility
+ Awaitility.await().atMost(10, TimeUnit.SECONDS).untilAsserted(() -> {
+ //now, verify GET succeeded
+ MsearchRequest.Builder builder = new MsearchRequest.Builder().index("twitter").searches(
+ new RequestItem.Builder().header(new MultisearchHeader.Builder().build())
+ .body(new MultisearchBody.Builder().query(b -> b.matchAll(x -> x)).build()).build(),
+ new RequestItem.Builder().header(new MultisearchHeader.Builder().build())
+ .body(new MultisearchBody.Builder().query(b -> b.matchAll(x -> x)).build()).build());
+ @SuppressWarnings("unchecked")
+ List<MultiSearchResponseItem<?>> response = template.requestBody("direct:multiSearch", builder, List.class);
+ assertNotNull(response, "response should not be null");
+ assertEquals(2, response.size(), "response should be == 2");
+ assertInstanceOf(MultiSearchResponseItem.class, response.get(0));
+ assertTrue(response.get(0).isResult());
+ assertNotNull(response.get(0).result());
+ assertTrue(response.get(0).result().hits().total().value() > 0);
+ assertInstanceOf(MultiSearchResponseItem.class, response.get(1));
+ assertTrue(response.get(1).isResult());
+ assertNotNull(response.get(1).result());
+ assertTrue(response.get(1).result().hits().total().value() > 0);
+ });
+ }
+
+ @Test
+ void testMultiSearchWithDocumentType() {
+ //first, Index a value
+ Product product = new Product();
+ product.setId("book-world-records-2022");
+ product.setStockAvailable(1);
+ product.setPrice(100);
+ product.setDescription("The book of the year!");
+ product.setName("Guinness book of records 2022");
+ String indexId = template.requestBodyAndHeader("direct:index", product, ElasticsearchConstants.PARAM_INDEX_NAME,
+ "multi-search", String.class);
+ assertNotNull(indexId, "indexId should be set");
+
+ // the result may see stale data so use Awaitility
+ Awaitility.await().atMost(10, TimeUnit.SECONDS).untilAsserted(() -> {
+ //now, verify GET succeeded
+ MsearchRequest.Builder builder = new MsearchRequest.Builder().index("multi-search").searches(
+ new RequestItem.Builder().header(new MultisearchHeader.Builder().build())
+ .body(new MultisearchBody.Builder().query(b -> b.matchAll(x -> x)).build()).build(),
+ new RequestItem.Builder().header(new MultisearchHeader.Builder().build())
+ .body(new MultisearchBody.Builder().query(b -> b.matchAll(x -> x)).build()).build());
+ @SuppressWarnings("unchecked")
+ List<MultiSearchResponseItem<?>> response = template.requestBodyAndHeaders(
+ "direct:multiSearch", builder,
+ Map.of(
+ ElasticsearchConstants.PARAM_INDEX_NAME, "multi-search",
+ ElasticsearchConstants.PARAM_DOCUMENT_CLASS, Product.class),
+ List.class);
+ assertNotNull(response, "response should not be null");
+ assertEquals(2, response.size(), "response should be == 2");
+ assertInstanceOf(MultiSearchResponseItem.class, response.get(0));
+ assertTrue(response.get(0).isResult());
+ assertNotNull(response.get(0).result());
+ assertTrue(response.get(0).result().hits().total().value() > 0);
+ assertInstanceOf(MultiSearchResponseItem.class, response.get(1));
+ assertTrue(response.get(1).isResult());
+ assertNotNull(response.get(1).result());
+ assertTrue(response.get(1).result().hits().total().value() > 0);
+ });
+ }
+
+ @Test
+ void testUpdateWithMap() {
+ Map<String, String> map = createIndexedData();
+ String indexId = template.requestBody("direct:index", map, String.class);
+ assertNotNull(indexId, "indexId should be set");
+
+ Map<String, String> newMap = new HashMap<>();
+ newMap.put(createPrefix() + "key2", createPrefix() + "value2");
+ Map<String, Object> headers = new HashMap<>();
+ headers.put(ElasticsearchConstants.PARAM_INDEX_ID, indexId);
+ indexId = template.requestBodyAndHeaders("direct:update", Map.of("doc", newMap), headers, String.class);
+ assertNotNull(indexId, "indexId should be set");
+
+ //now, verify GET succeeded
+ GetResponse<?> response = template.requestBody("direct:get", indexId, GetResponse.class);
+ assertNotNull(response, "response should not be null");
+ assertNotNull(response.source(), "response source should not be null");
+ assertInstanceOf(ObjectNode.class, response.source(), "response source should be a ObjectNode");
+ assertTrue(((ObjectNode) response.source()).has(createPrefix() + "key2"));
+ assertEquals(createPrefix() + "value2", ((ObjectNode) response.source()).get(createPrefix() + "key2").asText());
+ }
+
+ @Test
+ void testGetWithHeaders() {
+ //first, Index a value
+ Map<String, String> map = createIndexedData();
+ Map<String, Object> headers = new HashMap<>();
+ headers.put(ElasticsearchConstants.PARAM_OPERATION, ElasticsearchOperation.Index);
+ headers.put(ElasticsearchConstants.PARAM_INDEX_NAME, "twitter");
+
+ String indexId = template.requestBodyAndHeaders("direct:start", map, headers, String.class);
+
+ //now, verify GET
+ headers.put(ElasticsearchConstants.PARAM_OPERATION, ElasticsearchOperation.GetById);
+ GetResponse<?> response = template.requestBodyAndHeaders("direct:start", indexId, headers, GetResponse.class);
+ assertNotNull(response, "response should not be null");
+ assertNotNull(response.source(), "response source should not be null");
+ }
+
+ @Test
+ void testExistsWithHeaders() {
+ //first, Index a value
+ Map<String, String> map = createIndexedData();
+ Map<String, Object> headers = new HashMap<>();
+ headers.put(ElasticsearchConstants.PARAM_OPERATION, ElasticsearchOperation.Index);
+ headers.put(ElasticsearchConstants.PARAM_INDEX_NAME, "twitter");
+
+ template.requestBodyAndHeaders("direct:start", map, headers, String.class);
+
+ //now, verify GET
+ headers.put(ElasticsearchConstants.PARAM_OPERATION, ElasticsearchOperation.Exists);
+ headers.put(ElasticsearchConstants.PARAM_INDEX_NAME, "twitter");
+ Boolean exists = template.requestBodyAndHeaders("direct:exists", "", headers, Boolean.class);
+ assertNotNull(exists, "response should not be null");
+ assertTrue(exists, "Index should exists");
+ }
+
+ @Test
+ void testNotExistsWithHeaders() {
+ Map<String, Object> headers = new HashMap<>();
+ headers.put(ElasticsearchConstants.PARAM_OPERATION, ElasticsearchOperation.Exists);
+ headers.put(ElasticsearchConstants.PARAM_INDEX_NAME, "twitter-tweet");
+ Boolean exists = template.requestBodyAndHeaders("direct:exists", "", headers, Boolean.class);
+ assertNotNull(exists, "response should not be null");
+ assertFalse(exists, "Index should not exists");
+ }
+
+ @Test
+ void testDeleteWithHeaders() {
+ //first, Index a value
+ Map<String, String> map = createIndexedData();
+ Map<String, Object> headers = new HashMap<>();
+ headers.put(ElasticsearchConstants.PARAM_OPERATION, ElasticsearchOperation.Index);
+ headers.put(ElasticsearchConstants.PARAM_INDEX_NAME, "twitter");
+
+ String indexId = template.requestBodyAndHeaders("direct:start", map, headers, String.class);
+
+ //now, verify GET
+ headers.put(ElasticsearchConstants.PARAM_OPERATION, ElasticsearchOperation.GetById);
+ GetResponse<?> response = template.requestBodyAndHeaders("direct:start", indexId, headers, GetResponse.class);
+ assertNotNull(response, "response should not be null");
+ assertNotNull(response.source(), "response source should not be null");
+
+ //now, perform Delete
+ headers.put(ElasticsearchConstants.PARAM_OPERATION, ElasticsearchOperation.Delete);
+ Result deleteResponse
+ = template.requestBodyAndHeaders("direct:start", indexId, headers, Result.class);
+ assertEquals(Result.Deleted, deleteResponse, "response should not be null");
+
+ //now, verify GET fails to find the indexed value
+ headers.put(ElasticsearchConstants.PARAM_OPERATION, ElasticsearchOperation.GetById);
+ response = template.requestBodyAndHeaders("direct:start", indexId, headers, GetResponse.class);
+ assertNotNull(response, "response should not be null");
+ assertNull(response.source(), "response source should be null");
+ }
+
+ @Test
+ void testUpdateWithIDInHeader() {
+ Map<String, String> map = createIndexedData();
+ Map<String, Object> headers = new HashMap<>();
+ headers.put(ElasticsearchConstants.PARAM_OPERATION, ElasticsearchOperation.Index);
+ headers.put(ElasticsearchConstants.PARAM_INDEX_NAME, "twitter");
+ headers.put(ElasticsearchConstants.PARAM_INDEX_ID, "123");
+
+ String indexId = template.requestBodyAndHeaders("direct:start", map, headers, String.class);
+ assertNotNull(indexId, "indexId should be set");
+ assertEquals("123", indexId, "indexId should be equals to the provided id");
+
+ headers.put(ElasticsearchConstants.PARAM_OPERATION, ElasticsearchOperation.Update);
+
+ indexId = template.requestBodyAndHeaders("direct:start", Map.of("doc", map), headers, String.class);
+ assertNotNull(indexId, "indexId should be set");
+ assertEquals("123", indexId, "indexId should be equals to the provided id");
+ }
+
+ @Test
+ void testGetRequestBody() {
+ String prefix = createPrefix();
+
+ // given
+ GetRequest.Builder builder = new GetRequest.Builder().index(prefix + "foo");
+
+ // when
+ String documentId = template.requestBody("direct:index",
+ new IndexRequest.Builder<>()
+ .index(prefix + "foo")
+ .id(prefix + "testId")
+ .document(Map.of(prefix + "content", prefix + "hello")),
+ String.class);
+ GetResponse<?> response = template.requestBody("direct:get",
+ builder.id(documentId), GetResponse.class);
+
+ // then
+ assertThat(response, notNullValue());
+
+ assertThat(response.source(), notNullValue());
+ ObjectNode node = (ObjectNode) response.source();
+ assertThat(node.has(prefix + "content"), equalTo(true));
+ assertThat(node.get(prefix + "content").asText(), equalTo(prefix + "hello"));
+ }
+
+ @Test
+ void testDeleteWithBuilder() {
+ String prefix = createPrefix();
+
+ // given
+ String documentId = template.requestBody("direct:index",
+ new IndexRequest.Builder<>()
+ .index(prefix + "foo")
+ .id(prefix + "testId")
+ .document(Map.of(prefix + "content", prefix + "hello")),
+ String.class);
+
+ GetResponse<?> getResponse = template.requestBodyAndHeader(
+ "direct:get", documentId, ElasticsearchConstants.PARAM_INDEX_NAME, prefix + "foo", GetResponse.class);
+ assertNotNull(getResponse, "response should not be null");
+ assertNotNull(getResponse.source(), "response source should not be null");
+
+ // when
+ Result response
+ = template.requestBody("direct:delete", new DeleteRequest.Builder().index(prefix + "foo").id(documentId),
+ Result.class);
+
+ // then
+ assertThat(response, equalTo(Result.Deleted));
+ getResponse = template.requestBodyAndHeader(
+ "direct:get", documentId, ElasticsearchConstants.PARAM_INDEX_NAME, prefix + "foo", GetResponse.class);
+ assertNotNull(getResponse, "response should not be null");
+ assertNull(getResponse.source(), "response source should be null");
+ }
+
+ @Test
+ void testUpdateWithString() {
+ Map<String, String> map = createIndexedData();
+ String indexId = template.requestBody("direct:index", map, String.class);
+ assertNotNull(indexId, "indexId should be set");
+ String key = map.keySet().iterator().next();
+ Object body = String.format("{ \"doc\": {\"%s\" : \"testUpdateWithString-updated\"}}", key);
+
+ Map<String, Object> headers = new HashMap<>();
+ headers.put(ElasticsearchConstants.PARAM_INDEX_ID, indexId);
+ indexId = template.requestBodyAndHeaders("direct:update", body, headers, String.class);
+ assertNotNull(indexId, "indexId should be set");
+
+ GetResponse<?> response = template.requestBody("direct:get", indexId, GetResponse.class);
+ assertThat(response.source(), notNullValue());
+ ObjectNode node = (ObjectNode) response.source();
+ assertThat(node.has(key), equalTo(true));
+ assertThat(node.get(key).asText(), equalTo("testUpdateWithString-updated"));
+ }
+
+ @Test
+ void testUpdateWithReader() {
+ Map<String, String> map = createIndexedData();
+ String indexId = template.requestBody("direct:index", map, String.class);
+ assertNotNull(indexId, "indexId should be set");
+ String key = map.keySet().iterator().next();
+ Object body = new StringReader(String.format("{ \"doc\": {\"%s\" : \"testUpdateWithReader-updated\"}}", key));
+
+ Map<String, Object> headers = new HashMap<>();
+ headers.put(ElasticsearchConstants.PARAM_INDEX_ID, indexId);
+ indexId = template.requestBodyAndHeaders("direct:update", body, headers, String.class);
+ assertNotNull(indexId, "indexId should be set");
+
+ GetResponse<?> response = template.requestBody("direct:get", indexId, GetResponse.class);
+ assertThat(response.source(), notNullValue());
+ ObjectNode node = (ObjectNode) response.source();
+ assertThat(node.has(key), equalTo(true));
+ assertThat(node.get(key).asText(), equalTo("testUpdateWithReader-updated"));
+ }
+
+ @Test
+ void testUpdateWithBytes() {
+ Map<String, String> map = createIndexedData();
+ String indexId = template.requestBody("direct:index", map, String.class);
+ assertNotNull(indexId, "indexId should be set");
+ String key = map.keySet().iterator().next();
+ Object body
+ = String.format("{ \"doc\": {\"%s\" : \"testUpdateWithBytes-updated\"}}", key).getBytes(StandardCharsets.UTF_8);
+
+ Map<String, Object> headers = new HashMap<>();
+ headers.put(ElasticsearchConstants.PARAM_INDEX_ID, indexId);
+ indexId = template.requestBodyAndHeaders("direct:update", body, headers, String.class);
+ assertNotNull(indexId, "indexId should be set");
+
+ GetResponse<?> response = template.requestBody("direct:get", indexId, GetResponse.class);
+ assertThat(response.source(), notNullValue());
+ ObjectNode node = (ObjectNode) response.source();
+ assertThat(node.has(key), equalTo(true));
+ assertThat(node.get(key).asText(), equalTo("testUpdateWithBytes-updated"));
+ }
+
+ @Test
+ void testUpdateWithInputStream() {
+ Map<String, String> map = createIndexedData();
+ String indexId = template.requestBody("direct:index", map, String.class);
+ assertNotNull(indexId, "indexId should be set");
+ String key = map.keySet().iterator().next();
+ Object body = new ByteArrayInputStream(
+ String.format("{ \"doc\": {\"%s\" : \"testUpdateWithInputStream-updated\"}}", key)
+ .getBytes(StandardCharsets.UTF_8));
+
+ Map<String, Object> headers = new HashMap<>();
+ headers.put(ElasticsearchConstants.PARAM_INDEX_ID, indexId);
+ indexId = template.requestBodyAndHeaders("direct:update", body, headers, String.class);
+ assertNotNull(indexId, "indexId should be set");
+
+ GetResponse<?> response = template.requestBody("direct:get", indexId, GetResponse.class);
+ assertThat(response.source(), notNullValue());
+ ObjectNode node = (ObjectNode) response.source();
+ assertThat(node.has(key), equalTo(true));
+ assertThat(node.get(key).asText(), equalTo("testUpdateWithInputStream-updated"));
+ }
+
+ @Test
+ void testUpdateWithDocumentType() {
+ Product product = new Product();
+ product.setId("book-world-records-2010");
+ product.setStockAvailable(200);
+ product.setPrice(80);
+ product.setDescription("The book of the year!");
+ product.setName("Guinness book of records 2010");
+
+ String indexId = template.requestBody("direct:index", product, String.class);
+ assertNotNull(indexId, "indexId should be set");
+
+ Product productUpdate = new Product();
+ productUpdate.setStockAvailable(250);
+ productUpdate.setPrice(82);
+ productUpdate.setName("Guinness book of records 2010 2nd edition");
+
+ Map<String, Object> headers = new HashMap<>();
+ headers.put(ElasticsearchConstants.PARAM_INDEX_ID, indexId);
+ headers.put(ElasticsearchConstants.PARAM_DOCUMENT_CLASS, Product.class);
+ indexId = template.requestBodyAndHeaders("direct:update", productUpdate, headers, String.class);
+ assertNotNull(indexId, "indexId should be set");
+
+ GetResponse<?> response = template.requestBodyAndHeader(
+ "direct:get", indexId, ElasticsearchConstants.PARAM_DOCUMENT_CLASS, Product.class, GetResponse.class);
+ assertThat(response.source(), notNullValue());
+ Product actual = (Product) response.source();
+ assertThat(actual.getId(), equalTo("book-world-records-2010"));
+ assertThat(actual.getStockAvailable(), equalTo(250));
+ assertThat(actual.getPrice(), equalTo(82d));
+ assertThat(actual.getDescription(), equalTo("The book of the year!"));
+ assertThat(actual.getName(), equalTo("Guinness book of records 2010 2nd edition"));
+ }
+
+ @Override
+ protected RouteBuilder createRouteBuilder() {
+ return new RouteBuilder() {
+ @Override
+ public void configure() {
+ from("direct:start")
+ .to("elasticsearch://elasticsearch?operation=Index");
+ from("direct:index")
+ .to("elasticsearch://elasticsearch?operation=Index&indexName=twitter");
+ from("direct:index-product")
+ .toF("elasticsearch://elasticsearch?operation=Index&indexName=twitter&documentClass=%s",
+ Product.class.getName());
+ from("direct:get")
+ .to("elasticsearch://elasticsearch?operation=GetById&indexName=twitter");
+ from("direct:multiget")
+ .to("elasticsearch://elasticsearch?operation=MultiGet&indexName=twitter");
+ from("direct:delete")
+ .to("elasticsearch://elasticsearch?operation=Delete&indexName=twitter");
+ from("direct:search")
+ .to("elasticsearch://elasticsearch?operation=Search&indexName=twitter");
+ from("direct:search-1")
+ .to("elasticsearch://elasticsearch?operation=Search");
+ from("direct:multiSearch")
+ .to("elasticsearch://elasticsearch?operation=MultiSearch");
+ from("direct:update")
+ .to("elasticsearch://elasticsearch?operation=Update&indexName=twitter");
+ from("direct:exists")
+ .to("elasticsearch://elasticsearch?operation=Exists");
+ }
+ };
+ }
+
+ public static class Product {
+
+ private String id;
+ private String name;
+ private String description;
+ private double price;
+ private int stockAvailable;
+
+ public String getId() {
+ return id;
+ }
+
+ public void setId(String id) {
+ this.id = id;
+ }
+
+ public String getName() {
+ return name;
+ }
+
+ public void setName(String name) {
+ this.name = name;
+ }
+
+ public String getDescription() {
+ return description;
+ }
+
+ public void setDescription(String description) {
+ this.description = description;
+ }
+
+ public double getPrice() {
+ return price;
+ }
+
+ public void setPrice(double price) {
+ this.price = price;
+ }
+
+ public int getStockAvailable() {
+ return stockAvailable;
+ }
+
+ public void setStockAvailable(int stockAvailable) {
+ this.stockAvailable = stockAvailable;
+ }
+
+ @Override
+ public boolean equals(Object o) {
+ if (this == o) {
+ return true;
+ }
+ if (o == null || getClass() != o.getClass()) {
+ return false;
+ }
+ Product product = (Product) o;
+ return Double.compare(product.price, price) == 0 && stockAvailable == product.stockAvailable
+ && Objects.equals(id, product.id) && Objects.equals(name, product.name)
+ && Objects.equals(description, product.description);
+ }
+
+ @Override
+ public int hashCode() {
+ return Objects.hash(id, name, description, price, stockAvailable);
+ }
+ }
+}
diff --git a/components/camel-elasticsearch/src/test/java/org/apache/camel/component/elasticsearch/integration/ElasticsearchIndexIT.java b/components/camel-elasticsearch/src/test/java/org/apache/camel/component/elasticsearch/integration/ElasticsearchIndexIT.java
new file mode 100644
index 00000000000..23fae982c0e
--- /dev/null
+++ b/components/camel-elasticsearch/src/test/java/org/apache/camel/component/elasticsearch/integration/ElasticsearchIndexIT.java
@@ -0,0 +1,129 @@
+/*
+ * 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.camel.component.elasticsearch.integration;
+
+import java.util.HashMap;
+import java.util.Map;
+
+import co.elastic.clients.elasticsearch.indices.DeleteIndexRequest;
+import org.apache.camel.builder.RouteBuilder;
+import org.apache.camel.component.elasticsearch.ElasticsearchConstants;
+import org.apache.camel.component.elasticsearch.ElasticsearchOperation;
+import org.junit.jupiter.api.Test;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertFalse;
+import static org.junit.jupiter.api.Assertions.assertNotNull;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
+class ElasticsearchIndexIT extends ElasticsearchTestSupport {
+
+ @Test
+ void testIndex() {
+ Map<String, String> map = createIndexedData();
+ String indexId = template.requestBody("direct:index", map, String.class);
+ assertNotNull(indexId, "indexId should be set");
+ }
+
+ @Test
+ void testIndexDeleteWithBuilder() {
+ Map<String, String> map = createIndexedData();
+ String indexId = template.requestBody("direct:index", map, String.class);
+ assertNotNull(indexId, "indexId should be set");
+
+ boolean exists = template.requestBody("direct:exists", null, Boolean.class);
+ assertTrue(exists, "index should be present");
+
+ DeleteIndexRequest.Builder builder = new DeleteIndexRequest.Builder().index("twitter");
+ Boolean status = template.requestBody("direct:deleteIndex", builder, Boolean.class);
+ assertEquals(true, status, "status should be 200");
+
+ exists = template.requestBody("direct:exists", null, Boolean.class);
+ assertFalse(exists, "index should be absent");
+ }
+
+ @Test
+ void testIndexDeleteWithString() {
+ Map<String, String> map = createIndexedData();
+ String indexId = template.requestBody("direct:index", map, String.class);
+ assertNotNull(indexId, "indexId should be set");
+
+ boolean exists = template.requestBody("direct:exists", null, Boolean.class);
+ assertTrue(exists, "index should be present");
+
+ Boolean status = template.requestBody("direct:deleteIndex", "twitter", Boolean.class);
+ assertEquals(true, status, "status should be 200");
+
+ exists = template.requestBody("direct:exists", null, Boolean.class);
+ assertFalse(exists, "index should be absent");
+ }
+
+ @Test
+ void testIndexWithHeaders() {
+ Map<String, String> map = createIndexedData();
+ Map<String, Object> headers = new HashMap<>();
+ headers.put(ElasticsearchConstants.PARAM_OPERATION, ElasticsearchOperation.Index);
+ headers.put(ElasticsearchConstants.PARAM_INDEX_NAME, "twitter");
+
+ String indexId = template.requestBodyAndHeaders("direct:start", map, headers, String.class);
+ assertNotNull(indexId, "indexId should be set");
+ }
+
+ @Test
+ void testIndexWithIDInHeader() {
+ Map<String, String> map = createIndexedData();
+ Map<String, Object> headers = new HashMap<>();
+ headers.put(ElasticsearchConstants.PARAM_OPERATION, ElasticsearchOperation.Index);
+ headers.put(ElasticsearchConstants.PARAM_INDEX_NAME, "twitter");
+ headers.put(ElasticsearchConstants.PARAM_INDEX_ID, "123");
+
+ String indexId = template.requestBodyAndHeaders("direct:start", map, headers, String.class);
+ assertNotNull(indexId, "indexId should be set");
+ assertEquals("123", indexId, "indexId should be equals to the provided id");
+ }
+
+ @Test
+ void testExists() {
+ boolean exists = template.requestBodyAndHeader(
+ "direct:exists", null, ElasticsearchConstants.PARAM_INDEX_NAME, "test_exists", Boolean.class);
+ assertFalse(exists, "index should be absent");
+
+ Map<String, String> map = createIndexedData();
+ template.sendBodyAndHeader("direct:index", map, ElasticsearchConstants.PARAM_INDEX_NAME, "test_exists");
+
+ exists = template.requestBodyAndHeader(
+ "direct:exists", null, ElasticsearchConstants.PARAM_INDEX_NAME, "test_exists", Boolean.class);
+ assertTrue(exists, "index should be present");
+ }
+
+ @Override
+ protected RouteBuilder createRouteBuilder() {
+ return new RouteBuilder() {
+ @Override
+ public void configure() {
+ from("direct:start")
+ .to("elasticsearch://elasticsearch");
+ from("direct:index")
+ .to("elasticsearch://elasticsearch?operation=Index&indexName=twitter");
+ from("direct:exists")
+ .to("elasticsearch://elasticsearch?operation=Exists&indexName=twitter");
+ from("direct:deleteIndex")
+ .to("elasticsearch://elasticsearch?operation=DeleteIndex&indexName=twitter");
+ }
+ };
+ }
+}
diff --git a/components/camel-elasticsearch/src/test/java/org/apache/camel/component/elasticsearch/integration/ElasticsearchPingIT.java b/components/camel-elasticsearch/src/test/java/org/apache/camel/component/elasticsearch/integration/ElasticsearchPingIT.java
new file mode 100644
index 00000000000..ca1ce701eba
--- /dev/null
+++ b/components/camel-elasticsearch/src/test/java/org/apache/camel/component/elasticsearch/integration/ElasticsearchPingIT.java
@@ -0,0 +1,42 @@
+/*
+ * 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.camel.component.elasticsearch.integration;
+
+import org.apache.camel.builder.RouteBuilder;
+import org.junit.jupiter.api.Test;
+
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
+class ElasticsearchPingIT extends ElasticsearchTestSupport {
+
+ @Test
+ void testPing() {
+ boolean pingResult = template.requestBody("direct:ping", "test", Boolean.class);
+ assertTrue(pingResult, "indexId should be set");
+ }
+
+ @Override
+ protected RouteBuilder createRouteBuilder() {
+ return new RouteBuilder() {
+ @Override
+ public void configure() {
+ from("direct:ping")
+ .to("elasticsearch://elasticsearch?operation=Ping");
+ }
+ };
+ }
+}
diff --git a/components/camel-elasticsearch/src/test/java/org/apache/camel/component/elasticsearch/integration/ElasticsearchScrollSearchIT.java b/components/camel-elasticsearch/src/test/java/org/apache/camel/component/elasticsearch/integration/ElasticsearchScrollSearchIT.java
new file mode 100644
index 00000000000..eae84e5949a
--- /dev/null
+++ b/components/camel-elasticsearch/src/test/java/org/apache/camel/component/elasticsearch/integration/ElasticsearchScrollSearchIT.java
@@ -0,0 +1,169 @@
+/*
+ * 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.camel.component.elasticsearch.integration;
+
+import java.io.IOException;
+import java.util.ArrayList;
+import java.util.Collection;
+import java.util.Iterator;
+import java.util.List;
+import java.util.Map;
+
+import co.elastic.clients.elasticsearch._types.query_dsl.MatchAllQuery;
+import co.elastic.clients.elasticsearch._types.query_dsl.Query;
+import co.elastic.clients.elasticsearch.core.SearchRequest;
+import co.elastic.clients.elasticsearch.core.search.Hit;
+import org.apache.camel.Exchange;
+import org.apache.camel.builder.AggregationStrategies;
+import org.apache.camel.builder.ExchangeBuilder;
+import org.apache.camel.builder.RouteBuilder;
+import org.apache.camel.component.elasticsearch.ElasticsearchScrollRequestIterator;
+import org.apache.camel.component.mock.MockEndpoint;
+import org.elasticsearch.client.Request;
+import org.elasticsearch.client.Response;
+import org.junit.jupiter.api.Test;
+
+import static org.apache.camel.component.elasticsearch.ElasticsearchConstants.PARAM_SCROLL;
+import static org.apache.camel.component.elasticsearch.ElasticsearchConstants.PARAM_SCROLL_KEEP_ALIVE_MS;
+import static org.apache.camel.component.elasticsearch.ElasticsearchConstants.PROPERTY_SCROLL_ES_QUERY_COUNT;
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertNotNull;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
+class ElasticsearchScrollSearchIT extends ElasticsearchTestSupport {
+
+ private static final String TWITTER_ES_INDEX_NAME = "twitter";
+ private static final String SPLIT_TWITTER_ES_INDEX_NAME = "split-" + TWITTER_ES_INDEX_NAME;
+
+ @Test
+ void testScrollSearch() throws IOException {
+ // add some documents
+ for (int i = 0; i < 10; i++) {
+ Map<String, String> map = createIndexedData();
+ String indexId = template.requestBody("direct:scroll-index", map, String.class);
+ assertNotNull(indexId, "indexId should be set");
+ }
+
+ // perform a refresh
+ Response refreshResponse = getClient().performRequest(new Request("post", "/" + TWITTER_ES_INDEX_NAME + "/_refresh"));
+ assertEquals(200, refreshResponse.getStatusLine().getStatusCode(), "Cannot perform a refresh");
+
+ SearchRequest.Builder req = getScrollSearchRequestBuilder(TWITTER_ES_INDEX_NAME);
+
+ Exchange exchange = ExchangeBuilder.anExchange(context)
+ .withHeader(PARAM_SCROLL_KEEP_ALIVE_MS, 50000)
+ .withHeader(PARAM_SCROLL, true)
+ .withBody(req)
+ .build();
+
+ exchange = template.send("direct:scroll-search", exchange);
+
+ try (ElasticsearchScrollRequestIterator<?> scrollRequestIterator
+ = exchange.getIn().getBody(ElasticsearchScrollRequestIterator.class)) {
+ assertNotNull(scrollRequestIterator, "response should not be null");
+
+ List<Hit<?>> result = new ArrayList<>();
+ scrollRequestIterator.forEachRemaining(result::add);
+
+ assertEquals(10, result.size(), "response hits should be == 10");
+ assertEquals(11, scrollRequestIterator.getRequestCount(), "11 request should have been send to Elasticsearch");
+ }
+
+ ElasticsearchScrollRequestIterator<?> scrollRequestIterator
+ = exchange.getIn().getBody(ElasticsearchScrollRequestIterator.class);
+ assertTrue(scrollRequestIterator.isClosed(), "iterator should be closed");
+ assertEquals(11, (int) exchange.getProperty(PROPERTY_SCROLL_ES_QUERY_COUNT, Integer.class),
+ "11 request should have been send to Elasticsearch");
+ }
+
+ @Test
+ void testScrollAndSplitSearch() throws IOException, InterruptedException {
+ // add some documents
+ for (int i = 0; i < 10; i++) {
+ Map<String, String> map = createIndexedData();
+ String indexId = template.requestBody("direct:scroll-n-split-index", map, String.class);
+ assertNotNull(indexId, "indexId should be set");
+ }
+
+ // perform a refresh
+ Response refreshResponse
+ = getClient().performRequest(new Request("post", "/" + SPLIT_TWITTER_ES_INDEX_NAME + "/_refresh"));
+ assertEquals(200, refreshResponse.getStatusLine().getStatusCode(), "Cannot perform a refresh");
+
+ MockEndpoint mock = getMockEndpoint("mock:output");
+ mock.expectedMessageCount(1);
+ mock.setResultWaitTime(8000);
+
+ SearchRequest.Builder req = getScrollSearchRequestBuilder(SPLIT_TWITTER_ES_INDEX_NAME);
+
+ Exchange exchange = ExchangeBuilder.anExchange(context).withBody(req).build();
+ exchange = template.send("direct:scroll-n-split-search", exchange);
+
+ // wait for aggregation
+ mock.assertIsSatisfied();
+ Iterator<Exchange> iterator = mock.getReceivedExchanges().iterator();
+ assertTrue(iterator.hasNext(), "response should contain 1 exchange");
+ Collection<?> aggregatedExchanges = iterator.next().getIn().getBody(Collection.class);
+
+ assertEquals(10, aggregatedExchanges.size(), "response hits should be == 10");
+
+ ElasticsearchScrollRequestIterator<?> scrollRequestIterator
+ = exchange.getIn().getBody(ElasticsearchScrollRequestIterator.class);
+ assertTrue(scrollRequestIterator.isClosed(), "iterator should be closed");
+ assertEquals(11, scrollRequestIterator.getRequestCount(), "11 request should have been send to Elasticsearch");
+ assertEquals(11, (int) exchange.getProperty(PROPERTY_SCROLL_ES_QUERY_COUNT, Integer.class),
+ "11 request should have been send to Elasticsearch");
+ }
+
+ private SearchRequest.Builder getScrollSearchRequestBuilder(String indexName) {
+ SearchRequest.Builder builder = new SearchRequest.Builder().index(indexName);
+ builder.size(1);
+ builder.query(new Query.Builder().matchAll(new MatchAllQuery.Builder().build()).build());
+ return builder;
+ }
+
+ @Override
+ protected RouteBuilder createRouteBuilder() {
+ return new RouteBuilder() {
+ @Override
+ public void configure() {
+ from("direct:scroll-index")
+ .to("elasticsearch://elasticsearch?operation=Index&indexName=" + TWITTER_ES_INDEX_NAME);
+ from("direct:scroll-search")
+ .to("elasticsearch://elasticsearch?operation=Search&indexName=" + TWITTER_ES_INDEX_NAME);
+
+ from("direct:scroll-n-split-index")
+ .to("elasticsearch://elasticsearch?operation=Index&indexName=" + SPLIT_TWITTER_ES_INDEX_NAME);
+ from("direct:scroll-n-split-search")
+ .to("elasticsearch://elasticsearch?"
+ + "useScroll=true&scrollKeepAliveMs=50000&operation=Search&indexName="
+ + SPLIT_TWITTER_ES_INDEX_NAME)
+ .split()
+ .body()
+ .streaming()
+ .parallelProcessing()
+ .threads(12)
+ .aggregate(AggregationStrategies.groupedExchange())
+ .constant(true)
+ .completionSize(20)
+ .completionTimeout(2000)
+ .to("mock:output")
+ .end();
+ }
+ };
+ }
+}
diff --git a/components/camel-elasticsearch/src/test/java/org/apache/camel/component/elasticsearch/integration/ElasticsearchSizeLimitIT.java b/components/camel-elasticsearch/src/test/java/org/apache/camel/component/elasticsearch/integration/ElasticsearchSizeLimitIT.java
new file mode 100644
index 00000000000..d7d741862db
--- /dev/null
+++ b/components/camel-elasticsearch/src/test/java/org/apache/camel/component/elasticsearch/integration/ElasticsearchSizeLimitIT.java
@@ -0,0 +1,68 @@
+/*
+ * 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.camel.component.elasticsearch.integration;
+
+import java.util.HashMap;
+import java.util.Map;
+import java.util.concurrent.TimeUnit;
+
+import co.elastic.clients.elasticsearch.core.search.HitsMetadata;
+import org.apache.camel.builder.RouteBuilder;
+import org.awaitility.Awaitility;
+import org.junit.jupiter.api.Test;
+
+class ElasticsearchSizeLimitIT extends ElasticsearchTestSupport {
+
+ @Test
+ void testSize() {
+ //put 4
+ template.requestBody("direct:index", getContent("content"), String.class);
+ template.requestBody("direct:index", getContent("content1"), String.class);
+ template.requestBody("direct:index", getContent("content2"), String.class);
+ template.requestBody("direct:index", getContent("content3"), String.class);
+
+ String query = "{\"query\":{\"match_all\": {}}}";
+
+ // the result may see stale data so use Awaitility
+ Awaitility.await().atMost(10, TimeUnit.SECONDS).until(() -> {
+ HitsMetadata<?> searchWithSizeTwo = template.requestBody("direct:searchWithSizeTwo", query, HitsMetadata.class);
+ HitsMetadata<?> searchFrom3 = template.requestBody("direct:searchFrom3", query, HitsMetadata.class);
+ return searchWithSizeTwo.hits().size() == 2 && searchFrom3.hits().size() == 1;
+ });
+ }
+
+ @Override
+ protected RouteBuilder createRouteBuilder() {
+ return new RouteBuilder() {
+ @Override
+ public void configure() {
+ from("direct:index")
+ .to("elasticsearch://elasticsearch?operation=Index&indexName=twitter");
+ from("direct:searchWithSizeTwo")
+ .to("elasticsearch://elasticsearch?operation=Search&indexName=twitter&size=2");
+ from("direct:searchFrom3")
+ .to("elasticsearch://elasticsearch?operation=Search&indexName=twitter&from=3");
+ }
+ };
+ }
+
+ private Map<String, String> getContent(String content) {
+ Map<String, String> map = new HashMap<>();
+ map.put("content", content);
+ return map;
+ }
+}
diff --git a/components/camel-elasticsearch/src/test/java/org/apache/camel/component/elasticsearch/integration/ElasticsearchTestSupport.java b/components/camel-elasticsearch/src/test/java/org/apache/camel/component/elasticsearch/integration/ElasticsearchTestSupport.java
new file mode 100644
index 00000000000..bbce86849ff
--- /dev/null
+++ b/components/camel-elasticsearch/src/test/java/org/apache/camel/component/elasticsearch/integration/ElasticsearchTestSupport.java
@@ -0,0 +1,170 @@
+/*
+ * 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.camel.component.elasticsearch.integration;
+
+import java.io.IOException;
+import java.nio.file.Files;
+import java.nio.file.Path;
+import java.util.HashMap;
+import java.util.Map;
+
+import javax.net.ssl.SSLContext;
+
+import co.elastic.clients.elasticsearch.ElasticsearchClient;
+import co.elastic.clients.json.jackson.JacksonJsonpMapper;
+import co.elastic.clients.transport.rest_client.RestClientTransport;
+import org.apache.camel.CamelContext;
+import org.apache.camel.component.elasticsearch.ElasticsearchComponent;
+import org.apache.camel.test.infra.elasticsearch.services.ElasticSearchLocalContainerService;
+import org.apache.camel.test.infra.elasticsearch.services.ElasticSearchService;
+import org.apache.camel.test.infra.elasticsearch.services.ElasticSearchServiceFactory;
+import org.apache.camel.test.junit5.CamelTestSupport;
+import org.apache.http.HttpHost;
+import org.apache.http.auth.AuthScope;
+import org.apache.http.auth.UsernamePasswordCredentials;
+import org.apache.http.client.CredentialsProvider;
+import org.apache.http.impl.client.BasicCredentialsProvider;
+import org.elasticsearch.client.RestClient;
+import org.elasticsearch.client.RestClientBuilder;
+import org.junit.jupiter.api.TestInstance;
+import org.junit.jupiter.api.extension.RegisterExtension;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+import org.testcontainers.utility.Base58;
+
+@TestInstance(TestInstance.Lifecycle.PER_CLASS)
+public class ElasticsearchTestSupport extends CamelTestSupport {
+
+ public static final int ELASTICSEARCH_DEFAULT_PORT = 9200;
+ public static final int ELASTICSEARCH_DEFAULT_TCP_PORT = 9300;
+
+ @RegisterExtension
+ public static ElasticSearchService service = ElasticSearchServiceFactory
+ .builder()
+ .addLocalMapping(ElasticsearchTestSupport::createElasticSearchService)
+ .build();
+
+ protected static String clusterName = "docker-cluster";
+ protected static RestClient restClient;
+ protected static ElasticsearchClient client;
+ private static Path certPath;
+ private static SSLContext sslContext;
+ private static final String USER_NAME = "elastic";
+ private static final String PASSWORD = "s3cret";
+ private static final Logger LOG = LoggerFactory.getLogger(ElasticsearchTestSupport.class);
+
+ private static ElasticSearchLocalContainerService createElasticSearchService() {
+ ElasticSearchLocalContainerService ret
+ = new ElasticSearchLocalContainerService("docker.elastic.co/elasticsearch/elasticsearch:8.4.1") {
+ @Override
+ public void registerProperties() {
+ super.registerProperties();
+ getContainer().caCertAsBytes().ifPresent(content -> {
+ try {
+ certPath = Files.createTempFile("http_ca", ".crt");
+ Files.write(certPath, content);
+ } catch (IOException e) {
+ throw new RuntimeException(e);
+ }
+ });
+ sslContext = getContainer().createSslContextFromCa();
+ }
+ };
+
+ ret.getContainer()
+ .withNetworkAliases("elasticsearch-" + Base58.randomString(6))
+ .withPassword(PASSWORD)
+ .withExposedPorts(ELASTICSEARCH_DEFAULT_PORT, ELASTICSEARCH_DEFAULT_TCP_PORT);
+
+ return ret;
+ }
+
+ @Override
+ protected void setupResources() throws Exception {
+ super.setupResources();
+ HttpHost host
+ = new HttpHost(service.getElasticSearchHost(), service.getPort(), "https");
+ final RestClientBuilder builder = RestClient.builder(host);
+ final CredentialsProvider credentialsProvider = new BasicCredentialsProvider();
+ credentialsProvider.setCredentials(AuthScope.ANY, new UsernamePasswordCredentials(USER_NAME, PASSWORD));
+ builder.setHttpClientConfigCallback(
+ httpClientBuilder -> {
+ httpClientBuilder.setDefaultCredentialsProvider(credentialsProvider);
+ httpClientBuilder.setSSLContext(sslContext);
+ return httpClientBuilder;
+ });
+ restClient = builder.build();
+ client = new ElasticsearchClient(new RestClientTransport(restClient, new JacksonJsonpMapper()));
+ }
+
+ @Override
+ protected void cleanupResources() throws Exception {
+ super.cleanupResources();
+ if (restClient != null) {
+ restClient.close();
+ }
+ }
+
+ @Override
+ protected CamelContext createCamelContext() throws Exception {
+ final ElasticsearchComponent elasticsearchComponent = new ElasticsearchComponent();
+ elasticsearchComponent.setEnableSSL(true);
+ elasticsearchComponent.setHostAddresses(service.getHttpHostAddress());
+ elasticsearchComponent.setUser(USER_NAME);
+ elasticsearchComponent.setPassword(PASSWORD);
+ elasticsearchComponent.setCertificatePath(certPath.toString());
+
+ CamelContext context = super.createCamelContext();
+ context.addComponent("elasticsearch", elasticsearchComponent);
+
+ return context;
+ }
+
+ /**
+ * As we don't delete the {@code target/data} folder for <b>each</b> test below (otherwise they would run much
+ * slower), we need to make sure there's no side effect of the same used data through creating unique indexes.
+ */
+ Map<String, String> createIndexedData(String... additionalPrefixes) {
+ String prefix = createPrefix();
+
+ // take over any potential prefixes we may have been asked for
+ if (additionalPrefixes.length > 0) {
+ StringBuilder sb = new StringBuilder(prefix);
+ for (String additionalPrefix : additionalPrefixes) {
+ sb.append(additionalPrefix).append("-");
+ }
+ prefix = sb.toString();
+ }
+
+ String key = prefix + "key";
+ String value = prefix + "value";
+ LOG.info("Creating indexed data using the key/value pair {} => {}", key, value);
+
+ Map<String, String> map = new HashMap<>();
+ map.put(key, value);
+ return map;
+ }
+
+ String createPrefix() {
+ // make use of the test method name to avoid collision
+ return getCurrentTestName().toLowerCase() + "-";
+ }
+
+ RestClient getClient() {
+ return restClient;
+ }
+}
diff --git a/components/camel-elasticsearch/src/test/resources/log4j2.properties b/components/camel-elasticsearch/src/test/resources/log4j2.properties
new file mode 100644
index 00000000000..7a4f0cc7897
--- /dev/null
+++ b/components/camel-elasticsearch/src/test/resources/log4j2.properties
@@ -0,0 +1,29 @@
+## ---------------------------------------------------------------------------
+## 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.
+## ---------------------------------------------------------------------------
+
+appender.file.type = File
+appender.file.name = file
+appender.file.fileName = target/camel-elasticsearch-test.log
+appender.file.layout.type = PatternLayout
+appender.file.layout.pattern = %d [%-15.15t] %-5p %-30.30c{1} - %m%n
+appender.stdout.type = Console
+appender.stdout.name = stdout
+appender.stdout.layout.type = PatternLayout
+appender.stdout.layout.pattern = %d [%-15.15t] %-5p %-30.30c{1} - %m%n
+rootLogger.level = INFO
+rootLogger.appenderRef.file.ref = file
+
diff --git a/components/pom.xml b/components/pom.xml
index 5bc2edf5256..6b4e6b722f0 100644
--- a/components/pom.xml
+++ b/components/pom.xml
@@ -135,6 +135,7 @@
<module>camel-dropbox</module>
<module>camel-dynamic-router</module>
<module>camel-ehcache</module>
+ <module>camel-elasticsearch</module>
<module>camel-elasticsearch-rest</module>
<module>camel-elytron</module>
<module>camel-etcd3</module>
diff --git a/parent/pom.xml b/parent/pom.xml
index a093b82e811..fdb2c293d61 100644
--- a/parent/pom.xml
+++ b/parent/pom.xml
@@ -160,6 +160,8 @@
<eddsa-version>0.3.0</eddsa-version>
<egit-github-core-version>2.1.5</egit-github-core-version>
<ehcache3-version>3.9.6</ehcache3-version>
+ <elasticsearch-java-client-version>8.4.1</elasticsearch-java-client-version>
+ <elasticsearch-java-client-sniffer-version>8.4.1</elasticsearch-java-client-sniffer-version>
<elasticsearch-rest-sniffer-version>7.10.2</elasticsearch-rest-sniffer-version>
<elasticsearch-rest-version>7.10.2</elasticsearch-rest-version>
<elytron-web>1.10.2.Final</elytron-web>
@@ -1350,6 +1352,11 @@
<artifactId>camel-elasticsearch-rest</artifactId>
<version>${project.version}</version>
</dependency>
+ <dependency>
+ <groupId>org.apache.camel</groupId>
+ <artifactId>camel-elasticsearch</artifactId>
+ <version>${project.version}</version>
+ </dependency>
<dependency>
<groupId>org.apache.camel</groupId>
<artifactId>camel-elytron</artifactId>