opensearch-project / opensearch-project/data-prepper

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

Open
#7,003 1 comment 0 reactions 0 assignees View on GitHub

Nobody has claimed this yet.

bug
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.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.

Contributor guide

Open the contributing guide

First steps

  1. Read the whole issue, then the project's contributing guide.
  2. Comment on the issue to say you are picking it up — it saves two people doing the same work.
  3. Fork the repository and make your change on a branch.
  4. 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

Get new issues in your inbox

A short digest of beginner-friendly GitHub issues.