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.