Is BulkIngester (replacement of 'Bulk Processor') in elasticsearch java api thread safe?

Irfanulla · October 3, 2023, 6:26am

Use case: I have multiple kafka listeners for various topics. Each topic Listener will run in multiple threads (using spring's 'ConcurrentKafkaListenerContainer'). Listeners will be performing Update/Insert operations on Elasticsearch via its Java client (newer low-level client). I realized that performing bulk operations is very performant and hence will be do that. Also found that the client library provides a very good utility called 'BulkIngester'(replacement of 'Bulk Processor' in previous versions) which is exactly what I needed.

Query: Since multiple thread will be adding update/insert operations to 'BulkIngester', I would like to understand if it is thread safe? As its user, do I need to take care of any synchronization?

Also, I did go through the 'BulkIngester' code and functionally understood how it is working. It also has code to perform synchronization but couldn't fully wrap my head around it.

dadoonet · October 3, 2023, 11:37am

Welcome!

Yes. You can use it the same way you were using the BulkProcessor.

Small usage example here:

github.com

dadoonet/elasticsearch-java-client-demo/blob/main/src/test/java/fr/pilato/test/elasticsearch/hlclient/EsClientIT.java#L338-L377


      
          void bulkIngester() throws IOException {
              int size = 1000;
              try {
                  client.indices().delete(dir -> dir.index("bulk"));
              } catch (ElasticsearchException ignored) { }
              BulkIngester<Void> ingester = BulkIngester.of(b -> b
                      .client(client)
                      .listener(new BulkListener<Void>() {
                          @Override
                          public void beforeBulk(long executionId, BulkRequest request, List<Void> voids) {
                              logger.debug("going to execute bulk of {} requests", request.operations().size());
                          }
          
                          @Override
                          public void afterBulk(long executionId, BulkRequest request, List<Void> voids, BulkResponse response) {
                              logger.debug("bulk executed {} errors", response.errors() ? "with" : "without");
                          }
          
                          @Override
                          public void afterBulk(long executionId, BulkRequest request, List<Void> voids, Throwable failure) {

This file has been truncated. show original

AFAICS in the code, it's using an internal synchronisation so it's safe to use the same instance from multiple threads:

github.com

elastic/elasticsearch-java/blob/23973026f0780d62b18d618321c27a24cdb8a4c3/java-client/src/main/java/co/elastic/clients/elasticsearch/_helpers/bulk/BulkIngester.java#L73-L77


      
          // Synchronization objects
          private final ReentrantLock lock = new ReentrantLock();
          private final FnCondition addCondition = new FnCondition(lock, this::canAddOperation);
          private final FnCondition sendRequestCondition = new FnCondition(lock, this::canSendRequest);
          private final FnCondition closeCondition = new FnCondition(lock, this::closedAndFlushed);

Irfanulla · October 4, 2023, 7:12am

Thanks very much David for clarification.

Looks like the Java concurrency - Locks are being used, that are more flexible and sophisticated thread synchronization mechanism.

system · November 1, 2023, 7:12am

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