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 515378b199b8 CAMEL-26246: camel-azure-cosmosdb - parse the JSON of an
item given as text, bytes, a stream or a file, and fail a List body of the item
operations (#28954)
515378b199b8 is described below
commit 515378b199b85551f302bc6674874854bf082a1e
Author: Andrea Cosentino <[email protected]>
AuthorDate: Mon Nov 2 23:22:08 2026 +0100
CAMEL-26246: camel-azure-cosmosdb - parse the JSON of an item given as
text, bytes, a stream or a file, and fail a List body of the item operations
(#28954)
Co-Authored-By: Claude Opus 5.5 <[email protected]>
---
.../camel/catalog/components/azure-cosmosdb.json | 4 +-
.../catalog/docs/azure-cosmosdb-component.adoc | 31 +--
.../component/azure/cosmosdb/azure-cosmosdb.json | 4 +-
.../src/main/docs/azure-cosmosdb-component.adoc | 31 +--
.../CosmosDbConfigurationOptionsProxy.java | 102 +++++++-
.../component/azure/cosmosdb/CosmosDbEndpoint.java | 31 ++-
.../cosmosdb/CosmosDbProducerItemBodyTest.java | 278 +++++++++++++++++++++
.../ROOT/pages/camel-4x-upgrade-guide-4_24.adoc | 12 +
8 files changed, 446 insertions(+), 47 deletions(-)
diff --git
a/catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/components/azure-cosmosdb.json
b/catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/components/azure-cosmosdb.json
index a68f0fa00040..e36481d9dc15 100644
---
a/catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/components/azure-cosmosdb.json
+++
b/catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/components/azure-cosmosdb.json
@@ -58,8 +58,8 @@
},
"bodyTypes": {
"consumerOut": { "carriers": [ "list" ], "javaType":
"java.util.List<java.util.Map<String, Object>>", "description": "The items of
one batch of the change feed: a List with a Map per item created or changed,
with the system fields of Cosmos DB such as _etag and _lsn" },
- "producerIn": { "carriers": [ "any" ], "byOption": [ { "option":
"operation", "value": "createItem", "carriers": [ "map", "pojo", "bytes" ] }, {
"option": "operation", "value": "upsertItem", "carriers": [ "map", "pojo",
"bytes" ] }, { "option": "operation", "value": "replaceItem", "carriers": [
"map", "pojo", "bytes" ] } ], "description": "The item for createItem,
upsertItem and replaceItem: a Map or a POJO, which Jackson serializes, or the
JSON of the item as bytes. A JSON String fa [...]
- "producerOut": { "carriers": [ "list" ], "javaType":
"java.util.List<com.azure.cosmos.models.CosmosDatabaseProperties>",
"dependsOn": [ "contentResponseOnWriteEnabled" ], "byOption": [ { "option":
"operation", "value": "queryDatabases", "carriers": [ "list" ], "javaType":
"java.util.List<com.azure.cosmos.models.CosmosDatabaseProperties>" }, {
"option": "operation", "value": "listContainers", "carriers": [ "list" ],
"javaType": "java.util.List<com.azure.cosmos.models.CosmosContainerPr [...]
+ "producerIn": { "carriers": [ "any" ], "byOption": [ { "option":
"operation", "value": "createItem", "carriers": [ "map", "pojo", "text",
"bytes", "stream", "file" ] }, { "option": "operation", "value": "upsertItem",
"carriers": [ "map", "pojo", "text", "bytes", "stream", "file" ] }, { "option":
"operation", "value": "replaceItem", "carriers": [ "map", "pojo", "text",
"bytes", "stream", "file" ] } ], "description": "The item for createItem,
upsertItem and replaceItem: a Map or a POJO [...]
+ "producerOut": { "carriers": [ "list" ], "javaType":
"java.util.List<com.azure.cosmos.models.CosmosDatabaseProperties>",
"dependsOn": [ "contentResponseOnWriteEnabled" ], "byOption": [ { "option":
"operation", "value": "queryDatabases", "carriers": [ "list" ], "javaType":
"java.util.List<com.azure.cosmos.models.CosmosDatabaseProperties>" }, {
"option": "operation", "value": "listContainers", "carriers": [ "list" ],
"javaType": "java.util.List<com.azure.cosmos.models.CosmosContainerPr [...]
},
"properties": {
"databaseName": { "index": 0, "kind": "path", "displayName": "Database
Name", "group": "common", "label": "", "required": false, "type": "string",
"javaType": "java.lang.String", "deprecated": false, "autowired": false,
"secret": false, "configurationClass":
"org.apache.camel.component.azure.cosmosdb.CosmosDbConfiguration",
"configurationField": "configuration", "description": "The name of the Cosmos
database that component should connect to. In case you are producing data and
have c [...]
diff --git
a/catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/docs/azure-cosmosdb-component.adoc
b/catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/docs/azure-cosmosdb-component.adoc
index 30cd7422817c..dbfffa5fbdee 100644
---
a/catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/docs/azure-cosmosdb-component.adoc
+++
b/catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/docs/azure-cosmosdb-component.adoc
@@ -149,15 +149,28 @@ For these operations, `databaseName` and `containerName`
is *required* for all o
|Operation |Description
|`deleteContainer` | Delete a container from the specified Azure CosmosDB
database.
|`replaceContainerThroughput` | Replace the throughput for the specified
Azure CosmosDB container.
-|`createItem` | *`itemPartitionKey` is required* Creates an item in the
specified container, it accepts POJO or key value as `Map<String, ?>`; a JSON
String fails: unmarshal it first.
-|`upsertItem` | *`itemPartitionKey` is required* Creates an item in the
specified container if it doesn't exist otherwise overwrite it if it exists, it
accepts POJO or key value as `Map<String, ?>`; a JSON String fails: unmarshal
it first.
-|`replaceItem` | *`itemPartitionKey` and `itemId` are required* Overwrites an
item in the specified container, it accepts POJO or key value as `Map<String,
?>`; a JSON String fails: unmarshal it first.
+|`createItem` | *`itemPartitionKey` is required* Creates an item in the
specified container, it accepts POJO or key value as `Map<String, ?>`, or the
JSON of the item (see below).
+|`upsertItem` | *`itemPartitionKey` is required* Creates an item in the
specified container if it doesn't exist otherwise overwrite it if it exists, it
accepts POJO or key value as `Map<String, ?>`, or the JSON of the item (see
below).
+|`replaceItem` | *`itemPartitionKey` and `itemId` are required* Overwrites an
item in the specified container, it accepts POJO or key value as `Map<String,
?>`, or the JSON of the item (see below).
|`deleteItem` | *`itemPartitionKey` and `itemId` are required* Deletes an
item in the specified container.
|`readItem` | *`itemPartitionKey` and `itemId` are required* Gets an item in
the specified container as `Map<String,?>` set in the exchange body message.
|`readAllItems` | *`itemPartitionKey`* Gets a list of items in the specified
container per the `itemPartitionKey` as `List<Map<String,?>>` set in the
exchange body message.
|`queryItems` | *`query` is required* Execute an SQL query against the
container level in order, for example, return only matching items per the SQL
query. It will set `List<Map<String,>?>` in the exchange message body.
|===
+The message body of `createItem`, `upsertItem` and `replaceItem` is one item:
a POJO or a `Map<String, ?>`, which the
+Azure SDK serializes with Jackson, or the JSON of the item as a `String`, a
`byte[]`, a stream or a file (such as the
+body of the file consumer), which the component parses into a `Map` first.
After the operation, the message body is
+the item as written by Cosmos DB, with its system fields such as `_etag`: a
`Map` for a `Map` or JSON body, and the
+class of the POJO for a POJO. With `contentResponseOnWriteEnabled=false` there
is no message body.
+
+A `byte[]`, an `InputStream` or a file is read as bytes, in which Jackson
detects JSON in UTF-8, UTF-16 or UTF-32. For
+JSON in another charset, such as a file that the file consumer reads with
`charset=ISO-8859-1`, convert the body to a
+`String` first, for example with `convertBodyTo(String.class)`, which reads
the file with its charset.
+
+These operations write one item: a `List` body, or JSON that is not an object
(such as a JSON array), fails with an
+`IllegalArgumentException`. To write several items, send each item in its own
exchange, for example with the Split EIP.
+
Refer to the example section in this page to learn how to use these operations
into your camel application.
== Examples
@@ -558,7 +571,6 @@ from("direct:start")
.setHeader("CamelAzureCosmosDbContainerPartitionKeyPath",
constant("partition"))
.setHeader("CamelAzureCosmosDbItemPartitionKey", constant("test-1"))
.setBody(constant("{\"id\":\"test-id-1\",\"partition\":\"test-1\",\"field1\":\"awesome!\"}"))
- .unmarshal().json()
.to("azure-cosmosdb://?operation=createItem")
.to("mock:result");
----
@@ -584,9 +596,6 @@ XML::
<setBody>
<constant>{"id":"test-id-1","partition":"test-1","field1":"awesome!"}</constant>
</setBody>
- <unmarshal>
- <json/>
- </unmarshal>
<to uri="azure-cosmosdb://?operation=createItem"/>
<to uri="mock:result"/>
</route>
@@ -614,8 +623,6 @@ YAML::
constant: "test-1"
- setBody:
constant:
'{"id":"test-id-1","partition":"test-1","field1":"awesome!"}'
- - unmarshal:
- json: {}
- to:
uri: azure-cosmosdb://
parameters:
@@ -639,7 +646,6 @@ from("direct:start")
.setHeader("CamelAzureCosmosDbItemPartitionKey", constant("test-1"))
.setHeader("CamelAzureCosmosDbItemId", constant("test-id-1"))
.setBody(constant("{\"id\":\"test-id-1\",\"partition\":\"test-1\",\"field1\":\"awesome!\"}"))
- .unmarshal().json()
.to("azure-cosmosdb://?operation=replaceItem")
.to("mock:result");
----
@@ -665,9 +671,6 @@ XML::
<setBody>
<constant>{"id":"test-id-1","partition":"test-1","field1":"awesome!"}</constant>
</setBody>
- <unmarshal>
- <json/>
- </unmarshal>
<to uri="azure-cosmosdb://?operation=replaceItem"/>
<to uri="mock:result"/>
</route>
@@ -695,8 +698,6 @@ YAML::
constant: "test-id-1"
- setBody:
constant:
'{"id":"test-id-1","partition":"test-1","field1":"awesome!"}'
- - unmarshal:
- json: {}
- to:
uri: azure-cosmosdb://
parameters:
diff --git
a/components/camel-azure/camel-azure-cosmosdb/src/generated/resources/META-INF/org/apache/camel/component/azure/cosmosdb/azure-cosmosdb.json
b/components/camel-azure/camel-azure-cosmosdb/src/generated/resources/META-INF/org/apache/camel/component/azure/cosmosdb/azure-cosmosdb.json
index a68f0fa00040..e36481d9dc15 100644
---
a/components/camel-azure/camel-azure-cosmosdb/src/generated/resources/META-INF/org/apache/camel/component/azure/cosmosdb/azure-cosmosdb.json
+++
b/components/camel-azure/camel-azure-cosmosdb/src/generated/resources/META-INF/org/apache/camel/component/azure/cosmosdb/azure-cosmosdb.json
@@ -58,8 +58,8 @@
},
"bodyTypes": {
"consumerOut": { "carriers": [ "list" ], "javaType":
"java.util.List<java.util.Map<String, Object>>", "description": "The items of
one batch of the change feed: a List with a Map per item created or changed,
with the system fields of Cosmos DB such as _etag and _lsn" },
- "producerIn": { "carriers": [ "any" ], "byOption": [ { "option":
"operation", "value": "createItem", "carriers": [ "map", "pojo", "bytes" ] }, {
"option": "operation", "value": "upsertItem", "carriers": [ "map", "pojo",
"bytes" ] }, { "option": "operation", "value": "replaceItem", "carriers": [
"map", "pojo", "bytes" ] } ], "description": "The item for createItem,
upsertItem and replaceItem: a Map or a POJO, which Jackson serializes, or the
JSON of the item as bytes. A JSON String fa [...]
- "producerOut": { "carriers": [ "list" ], "javaType":
"java.util.List<com.azure.cosmos.models.CosmosDatabaseProperties>",
"dependsOn": [ "contentResponseOnWriteEnabled" ], "byOption": [ { "option":
"operation", "value": "queryDatabases", "carriers": [ "list" ], "javaType":
"java.util.List<com.azure.cosmos.models.CosmosDatabaseProperties>" }, {
"option": "operation", "value": "listContainers", "carriers": [ "list" ],
"javaType": "java.util.List<com.azure.cosmos.models.CosmosContainerPr [...]
+ "producerIn": { "carriers": [ "any" ], "byOption": [ { "option":
"operation", "value": "createItem", "carriers": [ "map", "pojo", "text",
"bytes", "stream", "file" ] }, { "option": "operation", "value": "upsertItem",
"carriers": [ "map", "pojo", "text", "bytes", "stream", "file" ] }, { "option":
"operation", "value": "replaceItem", "carriers": [ "map", "pojo", "text",
"bytes", "stream", "file" ] } ], "description": "The item for createItem,
upsertItem and replaceItem: a Map or a POJO [...]
+ "producerOut": { "carriers": [ "list" ], "javaType":
"java.util.List<com.azure.cosmos.models.CosmosDatabaseProperties>",
"dependsOn": [ "contentResponseOnWriteEnabled" ], "byOption": [ { "option":
"operation", "value": "queryDatabases", "carriers": [ "list" ], "javaType":
"java.util.List<com.azure.cosmos.models.CosmosDatabaseProperties>" }, {
"option": "operation", "value": "listContainers", "carriers": [ "list" ],
"javaType": "java.util.List<com.azure.cosmos.models.CosmosContainerPr [...]
},
"properties": {
"databaseName": { "index": 0, "kind": "path", "displayName": "Database
Name", "group": "common", "label": "", "required": false, "type": "string",
"javaType": "java.lang.String", "deprecated": false, "autowired": false,
"secret": false, "configurationClass":
"org.apache.camel.component.azure.cosmosdb.CosmosDbConfiguration",
"configurationField": "configuration", "description": "The name of the Cosmos
database that component should connect to. In case you are producing data and
have c [...]
diff --git
a/components/camel-azure/camel-azure-cosmosdb/src/main/docs/azure-cosmosdb-component.adoc
b/components/camel-azure/camel-azure-cosmosdb/src/main/docs/azure-cosmosdb-component.adoc
index 30cd7422817c..dbfffa5fbdee 100644
---
a/components/camel-azure/camel-azure-cosmosdb/src/main/docs/azure-cosmosdb-component.adoc
+++
b/components/camel-azure/camel-azure-cosmosdb/src/main/docs/azure-cosmosdb-component.adoc
@@ -149,15 +149,28 @@ For these operations, `databaseName` and `containerName`
is *required* for all o
|Operation |Description
|`deleteContainer` | Delete a container from the specified Azure CosmosDB
database.
|`replaceContainerThroughput` | Replace the throughput for the specified
Azure CosmosDB container.
-|`createItem` | *`itemPartitionKey` is required* Creates an item in the
specified container, it accepts POJO or key value as `Map<String, ?>`; a JSON
String fails: unmarshal it first.
-|`upsertItem` | *`itemPartitionKey` is required* Creates an item in the
specified container if it doesn't exist otherwise overwrite it if it exists, it
accepts POJO or key value as `Map<String, ?>`; a JSON String fails: unmarshal
it first.
-|`replaceItem` | *`itemPartitionKey` and `itemId` are required* Overwrites an
item in the specified container, it accepts POJO or key value as `Map<String,
?>`; a JSON String fails: unmarshal it first.
+|`createItem` | *`itemPartitionKey` is required* Creates an item in the
specified container, it accepts POJO or key value as `Map<String, ?>`, or the
JSON of the item (see below).
+|`upsertItem` | *`itemPartitionKey` is required* Creates an item in the
specified container if it doesn't exist otherwise overwrite it if it exists, it
accepts POJO or key value as `Map<String, ?>`, or the JSON of the item (see
below).
+|`replaceItem` | *`itemPartitionKey` and `itemId` are required* Overwrites an
item in the specified container, it accepts POJO or key value as `Map<String,
?>`, or the JSON of the item (see below).
|`deleteItem` | *`itemPartitionKey` and `itemId` are required* Deletes an
item in the specified container.
|`readItem` | *`itemPartitionKey` and `itemId` are required* Gets an item in
the specified container as `Map<String,?>` set in the exchange body message.
|`readAllItems` | *`itemPartitionKey`* Gets a list of items in the specified
container per the `itemPartitionKey` as `List<Map<String,?>>` set in the
exchange body message.
|`queryItems` | *`query` is required* Execute an SQL query against the
container level in order, for example, return only matching items per the SQL
query. It will set `List<Map<String,>?>` in the exchange message body.
|===
+The message body of `createItem`, `upsertItem` and `replaceItem` is one item:
a POJO or a `Map<String, ?>`, which the
+Azure SDK serializes with Jackson, or the JSON of the item as a `String`, a
`byte[]`, a stream or a file (such as the
+body of the file consumer), which the component parses into a `Map` first.
After the operation, the message body is
+the item as written by Cosmos DB, with its system fields such as `_etag`: a
`Map` for a `Map` or JSON body, and the
+class of the POJO for a POJO. With `contentResponseOnWriteEnabled=false` there
is no message body.
+
+A `byte[]`, an `InputStream` or a file is read as bytes, in which Jackson
detects JSON in UTF-8, UTF-16 or UTF-32. For
+JSON in another charset, such as a file that the file consumer reads with
`charset=ISO-8859-1`, convert the body to a
+`String` first, for example with `convertBodyTo(String.class)`, which reads
the file with its charset.
+
+These operations write one item: a `List` body, or JSON that is not an object
(such as a JSON array), fails with an
+`IllegalArgumentException`. To write several items, send each item in its own
exchange, for example with the Split EIP.
+
Refer to the example section in this page to learn how to use these operations
into your camel application.
== Examples
@@ -558,7 +571,6 @@ from("direct:start")
.setHeader("CamelAzureCosmosDbContainerPartitionKeyPath",
constant("partition"))
.setHeader("CamelAzureCosmosDbItemPartitionKey", constant("test-1"))
.setBody(constant("{\"id\":\"test-id-1\",\"partition\":\"test-1\",\"field1\":\"awesome!\"}"))
- .unmarshal().json()
.to("azure-cosmosdb://?operation=createItem")
.to("mock:result");
----
@@ -584,9 +596,6 @@ XML::
<setBody>
<constant>{"id":"test-id-1","partition":"test-1","field1":"awesome!"}</constant>
</setBody>
- <unmarshal>
- <json/>
- </unmarshal>
<to uri="azure-cosmosdb://?operation=createItem"/>
<to uri="mock:result"/>
</route>
@@ -614,8 +623,6 @@ YAML::
constant: "test-1"
- setBody:
constant:
'{"id":"test-id-1","partition":"test-1","field1":"awesome!"}'
- - unmarshal:
- json: {}
- to:
uri: azure-cosmosdb://
parameters:
@@ -639,7 +646,6 @@ from("direct:start")
.setHeader("CamelAzureCosmosDbItemPartitionKey", constant("test-1"))
.setHeader("CamelAzureCosmosDbItemId", constant("test-id-1"))
.setBody(constant("{\"id\":\"test-id-1\",\"partition\":\"test-1\",\"field1\":\"awesome!\"}"))
- .unmarshal().json()
.to("azure-cosmosdb://?operation=replaceItem")
.to("mock:result");
----
@@ -665,9 +671,6 @@ XML::
<setBody>
<constant>{"id":"test-id-1","partition":"test-1","field1":"awesome!"}</constant>
</setBody>
- <unmarshal>
- <json/>
- </unmarshal>
<to uri="azure-cosmosdb://?operation=replaceItem"/>
<to uri="mock:result"/>
</route>
@@ -695,8 +698,6 @@ YAML::
constant: "test-id-1"
- setBody:
constant:
'{"id":"test-id-1","partition":"test-1","field1":"awesome!"}'
- - unmarshal:
- json: {}
- to:
uri: azure-cosmosdb://
parameters:
diff --git
a/components/camel-azure/camel-azure-cosmosdb/src/main/java/org/apache/camel/component/azure/cosmosdb/CosmosDbConfigurationOptionsProxy.java
b/components/camel-azure/camel-azure-cosmosdb/src/main/java/org/apache/camel/component/azure/cosmosdb/CosmosDbConfigurationOptionsProxy.java
index 8fb6704b814f..d188d2309e63 100644
---
a/components/camel-azure/camel-azure-cosmosdb/src/main/java/org/apache/camel/component/azure/cosmosdb/CosmosDbConfigurationOptionsProxy.java
+++
b/components/camel-azure/camel-azure-cosmosdb/src/main/java/org/apache/camel/component/azure/cosmosdb/CosmosDbConfigurationOptionsProxy.java
@@ -16,12 +16,29 @@
*/
package org.apache.camel.component.azure.cosmosdb;
+import java.io.File;
+import java.io.IOException;
+import java.io.InputStream;
+import java.io.Reader;
+import java.io.UncheckedIOException;
+import java.nio.file.Path;
+import java.util.Collection;
import java.util.Collections;
import java.util.List;
+import java.util.Locale;
+import java.util.Map;
import java.util.function.Supplier;
import com.azure.cosmos.models.*;
+import com.fasterxml.jackson.core.JsonProcessingException;
+import com.fasterxml.jackson.core.type.TypeReference;
+import com.fasterxml.jackson.databind.DeserializationFeature;
+import com.fasterxml.jackson.databind.JsonNode;
+import com.fasterxml.jackson.databind.ObjectMapper;
+import com.fasterxml.jackson.databind.json.JsonMapper;
import org.apache.camel.Exchange;
+import org.apache.camel.StreamCache;
+import org.apache.camel.WrappedFile;
import org.apache.camel.util.ObjectHelper;
/**
@@ -30,6 +47,12 @@ import org.apache.camel.util.ObjectHelper;
*/
public class CosmosDbConfigurationOptionsProxy {
+ // parses a JSON body into the item to write; JSON with more than one
value (such as JSON lines) fails
+ private static final ObjectMapper JSON_MAPPER
+ =
JsonMapper.builder().enable(DeserializationFeature.FAIL_ON_TRAILING_TOKENS).build();
+ private static final TypeReference<Map<String, Object>> ITEM_TYPE = new
TypeReference<>() {
+ };
+
private final CosmosDbConfiguration configuration;
public CosmosDbConfigurationOptionsProxy(final CosmosDbConfiguration
configuration) {
@@ -99,8 +122,85 @@ public class CosmosDbConfigurationOptionsProxy {
String.class));
}
+ /**
+ * The item that createItem, upsertItem and replaceItem write. A Map or a
POJO is returned as it is: the SDK
+ * serializes it with Jackson. JSON as text, bytes, a stream or a file is
parsed into a Map: the SDK would write a
+ * String (or a file, as its path) as a JSON string and cannot serialize a
stream. These operations write one item,
+ * so a List of items, or JSON that is not an object, fails.
+ */
public Object getItem(final Exchange exchange) {
- return exchange.getIn().getBody();
+ final Object body = exchange.getIn().getBody();
+ if (ObjectHelper.isEmpty(body)) {
+ // the operation fails as the item is empty
+ return body;
+ }
+ if (body instanceof Collection || (body.getClass().isArray() && !(body
instanceof byte[]))) {
+ throw new IllegalArgumentException(
+ "The body of the " + getOperation(exchange) + " operation
must be one item, but it is a "
+ + body.getClass().getTypeName()
+ + ": send each item in its own
exchange, for example with the Split EIP");
+ }
+ if (isJson(body)) {
+ return parseItem(exchange, body);
+ }
+ return body;
+ }
+
+ /**
+ * Whether the body is a payload, which holds the JSON of the item: text,
bytes, a stream or a file.
+ */
+ private static boolean isJson(final Object body) {
+ return body instanceof String || body instanceof byte[]
+ || body instanceof InputStream || body instanceof Reader ||
body instanceof StreamCache
+ || body instanceof File || body instanceof Path || body
instanceof WrappedFile;
+ }
+
+ private Map<String, Object> parseItem(final Exchange exchange, final
Object body) {
+ final JsonNode json;
+ try {
+ json = readJson(exchange, body);
+ } catch (JsonProcessingException e) {
+ throw new IllegalArgumentException(
+ "The body of the " + getOperation(exchange)
+ + " operation is not valid
JSON, or holds more than one JSON value",
+ e);
+ } catch (IOException e) {
+ throw new UncheckedIOException(e);
+ }
+ if (json == null || json.isMissingNode()) {
+ // the operation fails as the item is empty
+ return null;
+ }
+ if (json.isArray()) {
+ throw new IllegalArgumentException(
+ "The body of the " + getOperation(exchange) + " operation
must be one item, but it is a JSON array:"
+ + " send each item in its own
exchange, for example with the Split EIP");
+ }
+ if (!json.isObject()) {
+ throw new IllegalArgumentException(
+ "The body of the " + getOperation(exchange) + " operation
must be a JSON object, but it is a JSON "
+ +
json.getNodeType().name().toLowerCase(Locale.ROOT));
+ }
+ return JSON_MAPPER.convertValue(json, ITEM_TYPE);
+ }
+
+ private static JsonNode readJson(final Exchange exchange, final Object
body) throws IOException {
+ if (body instanceof String text) {
+ return JSON_MAPPER.readTree(text);
+ }
+ if (body instanceof byte[] bytes) {
+ return JSON_MAPPER.readTree(bytes);
+ }
+ if (body instanceof Reader reader) {
+ return JSON_MAPPER.readTree(reader);
+ }
+ // a stream or a file: its content, as bytes, from which Jackson
detects the encoding of the JSON
+ final byte[] content = exchange.getIn().getBody(byte[].class);
+ if (content == null) {
+ throw new IllegalArgumentException(
+ "Cannot read the body of type " +
body.getClass().getName() + " as the JSON of the item");
+ }
+ return JSON_MAPPER.readTree(content);
}
public List<Object> getItems(final Exchange exchange) {
diff --git
a/components/camel-azure/camel-azure-cosmosdb/src/main/java/org/apache/camel/component/azure/cosmosdb/CosmosDbEndpoint.java
b/components/camel-azure/camel-azure-cosmosdb/src/main/java/org/apache/camel/component/azure/cosmosdb/CosmosDbEndpoint.java
index 350c78bf6cd8..a68e6e9635d5 100644
---
a/components/camel-azure/camel-azure-cosmosdb/src/main/java/org/apache/camel/component/azure/cosmosdb/CosmosDbEndpoint.java
+++
b/components/camel-azure/camel-azure-cosmosdb/src/main/java/org/apache/camel/component/azure/cosmosdb/CosmosDbEndpoint.java
@@ -47,16 +47,22 @@ import org.apache.camel.util.ObjectHelper;
producerIn = @BodyType(value = Carrier.ANY,
byOption = {
@BodyType.When(option = "operation",
value = "createItem",
- carriers = {
Carrier.MAP, Carrier.POJO, Carrier.BYTES }),
+ carriers = {
+ Carrier.MAP,
Carrier.POJO, Carrier.TEXT, Carrier.BYTES,
+
Carrier.STREAM, Carrier.FILE }),
@BodyType.When(option = "operation",
value = "upsertItem",
- carriers = {
Carrier.MAP, Carrier.POJO, Carrier.BYTES }),
+ carriers = {
+ Carrier.MAP,
Carrier.POJO, Carrier.TEXT, Carrier.BYTES,
+
Carrier.STREAM, Carrier.FILE }),
@BodyType.When(option = "operation",
value = "replaceItem",
- carriers = {
Carrier.MAP, Carrier.POJO, Carrier.BYTES }) },
+ carriers = {
+ Carrier.MAP,
Carrier.POJO, Carrier.TEXT, Carrier.BYTES,
+
Carrier.STREAM, Carrier.FILE }) },
description = "The item for createItem,
upsertItem and replaceItem: a Map or a POJO, which"
- + " Jackson serializes, or the
JSON of the item as bytes. A JSON String fails, as"
- + " it is sent as a JSON
string and not as an object: unmarshal it first. The other"
- + " operations do not use the
body: the item id, the partition key and the query"
- + " are options or headers"),
+ + " Jackson serializes, or the
JSON of the item as text, bytes, a stream or a"
+ + " file, which is parsed into
a Map. These operations write one item: a List,"
+ + " or JSON that is not an
object, fails. The other operations do not use the"
+ + " body: the item id, the
partition key and the query are options or headers"),
producerOut = @BodyType(value = Carrier.LIST,
javaType =
"java.util.List<com.azure.cosmos.models.CosmosDatabaseProperties>",
dependsOn = "contentResponseOnWriteEnabled",
@@ -83,11 +89,11 @@ import org.apache.camel.util.ObjectHelper;
@BodyType.When(option =
"operation", value = "replaceContainerThroughput",
carriers =
Carrier.UNCHANGED),
@BodyType.When(option =
"operation", value = "createItem",
- carriers = {
Carrier.MAP, Carrier.POJO, Carrier.BYTES }),
+ carriers = {
Carrier.MAP, Carrier.POJO }),
@BodyType.When(option =
"operation", value = "upsertItem",
- carriers = {
Carrier.MAP, Carrier.POJO, Carrier.BYTES }),
+ carriers = {
Carrier.MAP, Carrier.POJO }),
@BodyType.When(option =
"operation", value = "replaceItem",
- carriers = {
Carrier.MAP, Carrier.POJO, Carrier.BYTES }),
+ carriers = {
Carrier.MAP, Carrier.POJO }),
@BodyType.When(option =
"operation", value = "readItem", carriers = Carrier.MAP,
javaType =
"java.util.Map<String, Object>"),
@BodyType.When(option =
"operation", value = "readAllItems", carriers = Carrier.LIST,
@@ -97,8 +103,9 @@ import org.apache.camel.util.ObjectHelper;
carriers =
Carrier.NULL) },
description = "A List of the properties of
the SDK for listDatabases, the default, and the"
+ " list and query operations
of databases and containers; the item as written"
- + " for createItem,
upsertItem and replaceItem, of the type of the body sent, or"
- + " no body with
contentResponseOnWriteEnabled=false; the item as a Map for"
+ + " for createItem,
upsertItem and replaceItem, as a Map for a Map or JSON body"
+ + " and of the class of the
POJO for a POJO, or no body with"
+ + "
contentResponseOnWriteEnabled=false; the item as a Map for"
+ " readItem, and a List of
them for readAllItems and queryItems, whose elements"
+ " are values for a SELECT
VALUE query. The operations on databases, containers"
+ " and throughput leave the
body and set the response in headers"))
diff --git
a/components/camel-azure/camel-azure-cosmosdb/src/test/java/org/apache/camel/component/azure/cosmosdb/CosmosDbProducerItemBodyTest.java
b/components/camel-azure/camel-azure-cosmosdb/src/test/java/org/apache/camel/component/azure/cosmosdb/CosmosDbProducerItemBodyTest.java
new file mode 100644
index 000000000000..06c94c8f9cb9
--- /dev/null
+++
b/components/camel-azure/camel-azure-cosmosdb/src/test/java/org/apache/camel/component/azure/cosmosdb/CosmosDbProducerItemBodyTest.java
@@ -0,0 +1,278 @@
+/*
+ * 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.azure.cosmosdb;
+
+import java.io.ByteArrayInputStream;
+import java.io.StringReader;
+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 com.azure.cosmos.CosmosAsyncClient;
+import com.azure.cosmos.CosmosAsyncContainer;
+import com.azure.cosmos.CosmosAsyncDatabase;
+import com.azure.cosmos.CosmosItemSerializer;
+import com.azure.cosmos.models.CosmosItemResponse;
+import com.azure.cosmos.models.PartitionKey;
+import org.apache.camel.BindToRegistry;
+import org.apache.camel.Exchange;
+import org.apache.camel.builder.RouteBuilder;
+import org.apache.camel.component.mock.MockEndpoint;
+import org.apache.camel.test.junit6.CamelTestSupport;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.io.TempDir;
+import org.junit.jupiter.params.ParameterizedTest;
+import org.junit.jupiter.params.provider.ValueSource;
+import reactor.core.publisher.Mono;
+
+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.assertNull;
+import static org.junit.jupiter.api.Assertions.assertSame;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+import static org.mockito.ArgumentMatchers.any;
+import static org.mockito.ArgumentMatchers.anyString;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.when;
+
+/**
+ * The body of createItem, upsertItem and replaceItem is one item. The SDK
serializes a Map or a POJO with Jackson, but
+ * writes a String (or a file, as its path) as a JSON string and cannot
serialize a stream: the JSON of the item, as
+ * text, bytes, a stream or a file, is parsed into a Map first. A List of
items, or JSON that is not one object, fails
+ * before anything is written.
+ */
+class CosmosDbProducerItemBodyTest extends CamelTestSupport {
+
+ private static final String JSON =
"{\"id\":\"1\",\"pk\":\"a\",\"city\":\"Zürich\",\"price\":9.5,\"tags\":[\"x\",\"y\"]}";
+
+ @BindToRegistry("cosmosClient")
+ private final CosmosAsyncClient client = mock(CosmosAsyncClient.class);
+ private final CosmosAsyncDatabase database =
mock(CosmosAsyncDatabase.class);
+ private final CosmosAsyncContainer container =
mock(CosmosAsyncContainer.class);
+ // the items given to the SDK
+ private final List<Object> written = new CopyOnWriteArrayList<>();
+
+ @Override
+ protected RouteBuilder createRouteBuilder() {
+ return new RouteBuilder() {
+ @Override
+ public void configure() {
+
from("direct:createItem").to("azure-cosmosdb:myDb/items?operation=createItem&itemPartitionKey=a");
+
from("direct:upsertItem").to("azure-cosmosdb:myDb/items?operation=upsertItem&itemPartitionKey=a");
+ from("direct:replaceItem")
+
.to("azure-cosmosdb:myDb/items?operation=replaceItem&itemPartitionKey=a&itemId=1");
+ }
+ };
+ }
+
+ @BeforeEach
+ @SuppressWarnings("unchecked")
+ void mockContainer() {
+ CosmosItemResponse<Object> response = mock(CosmosItemResponse.class);
+ when(response.getStatusCode()).thenReturn(201);
+ when(response.getItem()).thenAnswer(invocation ->
written.get(written.size() - 1));
+
+ when(client.getDatabase("myDb")).thenReturn(database);
+ when(database.getContainer("items")).thenReturn(container);
+ when(container.createItem(any(), any(PartitionKey.class),
any())).thenAnswer(invocation -> {
+ written.add(invocation.getArgument(0));
+ return Mono.just(response);
+ });
+ when(container.upsertItem(any(), any(PartitionKey.class),
any())).thenAnswer(invocation -> {
+ written.add(invocation.getArgument(0));
+ return Mono.just(response);
+ });
+ when(container.replaceItem(any(), anyString(),
any(PartitionKey.class), any())).thenAnswer(invocation -> {
+ written.add(invocation.getArgument(0));
+ return Mono.just(response);
+ });
+ }
+
+ @ParameterizedTest
+ @ValueSource(strings = { "createItem", "upsertItem", "replaceItem" })
+ void jsonTextIsWrittenAsAnObject(String operation) {
+ Exchange exchange = template.send("direct:" + operation, e ->
e.getIn().setBody(JSON));
+
+ assertNull(exchange.getException());
+ Map<?, ?> item = assertInstanceOf(Map.class, writtenItem());
+ assertEquals("1", document(item).get("id"));
+ assertEquals("Zürich", item.get("city"));
+ assertEquals(9.5, item.get("price"));
+ assertEquals(List.of("x", "y"), item.get("tags"));
+ // the body is the item as written
+ assertSame(item, exchange.getMessage().getBody());
+ assertEquals(201,
exchange.getMessage().getHeader(CosmosDbConstants.STATUS_CODE));
+ }
+
+ @Test
+ void jsonBytesAreParsed() {
+ Exchange exchange = template.send("direct:createItem", e ->
e.getIn().setBody(JSON.getBytes(StandardCharsets.UTF_8)));
+
+ assertNull(exchange.getException());
+ Map<?, ?> item = assertInstanceOf(Map.class, writtenItem());
+ assertEquals("1", item.get("id"));
+ assertEquals("Zürich", item.get("city"));
+ }
+
+ @Test
+ void streamIsParsed() {
+ Exchange exchange = template.send("direct:createItem",
+ e -> e.getIn().setBody(new
ByteArrayInputStream(JSON.getBytes(StandardCharsets.UTF_8))));
+
+ assertNull(exchange.getException());
+ Map<?, ?> item = assertInstanceOf(Map.class, writtenItem());
+ assertEquals("1", document(item).get("id"));
+ assertEquals("Zürich", item.get("city"));
+ }
+
+ @Test
+ void readerIsParsed() {
+ Exchange exchange = template.send("direct:upsertItem", e ->
e.getIn().setBody(new StringReader(JSON)));
+
+ assertNull(exchange.getException());
+ Map<?, ?> item = assertInstanceOf(Map.class, writtenItem());
+ assertEquals("1", document(item).get("id"));
+ }
+
+ @Test
+ void fileAndPathAreParsed(@TempDir Path dir) throws Exception {
+ Path file = Files.writeString(dir.resolve("item.json"), JSON);
+
+ Exchange exchange = template.send("direct:createItem", e ->
e.getIn().setBody(file.toFile()));
+ assertNull(exchange.getException());
+ Map<?, ?> item = assertInstanceOf(Map.class, writtenItem());
+ assertEquals("1", document(item).get("id"));
+ assertEquals("Zürich", item.get("city"));
+
+ exchange = template.send("direct:replaceItem", e ->
e.getIn().setBody(file));
+ assertNull(exchange.getException());
+ item = assertInstanceOf(Map.class, writtenItem());
+ assertEquals("1", document(item).get("id"));
+ }
+
+ @Test
+ void fileOfTheFileConsumerIsParsed(@TempDir Path dir) throws Exception {
+ Files.writeString(dir.resolve("item.json"), JSON);
+ MockEndpoint mock = getMockEndpoint("mock:written");
+ mock.expectedMessageCount(1);
+
+ context.addRoutes(new RouteBuilder() {
+ @Override
+ public void configure() {
+ from("file:" + dir.toAbsolutePath() +
"?noop=true&initialDelay=0&delay=10")
+
.to("azure-cosmosdb:myDb/items?operation=createItem&itemPartitionKey=a")
+ .to("mock:written");
+ }
+ });
+
+ mock.assertIsSatisfied();
+ Map<?, ?> item = assertInstanceOf(Map.class, writtenItem());
+ assertEquals("1", document(item).get("id"));
+ assertEquals("Zürich", item.get("city"));
+ }
+
+ @Test
+ void mapAndPojoAreWrittenAsTheyAre() {
+ Map<String, Object> map = Map.of("id", "1", "pk", "a");
+ template.sendBody("direct:createItem", map);
+ assertSame(map, writtenItem());
+
+ Item pojo = new Item("2", "a");
+ template.sendBody("direct:upsertItem", pojo);
+ assertSame(pojo, writtenItem());
+ }
+
+ @ParameterizedTest
+ @ValueSource(strings = { "createItem", "upsertItem", "replaceItem" })
+ void listOfItemsFails(String operation) {
+ Exchange exchange = template.send("direct:" + operation,
+ e -> e.getIn().setBody(List.of(Map.of("id", "1", "pk", "a"),
Map.of("id", "2", "pk", "a"))));
+
+ IllegalArgumentException error =
assertInstanceOf(IllegalArgumentException.class, exchange.getException());
+ assertTrue(error.getMessage().contains(operation + " operation must be
one item"), error.getMessage());
+ assertTrue(written.isEmpty());
+ }
+
+ @Test
+ void jsonArrayFails() {
+ Exchange exchange = template.send("direct:createItem", e ->
e.getIn().setBody("[" + JSON + "," + JSON + "]"));
+
+ IllegalArgumentException error =
assertInstanceOf(IllegalArgumentException.class, exchange.getException());
+ assertTrue(error.getMessage().contains("must be one item, but it is a
JSON array"), error.getMessage());
+ assertTrue(written.isEmpty());
+ }
+
+ @Test
+ void jsonThatIsNotOneObjectFails() {
+ Exchange exchange = template.send("direct:createItem", e ->
e.getIn().setBody("\"some text\""));
+ IllegalArgumentException error =
assertInstanceOf(IllegalArgumentException.class, exchange.getException());
+ assertTrue(error.getMessage().contains("must be a JSON object, but it
is a JSON string"), error.getMessage());
+
+ // two items as JSON lines: only the first would be written
+ exchange = template.send("direct:createItem", e ->
e.getIn().setBody(JSON + "\n" + JSON));
+ error = assertInstanceOf(IllegalArgumentException.class,
exchange.getException());
+ assertTrue(error.getMessage().contains("more than one JSON value"),
error.getMessage());
+
+ assertTrue(written.isEmpty());
+ }
+
+ @Test
+ void contentThatIsNotJsonFails() {
+ Exchange exchange = template.send("direct:createItem", e ->
e.getIn().setBody("id=1"));
+ IllegalArgumentException error =
assertInstanceOf(IllegalArgumentException.class, exchange.getException());
+ assertTrue(error.getMessage().contains("createItem operation is not
valid JSON"), error.getMessage());
+
+ // 0xff is not UTF-8
+ byte[] bytes = { '{', '"', 'i', 'd', '"', ':', '"', (byte) 0xff, '"',
'}' };
+ exchange = template.send("direct:createItem", e ->
e.getIn().setBody(bytes));
+ error = assertInstanceOf(IllegalArgumentException.class,
exchange.getException());
+ assertTrue(error.getMessage().contains("createItem operation is not
valid JSON"), error.getMessage());
+
+ assertTrue(written.isEmpty());
+ }
+
+ @Test
+ void noBodyOrAnEmptyStreamFails() {
+ Exchange exchange = template.send("direct:createItem", e ->
e.getIn().setBody(null));
+ IllegalArgumentException error =
assertInstanceOf(IllegalArgumentException.class, exchange.getException());
+ assertEquals("item cannot be empty!", error.getMessage());
+
+ exchange = template.send("direct:createItem", e ->
e.getIn().setBody(new ByteArrayInputStream(new byte[0])));
+ error = assertInstanceOf(IllegalArgumentException.class,
exchange.getException());
+ assertEquals("item cannot be empty!", error.getMessage());
+
+ assertTrue(written.isEmpty());
+ }
+
+ private Object writtenItem() {
+ assertFalse(written.isEmpty(), "no item was written");
+ return written.get(written.size() - 1);
+ }
+
+ // the fields of the document the SDK writes for the item, as its default
serializer gives them
+ private static Map<String, Object> document(Object item) {
+ return CosmosItemSerializer.DEFAULT_SERIALIZER.serialize(item);
+ }
+
+ record Item(String id, String pk) {
+ }
+}
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 f94f49502e3e..10f5a47c1b89 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
@@ -3635,6 +3635,18 @@ and from then on the other endpoints, and the
application, failed every request
Camel now closes only the clients that its endpoints create. Close a client
you set yourself when the application no
longer needs it (a client that is a Spring bean is closed by Spring).
+=== camel-azure-cosmosdb - the JSON of an item is parsed into a Map
+
+The `createItem`, `upsertItem` and `replaceItem` operations now parse a
message body that holds the JSON of the item
+(a `String`, a `byte[]`, a stream or a file) into a `Map` before they write
it. Before, a JSON `String` or a file was
+written as a JSON string and failed, and a stream could not be serialized. A
`byte[]` body was sent as it is, and the
+message body after the operation was the written item as a `byte[]`: it is now
a `Map`, as for a `Map` body. If the
+next steps need the JSON as bytes, marshal the body, for example with
`marshal().json()`.
+
+A `List` body, or JSON that is not one object (such as a JSON array), now
fails with an `IllegalArgumentException`
+before Cosmos DB is called; before, Cosmos DB rejected it with an error 400.
These operations write one item: send each
+item in its own exchange, for example with the Split EIP.
+
=== camel-telegram - a failed send fails the exchange once
When Telegram answers a message with an error status (for example `400` for an
unknown chat), or with a body that
