Fix BulkProcessor hangs for threads deadlocked (#44556)

[suxinglee]

In #44556 bulk processor threads gets deadlocked, and Java application using BulkProcessing hangs without any data flush. This problem have been discussed in two issues. One way to solve this deadlock is modify corePoolSize for scheduler in client.

[elasticmachine]

Pinging @elastic/es-core-features

[suxinglee]

@jakelandis PTAL

[jakelandis]

@suxinglee - Thank you very much for finding this bug and providing a test case. After taking a look, it appears this is fixed in 7.3.0 by #41451. I logged #47599 with more details and will close out this PR these related issues since it is addressed and there is a workaround.

You can workaround the deadlock by changing the BackoffPolicy to noBackoff.

From your test:

private Future<?> buildAndExecuteBulkProcessor(Scheduler scheduler) throws InterruptedException {
...
       // final BackoffPolicy backoff = BackoffPolicy.constantBackoff(TimeValue.timeValueMillis(100L), 1);
        final BackoffPolicy backoff = BackoffPolicy.noBackoff();