[Discussion] Concurrency Mechanism in the Elastic BulkProcessor

[wu-sheng]

BulkProcessor has a parameter(concurrentRequests=2, elasticsearch/concurrentRequests) to control the concurrency(from codes) through a Semaphore.
Meanwhile, PersistenceTimer uses ExecutorService(pool size = 2, core/prepareThreads) to run prepare and essentially do batchDAO.flush(innerPrepareRequests). In the flush, it uses a singleton BulkProcessor instance, which causes BulkProcessor#internalAdd to be called by two threads concurrently.

When I reviewed BulkProcessor#internalAdd, I noticed flushIfNeeded is only checking the size of the queue, without any lock mechanism. This means when this queue size reaches the threshold, two threads(from PersistenceTimer) are going to pass this check. Then until one requests.drainTo(batch);, the other is always getting an empty queue, and returns directly. **Note, at this point, the two threads are held, so, no extra thread to put the data into the queue.

If I read the codes correctly, there is no chance we could run concurrent requests to the ElasticSearch search. We have to leave another thread B to return the loop and wait for the next chance.

I know that, these two threads are not going to run synchronously. And the reality could be much better than the above description.
@kezhenxu94 My question is majorly about, is this our design of Concurrency Mechanism of the BulkProcessor?.
If YES, is this also meaning, PersistenceTimer's thread pool could be blocked by BulkProcessor if the size of PersistenceTimer's pool is larger than the size of the BulkProcessor's concurrentRequests setting.

FYI @wankai123 @hanahmily

[kezhenxu94]

When I reviewed BulkProcessor#internalAdd, I noticed flushIfNeeded is only checking the size of the queue, without any lock mechanism. This means when this queue size reaches the threshold, two threads(from PersistenceTimer) are going to pass this check. Then until one requests.drainTo(batch);, the other is always getting an empty queue, and returns directly. **Note, at this point, the two threads are held, so, no extra thread to put the data into the queue.
If I read the codes correctly, there is no chance we could run concurrent requests to the ElasticSearch search. We have to leave another thread B to return the loop and wait for the next chance.

Here is my understanding, 2 threads would pass the size check but only one (say T1) would actually send requests to ES, and the other one (say T2) would return directly and continue to fill the queue, and if the traffic load is high enough, the T2 would fill up the queue to the size of bulkActions again before T1 finishes the flush, at this time T2 would send requests to ES.

I know that, these two threads are not going to run synchronously. And the reality could be much better than the above description. @kezhenxu94 My question is majorly about, is this our design of Concurrency Mechanism of the BulkProcessor?. If YES, is this also meaning, PersistenceTimer's thread pool could be blocked by BulkProcessor if the size of PersistenceTimer's pool is larger than the size of the BulkProcessor's concurrentRequests setting.

Right, PersistenceTimer's thread pool would be blocked if the BulkProcessor cannot process the requests quick enough, to me, this is a producer-consumer paradigm with fixed queue size, if the queue is full and consumer doesn't consume the queue, the producer would block and wait.

[wu-sheng]

Thanks. This makes it clear.

For others reading this thread, I want to add more context for BulkProcessor.
BulkProcessor is a singleton, not just for metrics, but as well as for records. PersistenceTimer on the other hand is only for metrics. So, in the product environment, the elasticsearch/concurrentRequests could be a little larger than core/prepareThreads, which still makes sense, if the trace throughput is high.