Bulk processor concurrent requests

[jakelandis]

org.elasticsearch.action.bulk.BulkProcessor is a threadsafe class that
allows for simple semantics to deal with sending bulk requests. Once a
bulk reaches it's pre-defined size, documents, or flush interval it will
execute sending the bulk. One configurable option is the number of concurrent
outstanding bulk requests. That concurrency is implemented in
org.elasticsearch.action.bulk.BulkRequestHandler via a semaphore. However,
the only code that currently calls into this code is blocked by synchronized
methods. This results in the in-ability for the BulkProcessor to behave concurrently
despite supporting configurable amounts of concurrent requests.

This change removes the synchronized method in favor an explicit
lock around the non-thread safe parts of the method. The call into
org.elasticsearch.action.bulk.BulkRequestHandler is no longer blocking, which
allows org.elasticsearch.action.bulk.BulkRequestHandler to handle it's own concurrency.

---

Currently the only production code that uses this class is Watcher
and it is defaulted to 0 concurrency. This change allows Watcher to optionally
increase the concurrency.

I have run this test a few thousand times and can not get it to error. Also, if I remove the lock locally the test will fail pretty quickly.

This is tangentially related to #41418 since it is the same synchronized method that can cause issues. However, this change without that fix only moves where the blocking happens (from synchronized block to semaphore block).

Adding HLRC and Watcher since outside of tests, they are the largest consumers.

[elasticmachine]

Pinging @elastic/es-core-features

[jakelandis]

Hold off reviewing...I think I need to wrap the other instance swapping inside the same lock.

[jakelandis]

@elasticmachine run elasticsearch-ci/bwc
@elasticmachine run elasticsearch-ci/default-distro

[jakelandis]

@martijnvg @hub-cap - this is ready for review. I have re-run the tests a few hundred more times with the code as-is and can't get it to fail. To see this change in action you need to add the following println in BulkRequestHandler and run the concurrency tests.

   public void execute(BulkRequest bulkRequest, long executionId) {
        Runnable toRelease = () -> {};
        boolean bulkRequestSetupSuccessful = false;
        try {
            listener.beforeBulk(executionId, bulkRequest);
            semaphore.acquire();
            System.out.println("permits:" + semaphore.availablePermits() + " of " + concurrentRequests + " queue length: " + semaphore.getQueueLength());

You will notice that before this change, it at most ever acquired 1 permit (due to the upstream synchronous method). With this change it can consume all the permits and add to the queue.

[hub-cap]

LGTM thank you for the detailed commit msg and note about the testing. I had one question but I will defer to those who are more well versed than I in the art of locks in java!

[jaymode]

I left some comments about lock usage and the testing

[jaymode]

I am not sure this is the best way to test this. Ideally you want all of your threads ready to go prior to starting work. Usually we use a cyclic barrier or countdown latch so that we know all threads are ready prior to execution of the concurrent tasks. An example use of a cyclic barrier would be

elasticsearch/server/src/test/java/org/elasticsearch/common/cache/CacheTests.java

Lines 843 to 878 in 0531987

	public void testExceptionThrownDuringConcurrentComputeIfAbsent() throws BrokenBarrierException, InterruptedException {
	int numberOfThreads = randomIntBetween(2, 32);
	final Cache<String, String> cache = CacheBuilder.<String, String>builder().build();
	CyclicBarrier barrier = new CyclicBarrier(1 + numberOfThreads);
	final String key = randomAlphaOfLengthBetween(2, 32);
	for (int i = 0; i < numberOfThreads; i++) {
	Thread thread = new Thread(() -> {
	try {
	barrier.await();
	for (int j = 0; j < numberOfEntries; j++) {
	try {
	String value = cache.computeIfAbsent(key, k -> {
	throw new RuntimeException("failed to load");
	});
	fail("expected exception but got: " + value);
	} catch (ExecutionException e) {
	assertNotNull(e.getCause());
	assertThat(e.getCause(), instanceOf(RuntimeException.class));
	assertEquals(e.getCause().getMessage(), "failed to load");
	}
	}
	barrier.await();
	} catch (BrokenBarrierException | InterruptedException e) {
	throw new AssertionError(e);
	}
	});
	thread.start();
	}
	// wait for all threads to be ready
	barrier.await();
	// wait for all threads to finish
	barrier.await();
	}

.

[jakelandis]

@jaymode - thanks for the pointer. I added countdown latch to ensure that all the tasks are submitted to the executor before doing any work. Let me know if this missed the mark. - a1e456c

[jakelandis]

@jaymode @hub-cap - thanks for your reviews! I believe all comments have been addressed.

[jaymode]

I left some more comments and suggestions. Mainly concerning the handling of InterruptedException, keeping track of all exceptions, and tightening up the start gate latch so we know all items have been added and the executor has all of its threads constructed and has started running the expected number of concurrent tasks.

[bruckner]

Is there any chance this fix will be backported to 6.8? Seems on par with the later #48697 which was backported.