Elasticsearch 6.5.4: Initialising BulkProcessor

LGans316 · January 3, 2019, 10:17am

Trying to upgrade from 6.4.0 to 6.5.4 and noticed many deprecations. Fixed most of them except for one.

This is how the BulkProcessor was initialised in 6.4.0. indexClient is RestHighLevelClient and BulkProcessorListener implements BulkProcessor.Listener. This is all working fine.

			Builder builder = BulkProcessor.builder(indexClient::bulkAsync,
			        new BulkProcessorListener(this.stats, errorLogWriter, appConfig.isDebug()));
			builder.setConcurrentRequests(ElasticSearchConfig.concurrentRequests);
			builder.setBulkActions(indexConfig.getBulkActions());
			builder.setFlushInterval(TimeValue.timeValueSeconds(indexConfig.getBulkFlushInterval()));
			builder.setBackoffPolicy(BackoffPolicy.exponentialBackoff(TimeValue.timeValueMillis(100), 5));
			this.bulkProcessor = builder.build();

Now with 6.5.4**, indexClient::bulkAsync** has been deprecated. Also I am unable to pass RestHighLevelClient as first argument as it doesn't implement org.elasticsearch.client.Client interface. Noticed the TransportClient (which is to be deprecated) indirectly implemented Client interface but we don't use TransportClient in our codebase at all. So wondering how BulkProcessor can be initialised using RestHighLevelClient in 6.5.4.

https://www.elastic.co/guide/en/elasticsearch/client/java-api/6.5/java-docs-bulk-processor.html

Please advise as this is blocking our migration.

dadoonet · January 3, 2019, 10:31am

Here is how I'm creating it:

github.com

dadoonet/legacy-search/blob/02-bulk/src/main/java/fr/pilato/demo/legacysearch/dao/ElasticsearchDao.java#L57-L78


this.bulkProcessor = BulkProcessor.builder(
        (request, bulkListener) -> esClient.bulkAsync(request, RequestOptions.DEFAULT, bulkListener),
        new BulkProcessor.Listener() {
            @Override
            public void beforeBulk(long executionId, BulkRequest request) {
                logger.debug("going to execute bulk of {} requests", request.numberOfActions());
            }


            @Override
            public void afterBulk(long executionId, BulkRequest request, BulkResponse response) {
                logger.debug("bulk executed {} failures", response.hasFailures() ? "with" : "without");
            }


            @Override
            public void afterBulk(long executionId, BulkRequest request, Throwable failure) {
                logger.warn("error while executing bulk", failure);
            }
        })
        .setBulkActions(10000)
        .setFlushInterval(TimeValue.timeValueSeconds(5))

This file has been truncated. show original

HTH

LGans316 · January 3, 2019, 10:38am

Thanks you so much again. Looks like I need to refresh my Java 8 skills

Also are there any plans to create an overloaded method that doesn't accept RequestOptions as I find it a bit annoying to keep passing RequestOptions.DEFAULT

dadoonet · January 3, 2019, 11:22am

I found that a bit annoying too but I'm now used to it.
Anyway, I don't think there is a plan for this. @javanna do you know?

Topic		Replies	Views
[6.1] BulkProcessor and HighLevelRestClient Elasticsearch	3	960	February 6, 2018
Will ES 6.3.0 BulkProcessor class be updated to use HLRC? Elasticsearch	3	407	July 26, 2018
Need an example of BulkProcessor with the 5.6 version of high-level client Elasticsearch	3	511	April 13, 2018
[BUG] You can use BulkProcessor with high-level client in 5.6 version Elasticsearch	9	1403	April 11, 2018
Headache with RestHighLevelClient and bulkprocessor Elasticsearch	3	2947	December 11, 2017

Elasticsearch 6.5.4: Initialising BulkProcessor

Related topics