= Azure CosmosDB Component
:doctitle: Azure CosmosDB
:shortname: azure-cosmosdb
:artifactid: camel-azure-cosmosdb
:description: To read and write records to the CosmosDB database on Azure cloud platform.
:since: 3.10
:supportlevel: Stable
:component-header: Both producer and consumer are supported
//Manually maintained attributes
:group: Azure
:camel-spring-boot-name: azure-cosmosdb

*Since Camel {since}*

*{component-header}*


https://azure.microsoft.com/en-us/services/cosmos-db/[Azure Cosmos DB] is Microsoft's globally distributed, multi-model database service for operational and analytics
workloads. It offers multi-mastering feature by automatically scaling throughput, compute, and storage. This
component interacts with Azure CosmosDB through Azure SQL API.

Prerequisites

You must have a valid Windows Azure Storage account. More information is available at
https://docs.microsoft.com/azure/[Azure Documentation Portal].


[source,xml]
------------------------------------------------------------
<dependency>
    <groupId>org.apache.camel</groupId>
    <artifactId>camel-azure-cosmosdb</artifactId>
    <version>x.x.x</version>
    <!-- use the same version as your Camel core version -->
</dependency>
------------------------------------------------------------


== URI Format

[source,text]
------------------------------
azure-cosmosdb://[databaseName][/containerName][?options]
------------------------------

In case of the consumer, `databaseName`, `containerName` are required, In case of the producer, it depends on the operation that being
requested, for example if operation is on a database level, e.b: deleteDatabase, only `databaseName` is required, but in case
of operation being requested in container level, e.g: readItem, then `databaseName` and `containerName` are required.

You can append query options to the URI in the following format, `?options=value&option2=value&`...


// component-configure options: START

// component-configure options: END

// component options: START
include::partial$component-configure-options.adoc[]
include::partial$component-endpoint-options.adoc[]
// component options: END

// endpoint options: START

// endpoint options: END


== Authentication Information

To use this component, you have 2 options in order to provide the required Azure authentication information:

- Provide `accountKey` and `databaseEndpoint` for your Azure CosmosDB account. The account key can
be generated through your CosmosDB Azure portal.
- Provide a https://docs.microsoft.com/en-us/java/api/com.azure.cosmos.cosmosasyncclient?view=azure-java-stable[CosmosAsyncClient] instance which can be
provided into `cosmosAsyncClient`.


== Async Consumer and Producer

This component implements the async Consumer and producer.

This allows camel route to consume and produce events asynchronously without blocking any threads.

== Usage

For example in order to consume records from a specific container in a specific database to a file, use the following snippet:

[source,java]
--------------------------------------------------------------------------------
from("azure-cosmosdb://camelDb/myContainer?accountKey=MyaccountKey&databaseEndpoint=https//myazure.com:443&leaseDatabaseName=myLeaseDB&createLeaseDatabaseIfNotExists=true&createLeaseContainerIfNotExists=true").
to("file://directory");
--------------------------------------------------------------------------------

=== Message headers evaluated by the component producer
[width="100%",cols="10%,10%,10%,70%",options="header",]
|=======================================================================
|Header |Variable Name |Type |Description

