OpenSearch/docs/java-rest/high-level/document/bulk.asciidoc

--
:api: bulk
:request: BulkRequest
:response: BulkResponse
--

[id="{upid}-{api}"]
=== Bulk API

NOTE: The Java High Level REST Client provides the
<<{upid}-{api}-processor>> to assist with bulk requests.

[id="{upid}-{api}-request"]
==== Bulk Request

A +{request}+ can be used to execute multiple index, update and/or delete
operations using a single request.

It requires at least one operation to be added to the Bulk request:

["source","java",subs="attributes,callouts,macros"]
--------------------------------------------------
include-tagged::{doc-tests-file}[{api}-request]
--------------------------------------------------
<1> Creates the +{request}+
<2> Adds a first `IndexRequest` to the Bulk request. See <<{upid}-index>> for
more information on how to build `IndexRequest`.
<3> Adds a second `IndexRequest`
<4> Adds a third `IndexRequest`

WARNING: The Bulk API supports only documents encoded in JSON or SMILE.
Providing documents in any other format will result in an error.

And different operation types can be added to the same +{request}+:

["source","java",subs="attributes,callouts,macros"]
--------------------------------------------------
include-tagged::{doc-tests-file}[{api}-request-with-mixed-operations]
--------------------------------------------------
<1> Adds a `DeleteRequest` to the +{request}+. See <<{upid}-delete>>
for more information on how to build `DeleteRequest`.
<2> Adds an `UpdateRequest` to the +{request}+. See <<{upid}-update>>
for more information on how to build `UpdateRequest`.
<3> Adds an `IndexRequest` using the SMILE format

==== Optional arguments
The following arguments can optionally be provided:

["source","java",subs="attributes,callouts,macros"]
--------------------------------------------------
include-tagged::{doc-tests-file}[{api}-request-timeout]
--------------------------------------------------
<1> Timeout to wait for the bulk request to be performed as a `TimeValue`
<2> Timeout to wait for the bulk request to be performed as a `String`

["source","java",subs="attributes,callouts,macros"]
--------------------------------------------------
include-tagged::{doc-tests-file}[{api}-request-refresh]
--------------------------------------------------
<1> Refresh policy as a `WriteRequest.RefreshPolicy` instance
<2> Refresh policy as a `String`

["source","java",subs="attributes,callouts,macros"]
--------------------------------------------------
include-tagged::{doc-tests-file}[{api}-request-active-shards]
--------------------------------------------------
<1> Sets the number of shard copies that must be active before proceeding with
the index/update/delete operations.
<2> Number of shard copies provided as a `ActiveShardCount`: can be
`ActiveShardCount.ALL`, `ActiveShardCount.ONE` or
`ActiveShardCount.DEFAULT` (default)

["source","java",subs="attributes,callouts,macros"]
--------------------------------------------------
include-tagged::{doc-tests-file}[{api}-request-pipeline]
--------------------------------------------------
<1> Global pipelineId used on all sub requests, unless overridden on a sub request

["source","java",subs="attributes,callouts,macros"]
--------------------------------------------------
include-tagged::{doc-tests-file}[{api}-request-routing]
--------------------------------------------------
<1> Global routingId used on all sub requests, unless overridden on a sub request

["source","java",subs="attributes,callouts,macros"]
--------------------------------------------------
include-tagged::{doc-tests-file}[{api}-request-index-type]
--------------------------------------------------
<1> A bulk request with global index and type used on all sub requests, unless overridden on a sub request.
Both parameters are @Nullable and can only be set during +{request}+ creation.

include::../execution.asciidoc[]

[id="{upid}-{api}-response"]
==== Bulk Response

The returned +{response}+ contains information about the executed operations and
 allows to iterate over each result as follows:

["source","java",subs="attributes,callouts,macros"]
--------------------------------------------------
include-tagged::{doc-tests-file}[{api}-response]
--------------------------------------------------
<1> Iterate over the results of all operations
<2> Retrieve the response of the operation (successful or not), can be
`IndexResponse`, `UpdateResponse` or `DeleteResponse` which can all be seen as
`DocWriteResponse` instances
<3> Handle the response of an index operation
<4> Handle the response of a update operation
<5> Handle the response of a delete operation

