opensearch-project / opensearch-project/data-prepper
[BUG] OpenSearchSink leaks a full IOReactor (Apache HttpAsyncClient) on every AbstractSink init-retry attempt
Nobody has claimed this yet.
- Dominant language
- Java
- Stars
- 374
- Forks
- 354
- Avg merge
- 3d 18h
- Merged PRs (30d)
- 8
Description
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.usageset to ~80% of heap - Sink auth: static cert + username/password from pipeline config (no AWS Secrets Manager /
SigV4 rotation involved — ruled outOpenSearchClientRefresher's scheduled-refresh path as a
factor)
Steps to reproduce (best effort — the underlying doInitialize() failure is intermittent)
- Run a
data-prepperpipeline with an OpenSearch sink under sustained load for several hours. - Whenever the sink's
doInitialize()throws (the exact triggering exception was not isolated —
it is not adjacent toBulkRetryStrategyretry-exhaustion in the logs, so it looks like a
transient failure inside client construction itself, not a bulk-request failure),
AbstractSinkretriesdoInitialize()again on the sink's own worker thread. - 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.
Contributor guide
First steps
- Read the whole issue, then the project's contributing guide.
- Comment on the issue to say you are picking it up — it saves two people doing the same work.
- Fork the repository and make your change on a branch.
- Open a pull request that references the issue number.
Research direction
Start with OpenSearchSink.doInitialize() and doInitializeInternal(), then trace AbstractSink’s initialization retry flow to identify how each client is replaced after a failure. Compare the lifecycle handling with OpenSearchClientRefresher and the fix in #4770. Done means failed initialization retries close the previous RestHighLevelClient and repeated retries no longer increase I/O dispatcher thread counts.
Written by the indexing model from the issue text.
Assessment
- Tech stack
- java
- Domain
- backend
- Issue type
- Bug
- Difficulty
- 3/5
- Estimated time
- 1-2 days
- Activity status
- Active
- Clarity
- Clearly specified
- Newbie friendliness
- 72/100