|`CamelAzureCosmosDbDatabaseName`| `CosmosDbConstants.DATABASE_NAME`|`String`| Overrides the database name which is the name of the Cosmos database that component should connect to. In case you are producing data and have createDatabaseIfNotExists=true, the component will automatically auto create a Cosmos database.
|`CamelAzureCosmosDbContainerName`| `CosmosDbConstants.CONTAINER_NAME`|`String`| Overrides the container name which is the name of the Cosmos container that component should connect to. In case you are producing data and have createContainerIfNotExists=true, the component will automatically auto create a Cosmos container.
|`CamelAzureCosmosDbOperation`| `CosmosDbConstants.OPERATION`|`CosmosDbOperationsDefinition`| Sets the producer operation which can be used to execute a specific operation on the producer.
|`CamelAzureCosmosDbQuery`| `CosmosDbConstants.QUERY`|`String`| Sets the SQL query to execute on a given producer query operations.
|`CamelAzureCosmosDbQueryRequestOptions`| `CosmosDbConstants.QUERY_REQUEST_OPTIONS`|`CosmosQueryRequestOptions`| Set additional QueryRequestOptions that can be used with queryItems, queryContainers, queryDatabases, listDatabases, listItems, listContainers operations.
|`CamelAzureCosmosDbCreateDatabaseIfNotExist`| `CosmosDbConstants.CREATE_DATABASE_IF_NOT_EXIST`|`boolean`| Sets if the component should create Cosmos database automatically in case it doesn't exist in Cosmos account.
|`CamelAzureCosmosDbCreateContainerIfNotExist`| `CosmosDbConstants.CREATE_CONTAINER_IF_NOT_EXIST`|`boolean`| Sets if the component should create Cosmos container automatically in case it doesn't exist in Cosmos account.
|`CamelAzureCosmosDbThroughputProperties`| `CosmosDbConstants.THROUGHPUT_PROPERTIES`|`ThroughputProperties`| Sets throughput of the resources in the Azure Cosmos DB service.
|`CamelAzureCosmosDbDatabaseRequestOptions`| `CosmosDbConstants.DATABASE_REQUEST_OPTIONS`|`CosmosDatabaseRequestOptions`| Sets additional options to execute on database operations.
|`CamelAzureCosmosDbContainerPartitionKeyPath`| `CosmosDbConstants.CONTAINER_PARTITION_KEY_PATH`|`String`| Sets the container partition key path.
|`CamelAzureCosmosDbContainerRequestOptions`| `CosmosDbConstants.CONTAINER_REQUEST_OPTIONS`|`CosmosContainerRequestOptions`| Sets additional options to execute on container operations.
|`CamelAzureCosmosDbItemPartitionKey`| `CosmosDbConstants.ITEM_PARTITION_KEY`|`PartitionKey`|  Sets partition key. Represents a partition key value in the Azure Cosmos DB database service. A partition key identifies the partition where the item is stored in.
|`CamelAzureCosmosDbItemRequestOptions`| `CosmosDbConstants.ITEM_REQUEST_OPTIONS`|`CosmosItemRequestOptions`|  Sets additional options to execute on item operations.
|`CamelAzureCosmosDbItemId`| `CosmosDbConstants.ITEM_ID`|`String`|  Sets the itemId in case needed for operation on item like delete, replace.
|=======================================================================


=== Message headers set by the component producer
[width="100%",cols="10%,10%,10%,70%",options="header",]
|=======================================================================
|Header |Variable Name |Type |Description

|`CamelAzureCosmosDbRecourseId`| `CosmosDbConstants.RESOURCE_ID`|`String`| The resource ID of the requested resource.
|`CamelAzureCosmosDbEtag`| `CosmosDbConstants.E_TAG`|`String`| The Etag ID of the requested resource.
|`CamelAzureCosmosDbTimestamp`| `CosmosDbConstants.TIMESTAMP`|`String`| The timestamp of the requested resource.
|`CamelAzureCosmosDbResponseHeaders`| `CosmosDbConstants.RESPONSE_HEADERS`|`Map`| The response headers of the requested resource.
|`CamelAzureCosmosDbStatusCode`| `CosmosDbConstants.STATUS_CODE`|`Integer`| The status code of the requested resource.
|`CamelAzureCosmosDbDefaultTimeToLiveInSeconds`| `CosmosDbConstants.DEFAULT_TIME_TO_LIVE_SECONDS`|`Integer`| The TTL of the requested resource.
|`CamelAzureCosmosDbManualThroughput`| `CosmosDbConstants.MANUAL_THROUGHPUT`|`Integer`| The manual throughput of the requested resource.
|`CamelAzureCosmosDbAutoscaleMaxThroughput`| `CosmosDbConstants.AUTOSCALE_MAX_THROUGHPUT`|`Integer`| The AutoscaleMaxThroughput of the requested resource.
|=======================================================================

=== Azure CosmosDB Producer operations

Camel Azure CosmosDB component provides wide range of operations on the producer side:

*Operations on the service level*

For these operations, `databaseName` is *required* except for `queryDatabases` and `listDatabases` operations.
[width="100%",cols="10%,90%",options="header",]
|===
|Operation |Description
|`listDatabases`  | Gets a list of all databases as `List<CosmosDatabaseProperties>` set in the exchange message body.
|`createDatabase`  | Creates a database in the specified Azure CosmosDB account.
|`queryDatabases`  | *`query` is required* Execute an SQL query against the service level in order for example return only a small subset of the databases list. It will set `List<CosmosDatabaseProperties>` set in the exchange message body.
|===


*Operations on the database level*

For these operations, `databaseName` is *required* for all operations here and `containerName` only for `createContainer` and `queryContainers`.
[width="100%",cols="10%,90%",options="header",]
|===
|Operation |Description
|`deleteDatabase`  | Deletes a database from the Azure CosmosDB account.
|`createContainer`  | Creates a container in the specified Azure CosmosDB database.
|`replaceDatabaseThroughput`  | Replaces the throughput for the specified Azure CosmosDB database.
|`listContainers`  | Gets a list of all containers in the specified database as `List<CosmosContainerProperties>` set in the exchange message body.
|`queryContainers`  | *`query` is required* Executes an SQL query against the database level in order for example return only a small subset of the containers list for the specified database. It will set `List<CosmosContainerProperties>` set in the exchange message body.
|===


