Simplify BulkProcessor handling and retry logic

[Tim-Brooks]

This commit collapses the SyncBulkRequestHandler and
AsyncBulkRequestHandler into a single BulkRequestHandler. The new
handler executes a bulk request and awaits for the completion if the
BulkProcessor was configured with a concurrentRequests setting of 0.
Otherwise the execution happens asynchronously.

As part of this change the Retry class has been refactored.
withSyncBackoff and withAsyncBackoff have been replaced with two
versions of withBackoff. One method takes a listener that will be
called on completion. The other method returns a future that will been
complete on request completion.

[Tim-Brooks]

The PR maintains the existing BulkProcessor features. There is this setting concurrentRequests. This impacts the number of concurrent streams that are allowed by the BulkRequestHandler. If it is set to 0, then an "execute" action is blocked until the action is complete.

This setting requires the handler to still be kind of complicated. But I assume we want all the current features of the BulkProcessor to stay the same for backwards compatibility?

[s1monw]

This setting requires the handler to still be kind of complicated. But I assume we want all the current features of the BulkProcessor to stay the same for backwards compatibility?

my take on this is, if we can fail hard if it's used we can drop it. What would be the way to block for the user instead?

[s1monw]

I like this a lot! I left some suggestions can also be followups (the listener)

[s1monw]

that's just a suggestion, I tend to do thsi this way:

Runnable toRelease = () -> {};
//...
semaphore.acquire();
toRelease = semaphore:release

that way you don't need to check any boolean logic and can just call the runnable

[s1monw]

we do have a LatchedActionListener but I see that you need to also call the release method. I wonder if we can generalize LatchedActionListener into something that only takes a Runnable such that users that want to use it with a latch can just pass a method handle for latch::release we could overload ActionListener::wrap to take a runnable as a third parameter and be done with it?

[s1monw]

does this need to be public and also does this class need to be subclassable

[Tim-Brooks]

What would be the way to block for the user instead?

I just kind of feel weird about the fact that "we will block until completion" functionality is determined by the "concurrentStreams" config being set to 0. Obviously the API could return a future for the user to use to block. Or there could be a blockingFlush() api. But since a flush can happen implicitly (add() when a threshold is crossed) or explicitly (flush()) this is not super easy to do. So I'm not sure that there is an obvious change.