[BUG] OpenSearchSink leaks a full IOReactor (Apache HttpAsyncClient) on every AbstractSink init-retry attempt

[robsche91-ghub]

Description

OpenSearchSink extends AbstractSink and is constructed with
super(pluginSetting, Integer.MAX_VALUE, INITIALIZE_RETRY_WAIT_TIME_MS), meaning
AbstractSink's own retry wrapper will call doInitialize() (→ doInitializeInternal() →
connectionConfiguration.createClient(awsCredentialsSupplier)) again, essentially without limit,
any time doInitialize() throws.

Each retry attempt builds a brand-new RestHighLevelClient, which internally spins up a new
Apache HttpAsyncClient IOReactor (default ioThreadCount = host CPU core count). The
previous attempt's client/IOReactor is never closed — this only happens in the sink's final
shutdown(). Every failed-then-retried doInitialize() call permanently leaks a full IOReactor's
worth of threads (equal to the host's core count), which stay parked in epoll_wait forever
(visible as RUNNABLE in a thread dump despite being fully idle).

Over hours/days this leak accumulates until heap/thread pressure trips Data Prepper's own
circuit_breakers.heap.usage threshold, causing sustained TimeoutException: Buffer does not have enough capacity left ... timed out waiting for slots on the pipeline's source, well below
the JVM's actual configured heap ceiling — the leaked resources, not real data volume, are what
saturate it.

This looks like the same bug class as #4770 ("Close Opensearch RestHighLevelClient in
OpenSearchClientRefresher on shutdown and initialization failure," fixed in v2.10), but a
different, apparently uncovered code path: #4770's fix specifically targeted
OpenSearchClientRefresher not being closed during its own init-failure/shutdown handling. It
does not appear to touch AbstractSink's generic init-retry wrapper, which is the path observed
firing here. This was seen on data-prepper:2.15.0 (well past v2.10) — the leak is still present.

Environment

• Data Prepper version: opensearchproject/data-prepper:2.15.0 (Docker)
• Deployment: single log pipeline, OpenSearch sink, several sink routes, multiple sink workers
  configured
• Host: multi-core Linux, JVM heap several GB, circuit_breakers.heap.usage set to ~80% of heap
• Sink auth: static cert + username/password from pipeline config (no AWS Secrets Manager /
  SigV4 rotation involved — ruled out OpenSearchClientRefresher's scheduled-refresh path as a
  factor)

Steps to reproduce (best effort — the underlying doInitialize() failure is intermittent)

1. Run a data-prepper pipeline with an OpenSearch sink under sustained load for several hours.
2. Whenever the sink's doInitialize() throws (the exact triggering exception was not isolated —
   it is not adjacent to BulkRetryStrategy retry-exhaustion in the logs, so it looks like a
   transient failure inside client construction itself, not a bulk-request failure),
   AbstractSink retries doInitialize() again on the sink's own worker thread.
3. Take a thread dump (jcmd <pid> Thread.print) before and after such an event.

Observed vs. expected

Observed: thread count grows in discrete steps of exactly ioThreadCount (= CPU core count)
every time doInitialize() is retried; docker exec data-prepper jcmd 1 Thread.print | grep -c '^"I/O dispatcher' climbs over time and never drops until the container is restarted.

Expected: the previous (failed) client/IOReactor should be closed before or during the next
retry attempt, so thread count stays flat regardless of how many times doInitialize() retries.

Suggested fix

In OpenSearchSink.doInitialize()/doInitializeInternal(), hold a reference to any
partially-constructed RestHighLevelClient from a prior failed attempt and explicitly close()
it before constructing a new one on retry — mirroring the fix already applied to
OpenSearchClientRefresher in #4770, just applied to this second, independent creation path.

Workaround in the meantime

Scheduled restart of the data-prepper container to reclaim leaked threads before they
accumulate enough to trip the heap circuit breaker. Not a fix — a stopgap.