*Operations on the container level*

For these operations, `databaseName` and `containerName` is *required* for all operations here.
[width="100%",cols="10%,90%",options="header",]
|===
|Operation |Description
|`deleteContainer`  | Deletes a container from the specified Azure CosmosDB database.
|`replaceContainerThroughput`  | Replaces 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, ?>`.
|`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, ?>`.
|`replaceItem`  | *`itemPartitionKey` and `itemId` are required* Overwrites an item in the specified container , it accepts POJO or key value as `Map<String, ?>`.
|`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.
|`readItem`  | *`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.
|===

Refer to the example section in this page to learn how to use these operations into your camel application.

==== Examples
- `listDatabases`:

[source,java]
--------------------------------------------------------------------------------
from("direct:start")
  .to("azure-cosmosdb://?operation=listDatabases")
  .to("mock:result");
--------------------------------------------------------------------------------

- `createDatabase`:

[source,java]
--------------------------------------------------------------------------------
from("direct:start")
    .process(exchange -> {
        exchange.getIn().setHeader(CosmosDbConstants.DATABASE_NAME, "myDb");
      })
  .to("azure-cosmosdb://?operation=createDatabase")
  .to("mock:result");
--------------------------------------------------------------------------------

- `deleteDatabase`:

[source,java]
--------------------------------------------------------------------------------
from("direct:start")
    .process(exchange -> {
        exchange.getIn().setHeader(CosmosDbConstants.DATABASE_NAME, "myDb");
      })
  .to("azure-cosmosdb://?operation=deleteDatabase")
  .to("mock:result");
--------------------------------------------------------------------------------

- `createContainer`:

[source,java]
--------------------------------------------------------------------------------
from("direct:start")
    .process(exchange -> {
        exchange.getIn().setHeader(CosmosDbConstants.DATABASE_NAME, "databaseName");
        exchange.getIn().setHeader(CosmosDbConstants.CONTAINER_NAME, "containerName");
        exchange.getIn().setHeader(CosmosDbConstants.CONTAINER_PARTITION_KEY_PATH, "path");
        exchange.getIn().setHeader(CosmosDbConstants.CREATE_DATABASE_IF_NOT_EXIST, true);
      })
  .to("azure-cosmosdb://?operation=createContainer")
  .to("mock:result");
--------------------------------------------------------------------------------


- `deleteContainer`:

[source,java]
--------------------------------------------------------------------------------
from("direct:start")
    .process(exchange -> {
        exchange.getIn().setHeader(CosmosDbConstants.DATABASE_NAME, "databaseName");
        exchange.getIn().setHeader(CosmosDbConstants.CONTAINER_NAME, "containerName");
      })
  .to("azure-cosmosdb://?operation=deleteContainer")
  .to("mock:result");
--------------------------------------------------------------------------------

- `replaceDatabaseThroughput`:

[source,java]
--------------------------------------------------------------------------------
from("direct:start")
    .process(exchange -> {
        exchange.getIn().setHeader(CosmosDbConstants.DATABASE_NAME, "databaseName");
         exchange.getIn().setHeader(CosmosDbConstants.THROUGHPUT_PROPERTIES,
                    ThroughputProperties.createManualThroughput(700));
      })
  .to("azure-cosmosdb://?operation=replaceDatabaseThroughput")
  .to("mock:result");
--------------------------------------------------------------------------------

- `queryContainers`:

[source,java]
--------------------------------------------------------------------------------
from("direct:start")
    .process(exchange -> {
        exchange.getIn().setHeader(CosmosDbConstants.DATABASE_NAME, "databaseName");
            exchange.getIn().setHeader(CosmosDbConstants.QUERY, "SELECT * from c where c.id = 'myAwersomeContainer'");
      })
  .to("azure-cosmosdb://?operation=queryContainers")
  .to("mock:result");
--------------------------------------------------------------------------------


- `createItem`:

[source,java]
--------------------------------------------------------------------------------
from("direct:start")
    .process(exchange -> {
            // create item to send
            final Map<String, Object> item = new HashMap<>();
            item1.put("id", "test-id-1");
            item1.put("partition", "test-1");
            item1.put("field1", "awesome!");

            exchange.getIn().setHeader(CosmosDbConstants.DATABASE_NAME, "databaseName");
            exchange.getIn().setHeader(CosmosDbConstants.CONTAINER_NAME, "containerName");
            exchange.getIn().setHeader(CosmosDbConstants.CONTAINER_PARTITION_KEY_PATH, "partition");
            exchange.getIn().setHeader(CosmosDbConstants.ITEM_PARTITION_KEY, "test-1");
            exchange.getIn().setBody(item);
      })
  .to("azure-cosmosdb://?operation=createItem")
  .to("mock:result");
--------------------------------------------------------------------------------

- `replaceItem`:

[source,java]
--------------------------------------------------------------------------------
from("direct:start")
    .process(exchange -> {
            // create item to send
            final Map<String, Object> item = new HashMap<>();
            item1.put("id", "test-id-1");
            item1.put("partition", "test-1");
            item1.put("field1", "awesome!");

            exchange.getIn().setHeader(CosmosDbConstants.DATABASE_NAME, "databaseName");
            exchange.getIn().setHeader(CosmosDbConstants.CONTAINER_NAME, "containerName");
            exchange.getIn().setHeader(CosmosDbConstants.ITEM_PARTITION_KEY, "test-1");
            exchange.getIn().setHeader(CosmosDbConstants.ITEM_ID, "test-id-1");
            exchange.getIn().setBody(item);
      })
  .to("azure-cosmosdb://?operation=replaceItem")
  .to("mock:result");
--------------------------------------------------------------------------------


- `deleteItem`:

[source,java]
--------------------------------------------------------------------------------
from("direct:start")
    .process(exchange -> {
            exchange.getIn().setHeader(CosmosDbConstants.DATABASE_NAME, "databaseName");
            exchange.getIn().setHeader(CosmosDbConstants.CONTAINER_NAME, "containerName");
            exchange.getIn().setHeader(CosmosDbConstants.ITEM_PARTITION_KEY, "test-1");
            exchange.getIn().setHeader(CosmosDbConstants.ITEM_ID, "test-id-1");
            exchange.getIn().setBody(item);
      })
  .to("azure-cosmosdb://?operation=deleteItem")
  .to("mock:result");
--------------------------------------------------------------------------------


- `queryItems`:

[source,java]
--------------------------------------------------------------------------------
from("direct:start")
    .process(exchange -> {
            exchange.getIn().setHeader(CosmosDbConstants.DATABASE_NAME, "databaseName");
            exchange.getIn().setHeader(CosmosDbConstants.CONTAINER_NAME, "containerName");
            exchange.getIn().setHeader(CosmosDbConstants.QUERY, "SELECT c.id,c.field2,c.field1 from c where c.id = 'test-id-1'");
      })
  .to("azure-cosmosdb://?operation=queryItems")
  .to("mock:result");
--------------------------------------------------------------------------------


=== Azure CosmosDB Consumer
Camel Azure CosmosDB uses https://docs.microsoft.com/en-us/azure/cosmos-db/change-feed-design-patterns[ChangeFeed pattern] to capture a feed of events and feed them into the Camel in an Async manner, something similar to Change Data Capture (CDC) design pattern.
However, it doesn't capture deletes as these are removed from the feed as well.

To use the Camel Azure CosmosDB, `containerName` and `databaseName` are required. However, there are more options that need to be set in order to use this feature:

- `leaseDatabaseName` : Sets the lease database where the `leaseContainerName` will be stored. If it is not specified, this component will store the lease container in the same database that is specified in databaseName. It will be auto created if `createLeaseDatabaseIfNotExists` is set to true.
- `leaseContainerName` : Sets the lease container which acts as a state storage and coordinates processing the change feed across multiple workers. The lease container can be stored in the same account as the monitored container or in a separate account. It will be auto created if `createLeaseContainerIfNotExists` is set to true. If not specified, this component will create container called `camel-lease`.
- `hostName` : Sets the hostname. The host: a host is an application instance that uses the change feed processor to listen for changes. Multiple instances with the same lease configuration can run in parallel, but each instance should have a different instance name. If not specified, this will be a generated random hostname.
- `changeFeedProcessorOptions` : Sets additional options for the change feed processor.

The consumer will set `List<Map<String,>>` in exchange message body which reflect list of items in a single feed.

==== Example:
For example to listen to the events in `myContainer` container in `myDb`:

[source,java]
--------------------------------------------------------------------------------
from("azure-cosmosdb://myDb/myContainer?leaseDatabaseName=myLeaseDb&createLeaseDatabaseIfNotExists=true&createLeaseContainerIfNotExists=true")
  .to("mock:result");
--------------------------------------------------------------------------------


=== Development Notes (Important)

When developing on this component, you will need to obtain your Azure accessKey in order to run the integration tests. In addition to the mocked unit tests
you *will need to run the integration tests with every change you make or even client upgrade as the Azure client can break things even on minor versions upgrade.*
To run the integration tests, on this component directory, run the following maven command:

[source,bash]
----
mvn clean install -Dendpoint={{dbaddress}} -DaccessKey={{accessKey}}
----
Whereby `endpoint` is your Azure CosmosDB endpoint name and `accessKey` is the access key being generated from Azure CosmosDB portal.



include::spring-boot:partial$starter.adoc[]
