[BUG] OpenSearch sink is not resuming after failures

Describe the bug

We have noticed that failures writing to the OpenSearch sink may result in the opensearch sink failing to continue to write, even after the issue is resolved.

To Reproduce

1. Create this pipeline

opensearch-retry-pipeline:
  workers: 2
  delay: 100
  source:
    http:

  sink:
    - opensearch:
        hosts: [ "https://opensearch:9200" ]
        insecure: true
        username: admin
        password: admin
        index: test_forbidden

2. Write a message.

curl http://localhost:2021/log/ingest -X POST -H 'Content-Type: application/json' -d '[{"test": "hello"}]'

3. Search - the new document is available

GET test_forbidden/_search
{
  "query": {
    "match_all": {}
  }
}

4. Add a write block

PUT test_forbidden/_block/write?timeout=30m

5. Write more messages

curl http://localhost:2021/log/ingest -X POST -H 'Content-Type: application/json' -d '[{"test": "hello2"}]'
curl http://localhost:2021/log/ingest -X POST -H 'Content-Type: application/json' -d '[{"test": "hello3"}]'
curl http://localhost:2021/log/ingest -X POST -H 'Content-Type: application/json' -d '[{"test": "hello4"}]'

6. Search - no new documents

GET test_forbidden/_search
{
  "query": {
    "match_all": {}
  }
}

Also, see Data Prepper logs

data-prepper           | 2024-09-10T18:24:11,693 [opensearch-retry-pipeline-sink-worker-2-thread-2] WARN  org.opensearch.dataprepper.plugins.sink.opensearch.BulkRetryStrategy - operation = Index, error = index [test_forbidden] blocked by: [FORBIDDEN/8/index write (api)];

7. Remove the write block

PUT test_forbidden/_settings
{
  "index.blocks.read_only_allow_delete": null
}

8. Search

GET test_forbidden/_search
{
  "query": {
    "match_all": {}
  }
}

At this point, we should see new documents. But, they are not present.

Expected behavior

At step 8, we should see the documents.

Environment (please complete the following information):

Data Prepper 2.9.0

Start with the OpenSearch sink's BulkRetryStrategy, which is named in the Data Prepper warning, and reproduce the failure using the opensearch-retry-pipeline configuration and the listed OpenSearch block and unblock requests. Done means messages written after the write block is removed are eventually indexed and visible in the search results.