private void indexDocuments() {
        if (bulkOperations.isEmpty()) {
            return;
        }

        try {
            final AtomicBoolean succeeded = new AtomicBoolean(false);
            while (!succeeded.get()) {
                elasticClientCache.context(elasticCluster.getConnection(), elasticClient -> {
                    try {
                        // Create a new bulk indexing request, containing the current batch of documents
                        final Builder bulkRequestBuilder = new BulkRequest.Builder();

                        // For each document, create an indexing request and append to the bulk request
                        bulkRequestBuilder.operations(bulkOperations);

                        if (refreshAfterEachBatch) {
                            // Refresh upon completion of the batch index request
                            bulkRequestBuilder.refresh(Refresh.True);
                        } else {
                            bulkRequestBuilder.refresh(Refresh.False);
                        }

                        final BulkResponse response = elasticClient.bulk(bulkRequestBuilder.build());
                        if (response.errors()) {
                            boolean overloaded = response.items().stream()
                                    .anyMatch(item -> item.status() == ES_TOO_MANY_REQUESTS_STATUS);
                            if (overloaded) {
                                // Failure was due to Elasticsearch being overloaded, so retry after a delay
                                deleteBulkRequestDocuments(elasticClient, response);
                                throw new ElasticsearchOverloadedException(response.toString());
                            } else {
                                // Request failed for some other reason, so abort without retry
                                throw new IOException("Bulk indexing request failed: " + response);
                            }
                        } else {
                            succeeded.set(true);
                            final String retryMessage = currentRetry > 0
                                    ? " (retries: " + currentRetry + ")"
                                    : "";
                            LOGGER.info("Pipeline '{}' indexed {} documents from stream {} to Elasticsearch cluster " +
                                        "'{}' in {} seconds{}", pipelineName, bulkOperations.size(),
                                    metaHolder.getMeta().getId(), elasticCluster.getName(), response.took() / 1000,
                                    retryMessage);
                        }
                    } catch (ElasticsearchOverloadedException e) {
                        handleElasticsearchOverloadedException(e);
                    } catch (ElasticsearchException e) {
                        if (e.status() == ES_TOO_MANY_REQUESTS_STATUS) {
                            handleElasticsearchOverloadedException(e);
                        } else {
                            handleElasticsearchException(e);
                        }
                    } catch (ResponseException e) {
                        if (e.getResponse().getStatusLine().getStatusCode() == ES_TOO_MANY_REQUESTS_STATUS) {
                            handleElasticsearchOverloadedException(e);
                        } else {
                            handleElasticsearchException(e);
                        }
                    } catch (RuntimeException | IOException e) {
                        fatalError(e.getMessage() != null
                                ? e.getMessage().substring(0,
                                Math.min(ES_MAX_EXCEPTION_CHARS, e.getMessage().length()))
                                : "", e);
                    } finally {
                        currentRetry++;
                    }
                });
            }
        } finally {
            currentRetry = 0;
            bulkOperations.clear();
        }
    }
Index the current batch of documents
stroom-search/stroom-search-elastic/src/main/java/stroom/search/elastic/indexing/ElasticIndexingFilter.java