This is an automated email from the ASF dual-hosted git repository.
davsclaus pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/camel.git
The following commit(s) were added to refs/heads/main by this push:
new fb25ea44ee9d CAMEL-26232: camel-elasticsearch - send a built request
as it is, keep the whole JSON search body, take the DeleteIndex name from a
String body, fail a null Index/Update body and read a file as JSON (#29074)
fb25ea44ee9d is described below
commit fb25ea44ee9d2c2e69661537c50ee617c04a1220
Author: Andrea Cosentino <[email protected]>
AuthorDate: Tue Nov 3 18:04:39 2026 +0100
CAMEL-26232: camel-elasticsearch - send a built request as it is, keep the
whole JSON search body, take the DeleteIndex name from a String body, fail a
null Index/Update body and read a file as JSON (#29074)
Co-Authored-By: Claude Opus 5.5 <[email protected]>
Signed-off-by: Andrea Cosentino <[email protected]>
---
.../camel/catalog/components/elasticsearch.json | 2 +-
.../catalog/docs/elasticsearch-component.adoc | 31 +-
.../ElasticsearchActionRequestConverterLoader.java | 8 +
.../apache/camel/component/es/elasticsearch.json | 2 +-
.../src/main/docs/elasticsearch-component.adoc | 31 +-
.../camel/component/es/ElasticsearchEndpoint.java | 30 +-
.../camel/component/es/ElasticsearchProducer.java | 24 +-
.../ElasticsearchActionRequestConverter.java | 114 ++++++-
.../es/ElasticsearchProducerBodyTest.java | 339 +++++++++++++++++++++
.../ElasticsearchActionRequestConverterTest.java | 133 ++++++++
.../ROOT/pages/camel-4x-upgrade-guide-4_24.adoc | 20 ++
11 files changed, 676 insertions(+), 58 deletions(-)
diff --git
a/catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/components/elasticsearch.json
b/catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/components/elasticsearch.json
index f8930ee9c39d..3874e53d1b7c 100644
---
a/catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/components/elasticsearch.json
+++
b/catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/components/elasticsearch.json
@@ -55,7 +55,7 @@
"CamelElasticsearchEnableDocumentOnlyMode": { "index": 9, "kind":
"header", "displayName": "", "group": "producer", "label": "", "required":
false, "javaType": "Boolean", "deprecated": false, "deprecationNote": "",
"autowired": false, "secret": false, "defaultValue": "false", "description":
"Indicates whether the body of the message contains only documents.",
"constantName":
"org.apache.camel.component.es.ElasticsearchConstants#PARAM_DOCUMENT_MODE" }
},
"bodyTypes": {
- "producerIn": { "carriers": [ "text", "bytes", "stream", "map", "list",
"pojo" ], "byOption": [ { "option": "operation", "value": "Index", "carriers":
[ "text", "bytes", "stream", "map", "pojo" ] }, { "option": "operation",
"value": "Update", "carriers": [ "text", "bytes", "stream", "map", "pojo" ] },
{ "option": "operation", "value": "Bulk", "carriers": [ "list", "pojo" ] }, {
"option": "operation", "value": "GetById", "carriers": [ "text", "pojo" ] }, {
"option": "operation", "valu [...]
+ "producerIn": { "carriers": [ "text", "bytes", "stream", "file", "map",
"list", "pojo" ], "byOption": [ { "option": "operation", "value": "Index",
"carriers": [ "text", "bytes", "stream", "file", "map", "pojo" ] }, { "option":
"operation", "value": "Update", "carriers": [ "text", "bytes", "stream",
"file", "map", "pojo" ] }, { "option": "operation", "value": "Bulk",
"carriers": [ "list", "pojo" ] }, { "option": "operation", "value": "GetById",
"carriers": [ "text", "pojo" ] }, { "opt [...]
"producerOut": { "carriers": [ "text", "value", "pojo", "list", "iterator"
], "byOption": [ { "option": "operation", "value": "Index", "carriers": [
"text" ], "javaType": "java.lang.String" }, { "option": "operation", "value":
"Update", "carriers": [ "text" ], "javaType": "java.lang.String" }, { "option":
"operation", "value": "Bulk", "carriers": [ "list" ] }, { "option":
"operation", "value": "GetById", "carriers": [ "pojo" ], "javaType":
"co.elastic.clients.elasticsearch.core.GetRe [...]
},
"properties": {
diff --git
a/catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/docs/elasticsearch-component.adoc
b/catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/docs/elasticsearch-component.adoc
index 237d8111bfe4..ec04531a9a8c 100644
---
a/catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/docs/elasticsearch-component.adoc
+++
b/catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/docs/elasticsearch-component.adoc
@@ -48,37 +48,44 @@ 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.
+A request of the Elasticsearch client in the message body, such as an
`IndexRequest`, a `GetRequest` or a `SearchRequest`,
+sets the operation by its type, whatever the option or header say, and is sent
as it is: its index, id and the rest are
+not changed by the headers or options.
+
[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.
+|Index |*Map*, *String*, *byte[]*, *Reader*, *InputStream*, a file
(`java.io.File`, `Path` or the file of the file component), *IndexRequest* or
*IndexRequest.Builder* content to index |Adds content to an index and returns
the content's indexId in the body.
+A file is read as the JSON of the document, and a null body fails.
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.
+|GetById |*String*, *GetRequest* 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.
+|Delete |*String*, *DeleteRequest* 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* (its text is not used) or *DeleteIndexRequest.Builder*
|Deletes the index named by the indexName header or option, not the one in a
`String` body, and returns a `Boolean` in the body that is true when the
deletion is acknowledged.
+|DeleteIndex |*String* name of the index, *DeleteIndexRequest* or
*DeleteIndexRequest.Builder* |Deletes the index named by the indexName header
or option, or else the one named by a `String` body, and returns a `Boolean` in
the body that is true when the deletion is acknowledged.
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
+|Bulk |*Iterable*, *BulkRequest* 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.
+|Search |*Map*, *String*, *SearchRequest* or *SearchRequest.Builder* |Search
the content with the search request as a JSON `String` or a `Map`: its query,
size, from, sort, aggregations, _source and the rest are used.
+A JSON object with a single member other than query, such as `{"match":
{...}}`, is the query alone.
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".
+The size and from headers or options, when set, override those of the body.
-|MultiSearch |*MsearchRequest.Builder* |Multiple search in one
+|MultiSearch |*MsearchRequest* or *MsearchRequest.Builder* |Multiple search in
one
-|MultiGet |*Iterable<String>* or *MgetRequest.Builder* the id of the document
to retrieve |Multiple get in one
+|MultiGet |*Iterable<String>*, *MgetRequest* 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".
@@ -86,7 +93,8 @@ You can set the name of the target index by setting the
message header with the
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.
+|Update |*byte[]*, *InputStream*, *String*, *Reader*, *Map*, a file,
*UpdateRequest*, *UpdateRequest.Builder* or any document type content to update
|Updates content to an index and returns the content's indexId in the body.
+A file is read as JSON, and a null body fails.
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". Be aware of the fact that unlike the component
_camel-elasticsearch-rest_, by default, the expected content of
@@ -230,7 +238,8 @@ 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
+Pass in the search request as a JSON String or a Map: its size, sort,
aggregations and the other members
+are used with its query.
[tabs]
====
@@ -271,7 +280,7 @@ YAML::
._Java-only: ProducerTemplate test API_
[source,java]
----
-String query = "{\"query\":{\"match\":{\"content\":\"new release of
ApacheCamel\"}}}";
+String query = "{\"query\":{\"match\":{\"content\":\"new release of
ApacheCamel\"}},\"size\":5}";
HitsMetadata<?> response = template.requestBody("direct:search", query,
HitsMetadata.class);
----
diff --git
a/components/camel-elasticsearch/src/generated/java/org/apache/camel/component/es/converter/ElasticsearchActionRequestConverterLoader.java
b/components/camel-elasticsearch/src/generated/java/org/apache/camel/component/es/converter/ElasticsearchActionRequestConverterLoader.java
index ab733598ab46..01e4d2beec0d 100644
---
a/components/camel-elasticsearch/src/generated/java/org/apache/camel/component/es/converter/ElasticsearchActionRequestConverterLoader.java
+++
b/components/camel-elasticsearch/src/generated/java/org/apache/camel/component/es/converter/ElasticsearchActionRequestConverterLoader.java
@@ -84,6 +84,14 @@ public final class ElasticsearchActionRequestConverterLoader
implements TypeConv
}
return answer;
});
+ addTypeConverter(registry,
co.elastic.clients.elasticsearch.core.MsearchRequest.Builder.class,
java.lang.Object.class, false,
+ (type, exchange, value) -> {
+ Object answer =
org.apache.camel.component.es.converter.ElasticsearchActionRequestConverter.toMsearchRequestBuilder(value,
exchange);
+ if (false && answer == null) {
+ answer = Void.class;
+ }
+ return answer;
+ });
addTypeConverter(registry,
co.elastic.clients.elasticsearch.core.SearchRequest.Builder.class,
java.lang.Object.class, false,
(type, exchange, value) -> {
Object answer =
org.apache.camel.component.es.converter.ElasticsearchActionRequestConverter.toSearchRequestBuilder(value,
exchange);
diff --git
a/components/camel-elasticsearch/src/generated/resources/META-INF/org/apache/camel/component/es/elasticsearch.json
b/components/camel-elasticsearch/src/generated/resources/META-INF/org/apache/camel/component/es/elasticsearch.json
index f8930ee9c39d..3874e53d1b7c 100644
---
a/components/camel-elasticsearch/src/generated/resources/META-INF/org/apache/camel/component/es/elasticsearch.json
+++
b/components/camel-elasticsearch/src/generated/resources/META-INF/org/apache/camel/component/es/elasticsearch.json
@@ -55,7 +55,7 @@
"CamelElasticsearchEnableDocumentOnlyMode": { "index": 9, "kind":
"header", "displayName": "", "group": "producer", "label": "", "required":
false, "javaType": "Boolean", "deprecated": false, "deprecationNote": "",
"autowired": false, "secret": false, "defaultValue": "false", "description":
"Indicates whether the body of the message contains only documents.",
"constantName":
"org.apache.camel.component.es.ElasticsearchConstants#PARAM_DOCUMENT_MODE" }
},
"bodyTypes": {
- "producerIn": { "carriers": [ "text", "bytes", "stream", "map", "list",
"pojo" ], "byOption": [ { "option": "operation", "value": "Index", "carriers":
[ "text", "bytes", "stream", "map", "pojo" ] }, { "option": "operation",
"value": "Update", "carriers": [ "text", "bytes", "stream", "map", "pojo" ] },
{ "option": "operation", "value": "Bulk", "carriers": [ "list", "pojo" ] }, {
"option": "operation", "value": "GetById", "carriers": [ "text", "pojo" ] }, {
"option": "operation", "valu [...]
+ "producerIn": { "carriers": [ "text", "bytes", "stream", "file", "map",
"list", "pojo" ], "byOption": [ { "option": "operation", "value": "Index",
"carriers": [ "text", "bytes", "stream", "file", "map", "pojo" ] }, { "option":
"operation", "value": "Update", "carriers": [ "text", "bytes", "stream",
"file", "map", "pojo" ] }, { "option": "operation", "value": "Bulk",
"carriers": [ "list", "pojo" ] }, { "option": "operation", "value": "GetById",
"carriers": [ "text", "pojo" ] }, { "opt [...]
"producerOut": { "carriers": [ "text", "value", "pojo", "list", "iterator"
], "byOption": [ { "option": "operation", "value": "Index", "carriers": [
"text" ], "javaType": "java.lang.String" }, { "option": "operation", "value":
"Update", "carriers": [ "text" ], "javaType": "java.lang.String" }, { "option":
"operation", "value": "Bulk", "carriers": [ "list" ] }, { "option":
"operation", "value": "GetById", "carriers": [ "pojo" ], "javaType":
"co.elastic.clients.elasticsearch.core.GetRe [...]
},
"properties": {
diff --git
a/components/camel-elasticsearch/src/main/docs/elasticsearch-component.adoc
b/components/camel-elasticsearch/src/main/docs/elasticsearch-component.adoc
index 237d8111bfe4..ec04531a9a8c 100644
--- a/components/camel-elasticsearch/src/main/docs/elasticsearch-component.adoc
+++ b/components/camel-elasticsearch/src/main/docs/elasticsearch-component.adoc
@@ -48,37 +48,44 @@ 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.
+A request of the Elasticsearch client in the message body, such as an
`IndexRequest`, a `GetRequest` or a `SearchRequest`,
+sets the operation by its type, whatever the option or header say, and is sent
as it is: its index, id and the rest are
+not changed by the headers or options.
+
[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.
+|Index |*Map*, *String*, *byte[]*, *Reader*, *InputStream*, a file
(`java.io.File`, `Path` or the file of the file component), *IndexRequest* or
*IndexRequest.Builder* content to index |Adds content to an index and returns
the content's indexId in the body.
+A file is read as the JSON of the document, and a null body fails.
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.
+|GetById |*String*, *GetRequest* 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.
+|Delete |*String*, *DeleteRequest* 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* (its text is not used) or *DeleteIndexRequest.Builder*
|Deletes the index named by the indexName header or option, not the one in a
`String` body, and returns a `Boolean` in the body that is true when the
deletion is acknowledged.
+|DeleteIndex |*String* name of the index, *DeleteIndexRequest* or
*DeleteIndexRequest.Builder* |Deletes the index named by the indexName header
or option, or else the one named by a `String` body, and returns a `Boolean` in
the body that is true when the deletion is acknowledged.
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
+|Bulk |*Iterable*, *BulkRequest* 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.
+|Search |*Map*, *String*, *SearchRequest* or *SearchRequest.Builder* |Search
the content with the search request as a JSON `String` or a `Map`: its query,
size, from, sort, aggregations, _source and the rest are used.
+A JSON object with a single member other than query, such as `{"match":
{...}}`, is the query alone.
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".
+The size and from headers or options, when set, override those of the body.
-|MultiSearch |*MsearchRequest.Builder* |Multiple search in one
+|MultiSearch |*MsearchRequest* or *MsearchRequest.Builder* |Multiple search in
one
-|MultiGet |*Iterable<String>* or *MgetRequest.Builder* the id of the document
to retrieve |Multiple get in one
+|MultiGet |*Iterable<String>*, *MgetRequest* 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".
@@ -86,7 +93,8 @@ You can set the name of the target index by setting the
message header with the
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.
+|Update |*byte[]*, *InputStream*, *String*, *Reader*, *Map*, a file,
*UpdateRequest*, *UpdateRequest.Builder* or any document type content to update
|Updates content to an index and returns the content's indexId in the body.
+A file is read as JSON, and a null body fails.
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". Be aware of the fact that unlike the component
_camel-elasticsearch-rest_, by default, the expected content of
@@ -230,7 +238,8 @@ 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
+Pass in the search request as a JSON String or a Map: its size, sort,
aggregations and the other members
+are used with its query.
[tabs]
====
@@ -271,7 +280,7 @@ YAML::
._Java-only: ProducerTemplate test API_
[source,java]
----
-String query = "{\"query\":{\"match\":{\"content\":\"new release of
ApacheCamel\"}}}";
+String query = "{\"query\":{\"match\":{\"content\":\"new release of
ApacheCamel\"}},\"size\":5}";
HitsMetadata<?> response = template.requestBody("direct:search", query,
HitsMetadata.class);
----
diff --git
a/components/camel-elasticsearch/src/main/java/org/apache/camel/component/es/ElasticsearchEndpoint.java
b/components/camel-elasticsearch/src/main/java/org/apache/camel/component/es/ElasticsearchEndpoint.java
index 2d5b0936a843..45e0ad47fd5d 100644
---
a/components/camel-elasticsearch/src/main/java/org/apache/camel/component/es/ElasticsearchEndpoint.java
+++
b/components/camel-elasticsearch/src/main/java/org/apache/camel/component/es/ElasticsearchEndpoint.java
@@ -36,16 +36,16 @@ import org.elasticsearch.client.RestClient;
syntax = "elasticsearch:clusterName", producerOnly = true,
category = { Category.SEARCH, Category.MONITORING }, headersClass
= ElasticsearchConstants.class)
@BodyTypes(producerIn = @BodyType(value = {
- Carrier.TEXT, Carrier.BYTES, Carrier.STREAM, Carrier.MAP,
Carrier.LIST, Carrier.POJO },
+ Carrier.TEXT, Carrier.BYTES, Carrier.STREAM, Carrier.FILE,
Carrier.MAP, Carrier.LIST, Carrier.POJO },
byOption = {
@BodyType.When(option = "operation",
value = "Index",
carriers = {
- Carrier.TEXT,
Carrier.BYTES, Carrier.STREAM, Carrier.MAP,
- Carrier.POJO
}),
+ Carrier.TEXT,
Carrier.BYTES, Carrier.STREAM, Carrier.FILE,
+ Carrier.MAP,
Carrier.POJO }),
@BodyType.When(option = "operation",
value = "Update",
carriers = {
- Carrier.TEXT,
Carrier.BYTES, Carrier.STREAM, Carrier.MAP,
- Carrier.POJO
}),
+ Carrier.TEXT,
Carrier.BYTES, Carrier.STREAM, Carrier.FILE,
+ Carrier.MAP,
Carrier.POJO }),
@BodyType.When(option = "operation",
value = "Bulk",
carriers = {
Carrier.LIST, Carrier.POJO }),
@BodyType.When(option = "operation",
value = "GetById",
@@ -62,16 +62,18 @@ import org.elasticsearch.client.RestClient;
carriers = {
Carrier.TEXT, Carrier.MAP, Carrier.POJO }),
@BodyType.When(option = "operation",
value = "Exists", carriers = Carrier.ANY),
@BodyType.When(option = "operation",
value = "Ping", carriers = Carrier.ANY) },
- description = "By the operation, set by the
operation option or the CamelElasticsearchOperation header."
- + " Index and Update take the
document as JSON text, bytes or a stream, a Map"
- + " or a POJO (written as
JSON), or the request builder of the Elasticsearch"
- + " client; text that is not
JSON, a List, a java.io.File or no body fail."
+ description = "By the operation, set by the
operation option or the CamelElasticsearchOperation header,"
+ + " or by the request of the
Elasticsearch client in the body. Index and Update"
+ + " take the document as JSON
text, bytes, a stream or a file of it, or a Map"
+ + " or a POJO (written as
JSON); text that is not JSON, a List or no body fail."
+ " Update expects the Update
API form, {\"doc\": {...}}, unless enableDocumentOnlyMode is set. GetById and
Delete take the id as text, Bulk and MultiGet a List"
- + " of documents or ids,
Search a query as JSON text or a Map (only its query"
- + " member when it has one),
and MultiSearch only an MsearchRequest.Builder;"
- + " each also takes its
request builder. DeleteIndex needs a text body but"
- + " deletes the index of the
CamelElasticsearchIndexName header or indexName, not the one"
- + " in the body. Exists and
Ping do not use the body"),
+ + " of documents or ids,
Search the search request as JSON text or a Map, with"
+ + " its query, size, sort,
aggregations and the rest (an object with a single"
+ + " member other than query is
the query alone), and DeleteIndex the name of"
+ + " the index as text, used
when neither the CamelElasticsearchIndexName"
+ + " header nor indexName is
set. Each also takes its request of the"
+ + " Elasticsearch client, or
the builder of it, which is sent as it is; MultiSearch"
+ + " takes only those. Exists
and Ping do not use the body"),
producerOut = @BodyType(value = { Carrier.TEXT, Carrier.VALUE,
Carrier.POJO, Carrier.LIST, Carrier.ITERATOR },
byOption = {
@BodyType.When(option =
"operation", value = "Index", carriers = Carrier.TEXT,
diff --git
a/components/camel-elasticsearch/src/main/java/org/apache/camel/component/es/ElasticsearchProducer.java
b/components/camel-elasticsearch/src/main/java/org/apache/camel/component/es/ElasticsearchProducer.java
index c29ca8fa8015..58dc6f4862a7 100644
---
a/components/camel-elasticsearch/src/main/java/org/apache/camel/component/es/ElasticsearchProducer.java
+++
b/components/camel-elasticsearch/src/main/java/org/apache/camel/component/es/ElasticsearchProducer.java
@@ -239,7 +239,7 @@ class ElasticsearchProducer extends DefaultAsyncProducer {
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");
+ "Wrong body type. Only Map, String,
SearchRequest or SearchRequest.Builder is allowed as a type");
}
// is it a scroll request ?
boolean useScroll = message.getHeader(PARAM_SCROLL,
configuration.isUseScroll(), Boolean.class);
@@ -304,7 +304,8 @@ class ElasticsearchProducer extends DefaultAsyncProducer {
private void processMultiGetAsync(ActionContext ctx, Class<?>
documentClass) {
MgetRequest.Builder mgetRequestBuilder =
ctx.getMessage().getBody(MgetRequest.Builder.class);
if (mgetRequestBuilder == null) {
- throw new IllegalArgumentException("Wrong body type. Only
MgetRequest.Builder is allowed as a type");
+ throw new IllegalArgumentException(
+ "Wrong body type. Only Iterable, MgetRequest or
MgetRequest.Builder is allowed as a type");
}
onComplete(
ctx.getClient().mget(mgetRequestBuilder.build(), documentClass)
@@ -318,7 +319,8 @@ class ElasticsearchProducer extends DefaultAsyncProducer {
private void processMultiSearchAsync(ActionContext ctx, Class<?>
documentClass) {
MsearchRequest.Builder msearchRequestBuilder =
ctx.getMessage().getBody(MsearchRequest.Builder.class);
if (msearchRequestBuilder == null) {
- throw new IllegalArgumentException("Wrong body type. Only
MsearchRequest.Builder is allowed as a type");
+ throw new IllegalArgumentException(
+ "Wrong body type. Only MsearchRequest or
MsearchRequest.Builder is allowed as a type");
}
onComplete(
ctx.getClient().msearch(msearchRequestBuilder.build(),
documentClass)
@@ -345,7 +347,7 @@ class ElasticsearchProducer extends DefaultAsyncProducer {
DeleteIndexRequest.Builder deleteIndexRequestBuilder =
ctx.getMessage().getBody(DeleteIndexRequest.Builder.class);
if (deleteIndexRequestBuilder == null) {
throw new IllegalArgumentException(
- "Wrong body type. Only String or
DeleteIndexRequest.Builder is allowed as a type");
+ "Wrong body type. Only String, DeleteIndexRequest or
DeleteIndexRequest.Builder is allowed as a type");
}
onComplete(
ctx.getClient().indices().delete(deleteIndexRequestBuilder.build())
@@ -360,7 +362,7 @@ class ElasticsearchProducer extends DefaultAsyncProducer {
DeleteRequest.Builder deleteRequestBuilder =
ctx.getMessage().getBody(DeleteRequest.Builder.class);
if (deleteRequestBuilder == null) {
throw new IllegalArgumentException(
- "Wrong body type. Only String or DeleteRequest.Builder is
allowed as a type");
+ "Wrong body type. Only String, DeleteRequest or
DeleteRequest.Builder is allowed as a type");
}
onComplete(
ctx.getClient().delete(deleteRequestBuilder.build())
@@ -375,7 +377,7 @@ class ElasticsearchProducer extends DefaultAsyncProducer {
BulkRequest.Builder bulkRequestBuilder =
ctx.getMessage().getBody(BulkRequest.Builder.class);
if (bulkRequestBuilder == null) {
throw new IllegalArgumentException(
- "Wrong body type. Only Iterable or BulkRequest.Builder is
allowed as a type");
+ "Wrong body type. Only Iterable, BulkRequest or
BulkRequest.Builder is allowed as a type");
}
onComplete(
ctx.getClient().bulk(bulkRequestBuilder.build())
@@ -390,7 +392,7 @@ class ElasticsearchProducer extends DefaultAsyncProducer {
GetRequest.Builder getRequestBuilder =
ctx.getMessage().getBody(GetRequest.Builder.class);
if (getRequestBuilder == null) {
throw new IllegalArgumentException(
- "Wrong body type. Only String or GetRequest.Builder is
allowed as a type");
+ "Wrong body type. Only String, GetRequest or
GetRequest.Builder is allowed as a type");
}
onComplete(
ctx.getClient().get(getRequestBuilder.build(), documentClass),
@@ -401,6 +403,10 @@ class ElasticsearchProducer extends DefaultAsyncProducer {
* Updates asynchronously a document.
*/
private void processUpdateAsync(ActionContext ctx, Class<?> documentClass)
{
+ if (ctx.getMessage().getBody() == null) {
+ throw new IllegalArgumentException(
+ "The body of the Update operation cannot be null: it is
the update of the document, or its request");
+ }
UpdateRequest.Builder<?, ?> updateRequestBuilder =
ctx.getMessage().getBody(UpdateRequest.Builder.class);
onComplete(
ctx.getClient().update(updateRequestBuilder.build(),
documentClass)
@@ -412,6 +418,10 @@ class ElasticsearchProducer extends DefaultAsyncProducer {
* Indexes asynchronously a document.
*/
private void processIndexAsync(ActionContext ctx) {
+ if (ctx.getMessage().getBody() == null) {
+ throw new IllegalArgumentException(
+ "The body of the Index operation cannot be null: it is the
document to index, or its request");
+ }
IndexRequest.Builder<?> indexRequestBuilder =
ctx.getMessage().getBody(IndexRequest.Builder.class);
onComplete(
ctx.getClient().index(indexRequestBuilder.build())
diff --git
a/components/camel-elasticsearch/src/main/java/org/apache/camel/component/es/converter/ElasticsearchActionRequestConverter.java
b/components/camel-elasticsearch/src/main/java/org/apache/camel/component/es/converter/ElasticsearchActionRequestConverter.java
index 7e18c719dd8e..779712d2bfcf 100644
---
a/components/camel-elasticsearch/src/main/java/org/apache/camel/component/es/converter/ElasticsearchActionRequestConverter.java
+++
b/components/camel-elasticsearch/src/main/java/org/apache/camel/component/es/converter/ElasticsearchActionRequestConverter.java
@@ -17,10 +17,12 @@
package org.apache.camel.component.es.converter;
import java.io.ByteArrayInputStream;
+import java.io.File;
import java.io.IOException;
import java.io.InputStream;
import java.io.Reader;
import java.io.StringReader;
+import java.nio.file.Path;
import java.util.Map;
import co.elastic.clients.elasticsearch._types.WaitForActiveShards;
@@ -30,6 +32,7 @@ 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.core.bulk.BulkOperation;
@@ -43,6 +46,7 @@ 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.WrappedFile;
import org.apache.camel.component.es.ElasticsearchConstants;
import org.apache.camel.util.ObjectHelper;
import org.slf4j.Logger;
@@ -92,6 +96,11 @@ public final class ElasticsearchActionRequestConverter {
}
return indexReqBuilder;
}
+ if (document instanceof IndexRequest<?> indexRequest) {
+ // a built request is sent as it is, as a builder body is: the
producer gets a builder body without
+ // converting it, so the header rules of the builder above do not
apply to it either
+ return indexRequest.rebuild();
+ }
IndexRequest.Builder<Object> builder = new IndexRequest.Builder<>();
if (document instanceof byte[] byteArray) {
builder.withJson(new ByteArrayInputStream(byteArray));
@@ -104,6 +113,10 @@ public final class ElasticsearchActionRequestConverter {
} else if (document instanceof Map) {
ObjectMapper objectMapper = new ObjectMapper();
builder.withJson(new
StringReader(objectMapper.writeValueAsString(document)));
+ } else if (isFile(document)) {
+ try (InputStream inputStream = readFile(document, exchange)) {
+ builder.withJson(inputStream);
+ }
} else {
builder.document(document);
}
@@ -127,6 +140,10 @@ public final class ElasticsearchActionRequestConverter {
}
return updateReqBuilder;
}
+ if (document instanceof UpdateRequest<?, ?> updateRequest) {
+ // a built request is sent as it is, as a builder body is
+ return updateRequest.rebuild();
+ }
UpdateRequest.Builder<?, Object> builder = new
UpdateRequest.Builder<>();
Boolean enableDocumentOnlyMode
=
exchange.getIn().getHeader(ElasticsearchConstants.PARAM_DOCUMENT_MODE,
Boolean.FALSE, Boolean.class);
@@ -142,6 +159,10 @@ public final class ElasticsearchActionRequestConverter {
} else if (document instanceof Map) {
ObjectMapper objectMapper = new ObjectMapper();
mode.addDocToUpdateRequestBuilder(builder, new
StringReader(objectMapper.writeValueAsString(document)));
+ } else if (isFile(document)) {
+ try (InputStream inputStream = readFile(document, exchange)) {
+ mode.addDocToUpdateRequestBuilder(builder, inputStream);
+ }
} else {
builder.doc(document);
}
@@ -161,6 +182,10 @@ public final class ElasticsearchActionRequestConverter {
if (document instanceof GetRequest.Builder getReqBuilder) {
return getReqBuilder;
}
+ if (document instanceof GetRequest getRequest) {
+ // a built request is sent as it is, as a builder body is
+ return getRequest.rebuild();
+ }
if (document instanceof String string) {
return new GetRequest.Builder()
.index(exchange.getIn().getHeader(ElasticsearchConstants.PARAM_INDEX_NAME,
String.class))
@@ -174,6 +199,10 @@ public final class ElasticsearchActionRequestConverter {
if (document instanceof DeleteRequest.Builder deleteReqBuilder) {
return deleteReqBuilder;
}
+ if (document instanceof DeleteRequest deleteRequest) {
+ // a built request is sent as it is, as a builder body is
+ return deleteRequest.rebuild();
+ }
if (document instanceof String string) {
return new DeleteRequest.Builder()
.index(exchange.getIn().getHeader(ElasticsearchConstants.PARAM_INDEX_NAME,
String.class))
@@ -187,9 +216,22 @@ public final class ElasticsearchActionRequestConverter {
if (document instanceof DeleteIndexRequest.Builder
deleteIdxReqBuilder) {
return deleteIdxReqBuilder;
}
- if (document instanceof String) {
- return new DeleteIndexRequest.Builder()
-
.index(exchange.getIn().getHeader(ElasticsearchConstants.PARAM_INDEX_NAME,
String.class));
+ if (document instanceof DeleteIndexRequest deleteIndexRequest) {
+ // a built request is sent as it is, as a builder body is
+ return deleteIndexRequest.rebuild();
+ }
+ if (document instanceof String string) {
+ // the index of the header or of the indexName option, else the
one named by the body
+ String indexName =
exchange.getIn().getHeader(ElasticsearchConstants.PARAM_INDEX_NAME,
String.class);
+ if (ObjectHelper.isEmpty(indexName)) {
+ indexName = string;
+ }
+ DeleteIndexRequest.Builder builder = new
DeleteIndexRequest.Builder();
+ if (ObjectHelper.isNotEmpty(indexName)) {
+ builder.index(indexName);
+ }
+ // with no name at all, the request fails when it is built, as the
index is required
+ return builder;
}
return null;
}
@@ -199,6 +241,10 @@ public final class ElasticsearchActionRequestConverter {
if (documents instanceof MgetRequest.Builder mgetReqBuilder) {
return mgetReqBuilder;
}
+ if (documents instanceof MgetRequest mgetRequest) {
+ // a built request is sent as it is, as a builder body is
+ return mgetRequest.rebuild();
+ }
if (documents instanceof Iterable<?> documentIterable) {
MgetRequest.Builder builder = new MgetRequest.Builder();
builder.index(exchange.getIn().getHeader(ElasticsearchConstants.PARAM_INDEX_NAME,
String.class));
@@ -227,6 +273,10 @@ public final class ElasticsearchActionRequestConverter {
}
return searchReqBuilder;
}
+ if (queryObject instanceof SearchRequest searchRequest) {
+ // a built request is sent as it is, as a builder body is
+ return searchRequest.rebuild();
+ }
SearchRequest.Builder builder = new SearchRequest.Builder();
// Only setup the indexName if the message header has the
@@ -239,23 +289,18 @@ public final class ElasticsearchActionRequestConverter {
}
String queryText;
+ boolean queryAlone;
if (queryObject instanceof Map<?, ?> mapQuery) {
- // 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);
+ queryAlone = mapQuery.size() == 1 &&
!mapQuery.containsKey(ES_QUERY_DSL_PREFIX);
} else if (queryObject instanceof String queryString) {
queryText = queryString;
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();
- }
+ queryAlone = jsonTextObject != null && jsonTextObject.isObject()
&& jsonTextObject.size() == 1
+ && !jsonTextObject.has(ES_QUERY_DSL_PREFIX);
} else {
// Cannot convert the queryObject into SearchRequest
LOG.warn(
@@ -263,22 +308,45 @@ public final class ElasticsearchActionRequestConverter {
queryObject == null ? "null" :
queryObject.getClass().getName());
return null;
}
+ if (queryAlone) {
+ // a query alone, such as {"match": {...}}: an object with a
single member, other than query
+ builder.query(new Query.Builder().withJson(new
StringReader(queryText)).build());
+ } else {
+ // the body of a search request: its query, size, from, sort,
aggregations, _source and the rest
+ builder.withJson(new StringReader(queryText));
+ }
+ // the size and from of the headers or of the options, when set,
override those of the body
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 MsearchRequest.Builder toMsearchRequestBuilder(Object
queryObject, Exchange exchange) {
+ if (queryObject instanceof MsearchRequest.Builder msearchReqBuilder) {
+ return msearchReqBuilder;
+ }
+ if (queryObject instanceof MsearchRequest msearchRequest) {
+ // a built request is sent as it is, as a builder body is
+ return msearchRequest.rebuild();
+ }
+ return null;
+ }
+
@Converter
public static BulkRequest.Builder toBulkRequestBuilder(Object documents,
Exchange exchange) throws IOException {
if (documents instanceof BulkRequest.Builder bulkReqBuilder) {
return bulkReqBuilder;
}
+ if (documents instanceof BulkRequest bulkRequest) {
+ // a built request is sent as it is, as a builder body is
+ return bulkRequest.rebuild();
+ }
if (documents instanceof Iterable<?> documentIterable) {
BulkRequest.Builder builder = new BulkRequest.Builder();
builder.index(exchange.getIn().getHeader(ElasticsearchConstants.PARAM_INDEX_NAME,
String.class));
@@ -302,6 +370,26 @@ public final class ElasticsearchActionRequestConverter {
return null;
}
+ /**
+ * Whether the document is a file, a java.io.File, a Path or the file of
the file component, whose content is the
+ * JSON of the document.
+ */
+ private static boolean isFile(Object document) {
+ return document instanceof File || document instanceof Path ||
document instanceof WrappedFile;
+ }
+
+ /**
+ * Opens the content of the file, to read as JSON.
+ */
+ private static InputStream readFile(Object file, Exchange exchange) {
+ InputStream inputStream =
exchange.getContext().getTypeConverter().convertTo(InputStream.class, exchange,
file);
+ if (inputStream == null) {
+ throw new IllegalArgumentException(
+ "Cannot read the file of type " +
file.getClass().getName() + " as the JSON of the document");
+ }
+ return inputStream;
+ }
+
enum Mode {
DEFAULT {
@Override
diff --git
a/components/camel-elasticsearch/src/test/java/org/apache/camel/component/es/ElasticsearchProducerBodyTest.java
b/components/camel-elasticsearch/src/test/java/org/apache/camel/component/es/ElasticsearchProducerBodyTest.java
new file mode 100644
index 000000000000..36af5b848a20
--- /dev/null
+++
b/components/camel-elasticsearch/src/test/java/org/apache/camel/component/es/ElasticsearchProducerBodyTest.java
@@ -0,0 +1,339 @@
+/*
+ * 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.es;
+
+import java.io.File;
+import java.io.IOException;
+import java.io.OutputStream;
+import java.net.InetSocketAddress;
+import java.nio.charset.StandardCharsets;
+import java.nio.file.Files;
+import java.nio.file.Path;
+import java.util.List;
+import java.util.Map;
+import java.util.concurrent.CopyOnWriteArrayList;
+
+import co.elastic.clients.elasticsearch._types.Result;
+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.GetResponse;
+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.core.search.HitsMetadata;
+import co.elastic.clients.elasticsearch.indices.DeleteIndexRequest;
+import com.fasterxml.jackson.databind.JsonNode;
+import com.fasterxml.jackson.databind.ObjectMapper;
+import com.sun.net.httpserver.HttpExchange;
+import com.sun.net.httpserver.HttpServer;
+import org.apache.camel.CamelContext;
+import org.apache.camel.Exchange;
+import org.apache.camel.ProducerTemplate;
+import org.apache.camel.component.file.FileConsumer;
+import org.apache.camel.impl.DefaultCamelContext;
+import org.junit.jupiter.api.AfterEach;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.io.TempDir;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertInstanceOf;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
+/**
+ * The request the producer sends for the bodies it takes, checked against a
node that records the requests and answers
+ * them as Elasticsearch does.
+ */
+class ElasticsearchProducerBodyTest {
+
+ private static final ObjectMapper MAPPER = new ObjectMapper();
+
+ private static final String SHARDS =
"\"_shards\":{\"total\":1,\"successful\":1,\"failed\":0}";
+ private static final String WRITTEN =
"{\"_index\":\"idx\",\"_id\":\"1\",\"_version\":1,\"result\":\"%s\"," + SHARDS
+ +
",\"_seq_no\":0,\"_primary_term\":1}";
+ private static final String FOUND
+ =
"{\"_index\":\"idx\",\"_id\":\"1\",\"_version\":1,\"_seq_no\":0,\"_primary_term\":1,\"found\":true,"
+ + "\"_source\":{\"name\":\"camel\"}}";
+ private static final String HITS = "{\"took\":1,\"timed_out\":false," +
SHARDS
+ +
",\"hits\":{\"total\":{\"value\":0,\"relation\":\"eq\"},\"hits\":[]}}";
+
+ private final List<Recorded> requests = new CopyOnWriteArrayList<>();
+ private HttpServer server;
+ private CamelContext context;
+ private ProducerTemplate template;
+ private String uri;
+
+ @TempDir
+ Path dir;
+
+ @BeforeEach
+ void start() throws Exception {
+ server = HttpServer.create(new InetSocketAddress("127.0.0.1", 0), 0);
+ server.createContext("/", this::answer);
+ server.start();
+ uri = "elasticsearch:test-cluster?hostAddresses=127.0.0.1:" +
server.getAddress().getPort();
+ context = new DefaultCamelContext();
+ context.start();
+ template = context.createProducerTemplate();
+ }
+
+ @AfterEach
+ void stop() {
+ context.stop();
+ server.stop(0);
+ }
+
+ @Test
+ void aBuiltIndexRequestIsSentAsItIs() throws Exception {
+ IndexRequest<Object> request = IndexRequest.of(b ->
b.index("idx").id("1").document(Map.of("name", "camel")));
+
+ assertEquals("1", template.requestBody(uri + "&indexName=other",
request));
+
+ Recorded sent = single();
+ assertEquals("PUT", sent.method());
+ assertEquals("/idx/_doc/1", sent.path());
+ assertEquals(json("{\"name\":\"camel\"}"), json(sent.body()));
+ }
+
+ @Test
+ void aBuiltUpdateRequestIsSentAsItIs() throws Exception {
+ UpdateRequest<Object, Object> request
+ = UpdateRequest.of(b ->
b.index("idx").id("1").doc(Map.of("name", "camel")));
+
+ assertEquals("1", template.requestBody(uri + "&indexName=other",
request));
+
+ Recorded sent = single();
+ assertEquals("/idx/_update/1", sent.path());
+ assertEquals(json("{\"doc\":{\"name\":\"camel\"}}"),
json(sent.body()));
+ }
+
+ @Test
+ void aBuiltGetRequestIsSentAsItIs() {
+ GetResponse<?> response
+ = template.requestBody(uri, GetRequest.of(b ->
b.index("idx").id("1")), GetResponse.class);
+
+ assertTrue(response.found());
+ assertEquals("GET", single().method());
+ assertEquals("/idx/_doc/1", single().path());
+ }
+
+ @Test
+ void aBuiltDeleteRequestIsSentAsItIs() {
+ Object result = template.requestBody(uri, DeleteRequest.of(b ->
b.index("idx").id("1")));
+
+ assertEquals(Result.Deleted, result);
+ assertEquals("DELETE", single().method());
+ assertEquals("/idx/_doc/1", single().path());
+ }
+
+ @Test
+ void aBuiltSearchRequestIsSentAsItIs() throws Exception {
+ SearchRequest request = SearchRequest.of(b ->
b.index("idx").size(3).query(q -> q.matchAll(m -> m)));
+
+ assertInstanceOf(HitsMetadata.class, template.requestBody(uri +
"&indexName=other", request));
+
+ Recorded sent = single();
+ assertEquals("/idx/_search", sent.path());
+ assertEquals(json("{\"query\":{\"match_all\":{}},\"size\":3}"),
json(sent.body()));
+ }
+
+ @Test
+ void aBuiltMultiSearchRequestIsSentAsItIs() {
+ MsearchRequest request = MsearchRequest.of(b -> b.index("idx")
+ .searches(s -> s.header(h -> h).body(q -> q.query(m ->
m.matchAll(a -> a)))));
+
+ List<?> responses = template.requestBody(uri, request, List.class);
+
+ assertEquals(1, responses.size());
+ assertEquals("/idx/_msearch", single().path());
+ assertTrue(single().body().contains("match_all"), single().body());
+ }
+
+ @Test
+ void aBuiltMultiGetRequestIsSentAsItIs() throws Exception {
+ List<?> docs = template.requestBody(uri, MgetRequest.of(b ->
b.index("idx").ids("1")), List.class);
+
+ assertEquals(1, docs.size());
+ assertEquals("/idx/_mget", single().path());
+ assertEquals(json("{\"ids\":[\"1\"]}"), json(single().body()));
+ }
+
+ @Test
+ void aBuiltBulkRequestIsSentAsItIs() throws Exception {
+ BulkRequest request = BulkRequest.of(b -> b.operations(
+ o -> o.index(i ->
i.index("idx").id("1").document(Map.of("name", "camel")))));
+
+ List<?> items = template.requestBody(uri, request, List.class);
+
+ assertEquals(1, items.size());
+ assertEquals("/_bulk", single().path());
+ String[] lines = single().body().split("\n");
+ assertEquals(json("{\"index\":{\"_id\":\"1\",\"_index\":\"idx\"}}"),
json(lines[0]));
+ assertEquals(json("{\"name\":\"camel\"}"), json(lines[1]));
+ }
+
+ @Test
+ void aBuiltDeleteIndexRequestIsSentAsItIs() {
+ Object acknowledged = template.requestBody(uri,
DeleteIndexRequest.of(b -> b.index("idx")));
+
+ assertEquals(Boolean.TRUE, acknowledged);
+ assertEquals("DELETE", single().method());
+ assertEquals("/idx", single().path());
+ }
+
+ @Test
+ void deleteIndexDeletesTheIndexNamedByTheBodyWithoutHeaderOrOption() {
+ Object acknowledged = template.requestBody(uri +
"&operation=DeleteIndex", "idx");
+
+ assertEquals(Boolean.TRUE, acknowledged);
+ assertEquals("/idx", single().path());
+ }
+
+ @Test
+ void deleteIndexPrefersTheIndexNameOption() {
+ Object acknowledged = template.requestBody(uri +
"&operation=DeleteIndex&indexName=idx", "other");
+
+ assertEquals(Boolean.TRUE, acknowledged);
+ assertEquals("/idx", single().path());
+ }
+
+ @Test
+ void indexFailsClearlyWithoutBody() {
+ Exchange exchange = template.request(uri +
"&operation=Index&indexName=idx", e -> e.getMessage().setBody(null));
+
+ IllegalArgumentException e =
assertInstanceOf(IllegalArgumentException.class, exchange.getException());
+ assertEquals("The body of the Index operation cannot be null: it is
the document to index, or its request",
+ e.getMessage());
+ assertTrue(requests.isEmpty(), "a request was sent");
+ }
+
+ @Test
+ void updateFailsClearlyWithoutBody() {
+ Exchange exchange = template.request(uri +
"&operation=Update&indexName=idx", e -> {
+ e.getMessage().setHeader(ElasticsearchConstants.PARAM_INDEX_ID,
"1");
+ e.getMessage().setBody(null);
+ });
+
+ IllegalArgumentException e =
assertInstanceOf(IllegalArgumentException.class, exchange.getException());
+ assertEquals("The body of the Update operation cannot be null: it is
the update of the document, or its request",
+ e.getMessage());
+ assertTrue(requests.isEmpty(), "a request was sent");
+ }
+
+ @Test
+ void indexReadsAFileAsTheJsonOfTheDocument() throws Exception {
+ File file = write("doc.json", "{\"name\":\"camel\"}").toFile();
+
+ assertEquals("1", template.requestBody(uri +
"&operation=Index&indexName=idx", file));
+
+ assertEquals(json("{\"name\":\"camel\"}"), json(single().body()));
+ }
+
+ @Test
+ void indexReadsAPathAsTheJsonOfTheDocument() throws Exception {
+ Path path = write("doc.json", "{\"name\":\"camel\"}");
+
+ assertEquals("1", template.requestBody(uri +
"&operation=Index&indexName=idx", path));
+
+ assertEquals(json("{\"name\":\"camel\"}"), json(single().body()));
+ }
+
+ @Test
+ void indexReadsTheFileOfTheFileComponentAsTheJsonOfTheDocument() throws
Exception {
+ File file = write("doc.json", "{\"name\":\"camel\"}").toFile();
+ Object genericFile = FileConsumer.asGenericFile(dir.toString(), file,
null, false);
+
+ assertEquals("1", template.requestBody(uri +
"&operation=Index&indexName=idx", genericFile));
+
+ assertEquals(json("{\"name\":\"camel\"}"), json(single().body()));
+ }
+
+ @Test
+ void updateReadsAFileAsJson() throws Exception {
+ File file = write("update.json",
"{\"doc\":{\"name\":\"camel\"}}").toFile();
+
+ assertEquals("1", template.requestBodyAndHeader(uri +
"&operation=Update&indexName=idx", file,
+ ElasticsearchConstants.PARAM_INDEX_ID, "1"));
+
+ assertEquals("/idx/_update/1", single().path());
+ assertEquals(json("{\"doc\":{\"name\":\"camel\"}}"),
json(single().body()));
+ }
+
+ @Test
+ void getByIdStillTakesTheIdAsText() {
+ GetResponse<?> response = template.requestBody(uri +
"&operation=GetById&indexName=idx", "1", GetResponse.class);
+
+ assertTrue(response.found());
+ assertEquals("/idx/_doc/1", single().path());
+ }
+
+ private Path write(String name, String content) throws IOException {
+ return Files.writeString(dir.resolve(name), content);
+ }
+
+ private Recorded single() {
+ assertEquals(1, requests.size(), "requests sent: " + requests);
+ return requests.get(0);
+ }
+
+ private static JsonNode json(String text) throws IOException {
+ return MAPPER.readTree(text);
+ }
+
+ private void answer(HttpExchange exchange) throws IOException {
+ String body = new String(exchange.getRequestBody().readAllBytes(),
StandardCharsets.UTF_8);
+ String method = exchange.getRequestMethod();
+ String path = exchange.getRequestURI().getPath();
+ requests.add(new Recorded(method, path, body));
+ byte[] response = response(method,
path).getBytes(StandardCharsets.UTF_8);
+ exchange.getResponseHeaders().add("X-Elastic-Product",
"Elasticsearch");
+ exchange.getResponseHeaders().add("Content-Type", "application/json");
+ exchange.sendResponseHeaders(200, response.length);
+ try (OutputStream out = exchange.getResponseBody()) {
+ out.write(response);
+ }
+ }
+
+ private static String response(String method, String path) {
+ if (path.endsWith("/_search")) {
+ return HITS;
+ } else if (path.endsWith("/_msearch")) {
+ return "{\"took\":1,\"responses\":[" + HITS.replaceFirst("\\}$",
",\"status\":200}") + "]}";
+ } else if (path.endsWith("/_mget")) {
+ return "{\"docs\":[" + FOUND + "]}";
+ } else if (path.endsWith("/_bulk")) {
+ return "{\"took\":1,\"errors\":false,\"items\":[{\"index\":"
+ + String.format(WRITTEN, "created").replaceFirst("\\}$",
",\"status\":201}") + "}]}";
+ } else if (path.contains("/_update/")) {
+ return String.format(WRITTEN, "updated");
+ } else if (path.contains("/_doc")) {
+ return switch (method) {
+ case "GET" -> FOUND;
+ case "DELETE" -> String.format(WRITTEN, "deleted");
+ default -> String.format(WRITTEN, "created");
+ };
+ }
+ // the deletion of an index
+ return "{\"acknowledged\":true}";
+ }
+
+ private record Recorded(String method, String path, String body) {
+ }
+}
diff --git
a/components/camel-elasticsearch/src/test/java/org/apache/camel/component/es/converter/ElasticsearchActionRequestConverterTest.java
b/components/camel-elasticsearch/src/test/java/org/apache/camel/component/es/converter/ElasticsearchActionRequestConverterTest.java
index 796e859980b3..dd1385d5ee01 100644
---
a/components/camel-elasticsearch/src/test/java/org/apache/camel/component/es/converter/ElasticsearchActionRequestConverterTest.java
+++
b/components/camel-elasticsearch/src/test/java/org/apache/camel/component/es/converter/ElasticsearchActionRequestConverterTest.java
@@ -16,10 +16,17 @@
*/
package org.apache.camel.component.es.converter;
+import java.util.LinkedHashMap;
+import java.util.List;
import java.util.Map;
+import co.elastic.clients.elasticsearch._types.query_dsl.Query;
import co.elastic.clients.elasticsearch.core.IndexRequest;
+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.util.MissingRequiredPropertyException;
+import com.fasterxml.jackson.databind.ObjectMapper;
import org.apache.camel.CamelContext;
import org.apache.camel.Exchange;
import org.apache.camel.component.es.ElasticsearchConstants;
@@ -30,6 +37,9 @@ import org.junit.jupiter.api.BeforeEach;
import org.junit.jupiter.api.Test;
import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertNotNull;
+import static org.junit.jupiter.api.Assertions.assertThrows;
+import static org.junit.jupiter.api.Assertions.assertTrue;
public class ElasticsearchActionRequestConverterTest {
@@ -83,4 +93,127 @@ public class ElasticsearchActionRequestConverterTest {
// no CamelIndexId header -> the caller's id must be preserved, not
overwritten with null
assertEquals("original", result.build().id());
}
+
+ @Test
+ void aJsonSearchBodyKeepsItsSizeSortAggregationsAndSource() throws
Exception {
+ Exchange exchange = new DefaultExchange(context);
+ String body = """
+ {"query": {"match": {"name": "camel"}}, "size": 5, "from": 2,
+ "sort": [{"name": "asc"}], "aggs": {"names": {"terms":
{"field": "name"}}},
+ "_source": ["name"], "track_total_hits": true}
+ """;
+
+ SearchRequest request =
ElasticsearchActionRequestConverter.toSearchRequestBuilder(body,
exchange).build();
+
+ assertEquals(Query.Kind.Match, request.query()._kind());
+ assertEquals(5, request.size());
+ assertEquals(2, request.from());
+ assertEquals(1, request.sort().size());
+ assertEquals("name", request.sort().get(0).field().field());
+ assertTrue(request.aggregations().containsKey("names"));
+ assertNotNull(request.source());
+ assertNotNull(request.trackTotalHits());
+ }
+
+ @Test
+ void aJsonSearchBodyWithoutQueryIsASearchRequest() throws Exception {
+ Exchange exchange = new DefaultExchange(context);
+
+ SearchRequest request =
ElasticsearchActionRequestConverter.toSearchRequestBuilder(
+ "{\"size\": 0, \"aggs\": {\"names\": {\"terms\": {\"field\":
\"name\"}}}}", exchange).build();
+
+ assertEquals(0, request.size());
+ assertTrue(request.aggregations().containsKey("names"));
+ }
+
+ @Test
+ void aJsonQueryAloneIsStillTheQuery() throws Exception {
+ Exchange exchange = new DefaultExchange(context);
+ exchange.getIn().setHeader(ElasticsearchConstants.PARAM_INDEX_NAME,
"idx");
+
+ SearchRequest request = ElasticsearchActionRequestConverter
+ .toSearchRequestBuilder("{\"match\": {\"name\": \"camel\"}}",
exchange).build();
+
+ assertEquals(Query.Kind.Match, request.query()._kind());
+ assertEquals("name", request.query().match().field());
+ assertEquals(List.of("idx"), request.index());
+ }
+
+ @Test
+ void aMapSearchBodyKeepsItsSizeAndSort() throws Exception {
+ Exchange exchange = new DefaultExchange(context);
+ Map<String, Object> body = new LinkedHashMap<>();
+ body.put("query", Map.of("match", Map.of("name", "camel")));
+ body.put("size", 5);
+ body.put("sort", List.of(Map.of("name", "asc")));
+
+ SearchRequest request =
ElasticsearchActionRequestConverter.toSearchRequestBuilder(body,
exchange).build();
+
+ assertEquals(Query.Kind.Match, request.query()._kind());
+ assertEquals(5, request.size());
+ assertEquals(1, request.sort().size());
+ }
+
+ @Test
+ void aMapQueryAloneIsStillTheQuery() throws Exception {
+ Exchange exchange = new DefaultExchange(context);
+
+ SearchRequest request = ElasticsearchActionRequestConverter
+ .toSearchRequestBuilder(Map.of("match", Map.of("name",
"camel")), exchange).build();
+
+ assertEquals(Query.Kind.Match, request.query()._kind());
+ }
+
+ @Test
+ void aMapSearchBodyWhoseQueryIsNotAMap() throws Exception {
+ Exchange exchange = new DefaultExchange(context);
+ Map<String, Object> body = Map.of("query", new
ObjectMapper().readTree("{\"match_all\": {}}"));
+
+ SearchRequest request =
ElasticsearchActionRequestConverter.toSearchRequestBuilder(body,
exchange).build();
+
+ assertEquals(Query.Kind.MatchAll, request.query()._kind());
+ }
+
+ @Test
+ void theSizeAndFromHeadersOverrideThoseOfTheSearchBody() throws Exception {
+ Exchange exchange = new DefaultExchange(context);
+ exchange.getIn().setHeader(ElasticsearchConstants.PARAM_SIZE, 10);
+ exchange.getIn().setHeader(ElasticsearchConstants.PARAM_FROM, 20);
+
+ SearchRequest request =
ElasticsearchActionRequestConverter.toSearchRequestBuilder(
+ "{\"query\": {\"match_all\": {}}, \"size\": 5, \"from\": 2}",
exchange).build();
+
+ assertEquals(10, request.size());
+ assertEquals(20, request.from());
+ }
+
+ @Test
+ void deleteIndexTakesTheIndexNameOfTheBodyWithoutHeader() {
+ Exchange exchange = new DefaultExchange(context);
+
+ DeleteIndexRequest request
+ =
ElasticsearchActionRequestConverter.toDeleteIndexRequestBuilder("twitter",
exchange).build();
+
+ assertEquals(List.of("twitter"), request.index());
+ }
+
+ @Test
+ void deleteIndexPrefersTheIndexNameHeader() {
+ Exchange exchange = new DefaultExchange(context);
+ exchange.getIn().setHeader(ElasticsearchConstants.PARAM_INDEX_NAME,
"logs");
+
+ DeleteIndexRequest request
+ =
ElasticsearchActionRequestConverter.toDeleteIndexRequestBuilder("twitter",
exchange).build();
+
+ assertEquals(List.of("logs"), request.index());
+ }
+
+ @Test
+ void deleteIndexWithoutAnyIndexNameFailsBeforeSending() {
+ Exchange exchange = new DefaultExchange(context);
+
+ DeleteIndexRequest.Builder builder =
ElasticsearchActionRequestConverter.toDeleteIndexRequestBuilder(" ", exchange);
+
+ assertThrows(MissingRequiredPropertyException.class, builder::build);
+ }
}
diff --git
a/docs/user-manual/modules/ROOT/pages/camel-4x-upgrade-guide-4_24.adoc
b/docs/user-manual/modules/ROOT/pages/camel-4x-upgrade-guide-4_24.adoc
index f36c5a8746de..89212f6aeb35 100644
--- a/docs/user-manual/modules/ROOT/pages/camel-4x-upgrade-guide-4_24.adoc
+++ b/docs/user-manual/modules/ROOT/pages/camel-4x-upgrade-guide-4_24.adoc
@@ -5009,6 +5009,26 @@ correlation id of the request when it has none, so the
requester no longer times
The producer is unchanged: a route that changes the type of the body of a
consumed message and sends it on should
remove the `CamelSpringRabbitmqContentType` header.
+=== camel-elasticsearch - built requests, search bodies, DeleteIndex and file
bodies
+
+A request of the Elasticsearch client in the message body (an `IndexRequest`,
`UpdateRequest`, `GetRequest`,
+`DeleteRequest`, `SearchRequest`, `MsearchRequest`, `MgetRequest`,
`BulkRequest` or `DeleteIndexRequest`) is now sent
+as it is, like its builder. Before, a built `IndexRequest` was written as the
document itself, with a generated id, a
+built `UpdateRequest` failed without the id header, and the other built
requests failed with "Wrong body type".
+
+The `Search` operation now uses the whole search request of a JSON text or
`Map` body: its `size`, `from`, `sort`,
+aggregations, `_source` and the other members are no longer dropped, so the
hits returned can change. A member that
+is not part of a search request now fails the exchange with an error that
names it; before, it was dropped like every
+member other than `query`. A JSON object with a single member other than
`query`, such as `{"match": {...}}`, is
+still taken as the query alone. The `size` and `from` headers or options, when
set, still override those of the body.
+
+The `DeleteIndex` operation now deletes the index named by a `String` body
when neither the
+`CamelElasticsearchIndexName` header nor the `indexName` option is set.
Before, the call failed.
+
+The `Index` and `Update` operations now fail with an
`IllegalArgumentException` when the body is null, instead of
+failing with a Jackson error or an error of the server. A `java.io.File`, a
`java.nio.file.Path` or the file of the file
+component is now read as the JSON of the document, instead of being written as
a JSON string of its path.
+
=== camel-nats - the replies of an InOut consumer
A `nats` consumer with `exchangePattern=InOut` now answers with an empty reply
when the route ends with no body.
