[BUG] The BufferAccumulator class can infinitely backoff and retry and will not throw exception on shutdown

[graytaylor0]

Describe the bug
The BufferAccumulator class does backoff and retry to write and flush to the buffer. When the buffer is full or circuit breaker is open, the buffer will reject the writes and the buffer accumulator will backoff and retry infinitely.

However, when there is a shutdown command to Data Prepper, the BufferAccumulator will not stop backing off and retrying, which will lead to the threads using the BufferAccumulator to block and not exit, delaying the shutdown of the source (potentially for a long time if the buffer never empties or circuit breaker never recovers)

When circuit breaker opens, for example, this exception will get thrown (

data-prepper/data-prepper-core/src/main/java/org/opensearch/dataprepper/core/parser/CircuitBreakingBuffer.java

Line 62 in ca1552e

throw new TimeoutException("Circuit breaker is open. Unable to write to buffer.");

) to the BufferAccumulator and will be caught here (

data-prepper/data-prepper-plugins/buffer-common/src/main/java/org/opensearch/dataprepper/buffer/common/BufferAccumulator.java

Line 90 in ca1552e

LOG.debug("Timed out retrying buffer accumulator");

) and then will retry infinitely

To Reproduce
Steps to reproduce the behavior:

1. Configure a low circuit breaker threshold in the data-prepper-config.yaml
2. Start a pipeline that uses buffer accumulator
3. Make sure circuit breaker is open.
4. Call shutdown and see that Data Prepper does not exit until the source forces shutdown on the threads (if the source does so)

Expected behavior
A clear and concise description of what you expected to happen.

Screenshots
If applicable, add screenshots to help explain your problem.

Environment (please complete the following information):

• OS: [e.g. Ubuntu 20.04 LTS]
• Version [e.g. 22]

Additional context
Add any other context about the problem here.