OAK-11545 - Use a single long-lived bulk processor for Elastic reindexing

[nfsantos]

When writing to Elastic indexes, we use a BulkIngester (part of the Elastic Client Java API) to buffer operations before sending them to Elastic in a bulk request and to manage concurrent outgoing requests. This is essential to have good performance. In the current implementation, each instance of ElasticIndexWriter creates a new ElasticBulkProcessorHandler. Since index writers are created per-index, when indexing more than one Elastic index we are creating several BulkIngesters. This is wasteful because a BulkIngester can be used to write to several indexes, so a single instance can be shared by ElasticIndexWriters.

This PR moves the creation of the ElasticBulkProcessor into the ElasticIndexerProvider class so this single instance can be shared by all the index writers created by that indexer provider. This has the following advantages:

• Decreases the memory usage. Each bulk ingester contains buffers to keep the operations before sending them to Elastic. These buffers can take up significant space, especially if they are increased to optimize the size of the bulk requests.
• Potentially improve performance because the single BulkIngester would fill up faster with operations from several indexes, leading to fewer but bigger bulk requests.
• In the incremental indexer, eliminates the cost of creating and destroying several BulkIngesters for every incremental cycle (every 5 seconds).

Technical details

Bulk ingester configuration is now global

The following properties used to configure the Bulk Ingester could be set per-index, in the index definition. Now they are global configuration properties:

Bulk Ingester property	New system property	Default
maxOperations	oak.indexer.elastic.bulkProcessor.maxBulkOperations	8192
maxSize	oak.indexer.elastic.bulkProcessor.maxBulkSizeBytes	8MB
flushInterval	oak.indexer.elastic.bulkProcessor.bulkFlushIntervalMs	2000 (2s)
maxConcurrentRequests	oak.indexer.elastic.bulkProcessor.maxConcurrentRequests	1
failOnError	oak.indexer.elastic.bulkProcessor.failOnError	true

Changes in default values

The current defaults for the maximum number of operations per bulk and maximum bulk size are too small for good performance. This PR increases them to the following values:

• oak.indexer.elastic.bulkProcessor.maxBulkOperations: 250 -> 8192
• oak.indexer.elastic.bulkProcessor.maxBulkSizeBytes: 1MB -> 8MB

Closing writers without closing bulk ingester

The main complexity of this PR is to manage closing a index writer without closing the underlying bulk ingester. Until now, since each index writer contained a bulk ingester, closing the index writer and bulk ingester had the same lifecycle, so they could be closed together. But to use a single long lived bulk ingester for several writers, closing a index writer becomes more complex.

When closing a writer, we want to flush all the operations for that index. However, the bulk ingester will have in its buffers operations for several indexes and cannot easily distinguish between operations of each index. The solution used is to force a flush of the bulk ingester whenever a index writer is closed, and then wait until all bulk requests lower or equal to the one created by the flush request are processed. This will also flush operations for other indexes that are not being closed, but this will only slow down a little the time to close an index, and has not other problems.

[fabriziofortino]

I think this method name is a bit misleading. No indexes are actually closed here. Should we call something like flushIndex?

[fabriziofortino]

I guess this is called at every async lane run. Should we lower this to debug/trace?

[fabriziofortino]

response.took() returns milliseconds. Dividing it by 1M would result in 0 most of the times.