class StoreData {  
  private final RestHighLevelClient client;

  public seed() {
    try (var bulkProcessor = BulkProcessor
            .builder(client::bulkAsync, new BulkListener())
            .setBulkActions(500)
            .setConcurrentRequests(0)
            .setGlobalIndex("myIndex")
            .build()) {
            productDao.listIndexable(product -> {
                var indexProduct = new IndexRequest()
                    .id(product.getId().toString())
                    .source(elasticMapper.toIndexableProductViewJson(product), XContentType.JSON);
                bulkProcessor.add(indexProduct);
            });
        }
   }

  public void bulkAsync(BulkRequest request, ActionListener<BulkResponse> bulkListener) {
        client.bulkAsync(request, RequestOptions.DEFAULT, bulkListener);
    }

}

public class BulkListener implements BulkProcessor.Listener {

        @Override
        public void beforeBulk(long executionId, BulkRequest request) {
            var numberOfActions = request.numberOfActions();
            LOG.trace("Executing bulk [{}] with {} requests", executionId, numberOfActions);
        }

        @Override
        public void afterBulk(long executionId, BulkRequest request, BulkResponse response) {
            LOG.trace("Bulk [{}] completed in {} milliseconds", executionId, response.getTook().getMillis());
        }

        @Override
        public void afterBulk(long executionId, BulkRequest request, Throwable failure) {
            LOG.error("Failed to execute bulk", failure);
        }
    }