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?

system · January 31, 2019, 11:22am

This topic was automatically closed 28 days after the last reply. New replies are no longer allowed.

Topic		Replies	Views
[6.1] BulkProcessor and HighLevelRestClient Elasticsearch	3	925	February 6, 2018
[BUG] You can use BulkProcessor with high-level client in 5.6 version Elasticsearch	9	1332	April 11, 2018
Will ES 6.3.0 BulkProcessor class be updated to use HLRC? Elasticsearch	3	370	July 26, 2018
How to use BulkProcessor in Version 8.x.x? Elasticsearch	3	1230	August 25, 2022
BulkProcessor creation using RestHighLevelClient compilation error on ES5.6 Elasticsearch	7	3157	August 16, 2018

Elasticsearch 6.5.4: Initialising BulkProcessor

Related topics