The Bulk response provides a method to quickly check if one or more operation
has failed:
["source","java",subs="attributes,callouts,macros"]
--------------------------------------------------
include-tagged::{doc-tests-file}[{api}-has-failures]
--------------------------------------------------
<1> This method returns `true` if at least one operation failed

In such situation it is necessary to iterate over all operation results in order
to check if the operation failed, and if so, retrieve the corresponding failure:
["source","java",subs="attributes,callouts,macros"]
--------------------------------------------------
include-tagged::{doc-tests-file}[{api}-errors]
--------------------------------------------------
<1> Indicate if a given operation failed
<2> Retrieve the failure of the failed operation

[id="{upid}-{api}-processor"]
==== Bulk Processor

The `BulkProcessor` simplifies the usage of the Bulk API by providing
a utility class that allows index/update/delete operations to be
transparently executed as they are added to the processor.

In order to execute the requests, the `BulkProcessor` requires the following
components:

`RestHighLevelClient`:: This client is used to execute the +{request}+
and to retrieve the `BulkResponse`
`BulkProcessor.Listener`:: This listener is called before and after
every +{request}+ execution or when a +{request}+ failed

Then the `BulkProcessor.builder` method can be used to build a new
`BulkProcessor`:
["source","java",subs="attributes,callouts,macros"]
--------------------------------------------------
include-tagged::{doc-tests-file}[{api}-processor-init]
--------------------------------------------------
<1> Create the `BulkProcessor.Listener`
<2> This method is called before each execution of a +{request}+
<3> This method is called after each execution of a +{request}+
<4> This method is called when a +{request}+ failed
<5> Create the `BulkProcessor` by calling the `build()` method from
the `BulkProcessor.Builder`. The `RestHighLevelClient.bulkAsync()`
method will be used to execute the +{request}+ under the hood.

The `BulkProcessor.Builder` provides methods to configure how the
`BulkProcessor` should handle requests execution:
["source","java",subs="attributes,callouts,macros"]
--------------------------------------------------
include-tagged::{doc-tests-file}[{api}-processor-options]
--------------------------------------------------
<1> Set when to flush a new bulk request based on the number of
actions currently added (defaults to 1000, use -1 to disable it)
<2> Set when to flush a new bulk request based on the size of
actions currently added (defaults to 5Mb, use -1 to disable it)
<3> Set the number of concurrent requests allowed to be executed
(default to 1, use 0 to only allow the execution of a single request)
<4> Set a flush interval flushing any +{request}+ pending if the
interval passes (defaults to not set)
<5> Set a constant back off policy that initially waits for 1 second
and retries up to 3 times. See `BackoffPolicy.noBackoff()`,
`BackoffPolicy.constantBackoff()` and `BackoffPolicy.exponentialBackoff()`
for more options.

Once the `BulkProcessor` is created requests can be added to it:
["source","java",subs="attributes,callouts,macros"]
--------------------------------------------------
include-tagged::{doc-tests-file}[{api}-processor-add]
--------------------------------------------------

The requests will be executed by the `BulkProcessor`, which takes care of
calling the `BulkProcessor.Listener` for every bulk request.

The listener provides methods to access to the +{request}+ and the +{response}+:
["source","java",subs="attributes,callouts,macros"]
--------------------------------------------------
include-tagged::{doc-tests-file}[{api}-processor-listener]
--------------------------------------------------
<1> Called before each execution of a +{request}+, this method allows to know
the number of operations that are going to be executed within the +{request}+
<2> Called after each execution of a +{request}+, this method allows to know if
the +{response}+ contains errors
<3> Called if the +{request}+ failed, this method allows to know
the failure

Once all requests have been added to the `BulkProcessor`, its instance needs to
be closed using one of the two available closing methods.

The `awaitClose()` method can be used to wait until all requests have been
processed or the specified waiting time elapses:
["source","java",subs="attributes,callouts,macros"]
--------------------------------------------------
include-tagged::{doc-tests-file}[{api}-processor-await]
--------------------------------------------------
<1> The method returns `true` if all bulk requests completed and `false` if the
waiting time elapsed before all the bulk requests completed

The `close()` method can be used to immediately close the `BulkProcessor`:
["source","java",subs="attributes,callouts,macros"]
--------------------------------------------------
include-tagged::{doc-tests-file}[{api}-processor-close]
--------------------------------------------------

Both methods flush the requests added to the processor before closing the
processor and also forbid any new request to be added to it.