[Inference API] Fix inference initialization thread exhaustion by jonathan-buttner · Pull Request #147063 · elastic/elasticsearch

jonathan-buttner · 2026-04-21T21:18:16Z

This PR fixes some errors we've seen when the first inference API request is received for a node. The stacktrace indicates that waiting for the initialization to complete times out.

The issue is there's a potential for exhausting the inference utility threadpool while waiting for internal services to initialize.

My understanding is that the issue is temporary. Once some of the threads have finished timing out the actual initialization can be performed and requests should begin working.

This PR refactors the logic in a few ways:

Only the first request to kick off the initialization will use a thread execute, additional calls are queued in a subscribing listener which if succeeded/failed will do so for all the queued listeners
I also moved the startAsynchronously() call to be done in the send*() methods. This guarantees that it'll be called in all paths. Previously we were making a direct call to startAsynchronously() before each
HttpRequestSender::startSynchronously() now throws. Once we refactor Amazon Bedrock to not leverage SenderService we should be able to drop startSynchronously() from the interface entirely

…sender-startup

…elasticsearch into ia-http-sender-startup

…sender-startup

macroscopeapp · 2026-04-27T16:03:12Z

Approvability

Verdict: Needs human review

Refactors the inference HTTP sender initialization from blocking to async using SubscribableListener, fundamentally changing how and when startup occurs. Several review comments remain unresolved, including author requests for verification of thread safety and suggestions for API simplification.

No code changes detected at 4386f07. Prior analysis still applies.

^{You can customize Macroscope's approvability policy. Learn more.}

…sender-startup

…elasticsearch into ia-http-sender-startup

DonalEvans · 2026-04-27T21:26:29Z

+        var wrappedListener = timeout != null
+            ? ListenerTimeouts.wrapWithTimeout(
+                threadPool,
+                timeout,
+                threadPool.executor(UTILITY_THREAD_POOL_NAME),
+                contextPreservedListener,
+                ignored -> contextPreservedListener.onFailure(
+                    new ElasticsearchStatusException("Http sender startup did not complete in time", RestStatus.SERVICE_UNAVAILABLE)
+                )
+            )
+            : contextPreservedListener;


I think this is slightly easier to read as:

var wrappedListener = timeout == null ? contextPreservedListener : ListenerTimeouts.wrapWithTimeout( threadPool, timeout, threadPool.executor(UTILITY_THREAD_POOL_NAME), contextPreservedListener, ignored -> contextPreservedListener.onFailure( new ElasticsearchStatusException("Http sender startup did not complete in time", RestStatus.SERVICE_UNAVAILABLE) ) );

DonalEvans · 2026-04-27T22:10:13Z

+            : contextPreservedListener;
+        // All callers — first and concurrent — register here. SubscribableListener fires immediately
+        // if startup is already done, otherwise queues the listener until startInternal completes.
+        startupNotifier.addListener(wrappedListener);


It's certainly supposed to be thread safe. I think that if it's not, we'd have a lot of problems outside of this use.

DonalEvans · 2026-04-27T22:51:14Z

 public interface Sender extends Closeable {
+
+    /**
+     * Initialize the sender synchronously. This must be called before calling {@link #send} or {@link #sendWithoutQueuing}.


Nitpick, but would it be more accurate to say that the start methods must be called before or as part of calling send() or sendWithoutQueuing(), since that's what HttpRequestSender does?

At the moment it's a bit confusing as the bedrock implementation does not start itself on send. At the moment, it seems like startAsynchronously() is called from a few other places (e.g. on PUT).

I think we should clarify what the contract is here: do users of Sender need to manage starting or not? If not, I'd remove the start methods entirely from the interface.

At the moment it's a bit confusing as the bedrock implementation does not start itself on send. At the moment, it seems like startAsynchronously() is called from a few other places (e.g. on PUT).

Yeah Bedrock needs to be completely refactored. It should be more like the SageMaker integration because it doesn't need to rely on the SenderService functionality at all because the internal SDK handles rate limiting already.

We have an issue for that refactoring, but we haven't gotten to it yet. Once Bedrock no longer depends on SenderService then we can remove the startSynchronously() method.

You make a good point about PUT though. I don't think we need SenderService::start() to call start*() at all because the validation in the PUT logic calls the send*() first. I think the SenderService::start() can be a noop actually.

Then I think we can remove the startAsync/Sync() from the interface.

DonalEvans · 2026-04-27T22:58:29Z

    @Override
    public void embeddingInfer(Model model, EmbeddingRequest request, TimeValue timeout, ActionListener<InferenceServiceResults> listener) {
-        SubscribableListener.newForked(this::init).<InferenceServiceResults>andThen((embeddingInferListener) -> {
+        try {


Looks good.

DonalEvans · 2026-04-27T23:00:06Z

@@ -419,24 +427,20 @@ protected boolean supportsChunkedInfer() {
    }

    public void start(Model model, ActionListener<Boolean> listener) {


I think we can inline this method now, since it's only called in tests.

DonalEvans · 2026-04-27T23:31:54Z

+    public void start(Model model, @Nullable TimeValue timeout, ActionListener<Boolean> listener) {
+        SubscribableListener.<Void>newForked(l -> sender.startAsynchronously(l, timeout))
+            .<Boolean>andThen((doStartListener) -> doStart(model, doStartListener))
+            .addListener(listener);


It's a little awkward that we need to wait for startAsynchronously() to complete and then complete a new listener with a boolean here. Would it make sense to change start() to take ActionListener<Void> instead of ActionListener<Boolean>? We don't seem to actually use the boolean value returned by the listener anywhere, and I'm not sure in what situation we'd ever complete with false instead of just failing the listener with an exception.

If we make the listener passed in here ActionListener<Void> then I think we can simplify the method to just:

public void start(Model model, @Nullable TimeValue timeout, ActionListener<Void> listener) { sender.startAsynchronously(listener, timeout); }

I think the interface was originally designed that way to support whether an ML deployment was started:

https://github.com/elastic/elasticsearch/blob/main/x-pack/plugin/inference/src/main/java/org/elasticsearch/xpack/inference/services/elasticsearch/BaseElasticsearchInternalService.java#L130

Which is called here: https://github.com/elastic/elasticsearch/blob/main/x-pack/plugin/inference/src/main/java/org/elasticsearch/xpack/inference/action/TransportPutInferenceModelAction.java#L313

Doesn't look like the boolean is used in the response so we could remove it. I'll create an issue to remove the Boolean aspect. For SenderService, I think we can refactor because it doesn't seem like any services override doStart() so we can probably remove that and simply the method but for now I think I'll leave the Boolean portion.

DonalEvans · 2026-04-28T00:01:53Z

    }

-    public void testStart_ThrowsExceptionWaitingForStartToComplete() {
+    public void testStartSync_ThrowsServiceUnavailableWhenStartupTimesOut() {


This test doesn't actually call HttpRequestSender.startSynchronously(), since that method is deprecated and throws, so I think it should just be removed to avoid confusion. It's functionally the same as testStartAsync_ThrowsServiceUnavailableWhenStartupTimesOut(), except with the actionGet() call hidden in the static startSynchronously() method.

DonalEvans · 2026-04-28T00:07:37Z

+        var timedListener = new TestPlainActionFuture<Void>();
+        sender.startAsynchronously(timedListener, TimeValue.timeValueMillis(1));
+        var ex = expectThrows(ElasticsearchStatusException.class, () -> timedListener.actionGet(TEST_REQUEST_TIMEOUT));
+        assertThat(ex.getMessage(), is("Http sender startup did not complete in time"));
+        assertThat(ex.status(), is(RestStatus.SERVICE_UNAVAILABLE));

-        var listenerCompleted = new PlainActionFuture<Void>();
-        sender.startAsynchronously(listenerCompleted);
-        assertNull(listenerCompleted.actionGet(TIMEOUT));
+        // Unblock startup so startupNotifier resolves successfully
+        blockStartLatch.countDown();


This test could get a little more coverage if we add two listeners before calling blockStartLatch.countDown(), one with a very short timeout that we expect to fail (like the test is already doing) and one with a longer timeout (or no timeout at all) that we expect to succeed after we count down the latch. That way we also get confirmation that if there are multiple listeners queued and one times out, the other queued ones aren't impacted, which isn't currently covered here.

dimitris-athanasiou · 2026-04-28T08:25:23Z

-                executorServiceSettings,
-                requestSender
-            );
+            var service = new RequestExecutorService(serviceComponents.threadPool(), null, executorServiceSettings, requestSender);


Since we now always pass null as the start-completed latch here could we remove the argument entirely? Tests can be reworked if needed.

Tests of the RequestExecutorService leverage it so I'll expose a different constructor 👍

dimitris-athanasiou · 2026-04-28T08:31:25Z

+                logger.warn("Failed to execute http sender start thread", e);
+                startupNotifier.onFailure(
+                    new ElasticsearchStatusException(
+                        "Failed to begin initializing inference components",


Should we reword this to Failed to initialize inference components?

I intentionally made the string different from the one below so it'd be easier to differentiate between the two errors if we got an SDH or something. I can make them the same though if you'd like. We'll have to logs which should have a stacktrace.

dimitris-athanasiou · 2026-04-28T08:33:18Z

+        var wrappedListener = timeout != null
+            ? ListenerTimeouts.wrapWithTimeout(
+                threadPool,
+                timeout,
+                threadPool.executor(UTILITY_THREAD_POOL_NAME),
+                contextPreservedListener,
+                ignored -> contextPreservedListener.onFailure(
+                    new ElasticsearchStatusException("Http sender startup did not complete in time", RestStatus.SERVICE_UNAVAILABLE)
+                )
+            )
+            : contextPreservedListener;


dimitris-athanasiou · 2026-04-28T09:01:55Z

 public interface Sender extends Closeable {
+
+    /**
+     * Initialize the sender synchronously. This must be called before calling {@link #send} or {@link #sendWithoutQueuing}.


At the moment it's a bit confusing as the bedrock implementation does not start itself on send. At the moment, it seems like startAsynchronously() is called from a few other places (e.g. on PUT).

I think we should clarify what the contract is here: do users of Sender need to manage starting or not? If not, I'd remove the start methods entirely from the interface.

dimitris-athanasiou · 2026-04-28T09:06:18Z

+    public void start(Model model, @Nullable TimeValue timeout, ActionListener<Boolean> listener) {
+        SubscribableListener.<Void>newForked(l -> sender.startAsynchronously(l, timeout))
+            .<Boolean>andThen((doStartListener) -> doStart(model, doStartListener))
+            .addListener(listener);


dimitris-athanasiou · 2026-04-28T09:10:39Z

    }

+    public void testConcurrentSend_OnlyStartsOnce() throws Exception {
+        // At a minimum we need 1 for the startAsynchronously to kick off the init tasks


Do we need to give the pool 2 threads in order to properly check tasks are only started once? Having a single thread forces start to be finished effectively before it can be run again.

Yeah good point.

…sender-startup

Jan-Kazlouski-elastic

Great job Jonathan. Left a few small comments.

Jan-Kazlouski-elastic · 2026-04-28T13:23:50Z

-        assert startInitiated.get() : "call start() before sending a request";
-        waitForStartToComplete();
-        service.execute(requestCreator, inferenceInputs, timeout, listener);
+        SubscribableListener.<Void>newForked(l -> startAsynchronously(l, STARTUP_TIMEOUT))


Perhaps we would want to change it to more meaningful variable name for clarity?
Param can be renamed to actionListener and this one to just listener

Jan-Kazlouski-elastic · 2026-04-28T13:55:36Z

+            try {
+                threadPool.executor(UTILITY_THREAD_POOL_NAME).execute(this::startInternal);
+            } catch (Exception e) {
+                // Consider making this an error log because it's not recoverable


I think this comment is valid. If it is not recoverable - it should be ERROR level. Because it is reserved for unrecoverable states.

Good call out. I'll create an issue: #147731

I'd like these changes to be in prod for a bit before we switch to using error logs in case we have a failure. Error logs will trigger alerts by default.

Good thinking. I agree.

Jan-Kazlouski-elastic · 2026-04-28T14:18:22Z

+     * Initialize the sender asynchronously. This must be called before calling {@link #send} or {@link #sendWithoutQueuing}.
+     * The listener will be notified when the initialization is complete or if it fails.
+     * @param listener the listener to notify when initialization is complete or if it fails
+     * @param timeout the maximum time to wait for initialization to complete. If null, the implementation should use a default timeout.


It says If null, the implementation should use a default timeout.
Correct me if I'm wrong, but I can see an implication here that some default timeout exists within the startAsynchronously method. But that is not true. Timeout for startup is only passed from send and start methods. There is no default timeout wrapping happening inside if it is passed as null. I think that javadoc is misleading.
We should either make code enforce some default timeout or change the javadoc.

Also If for send we have 5 sec default timeout, for start we'll be waiting forever if null is passed.

Good call out. I've refactored this so the start() doesn't exist anymore. Take a look and let me know if I missed something though.

Thanks @jonathan-buttner
I believe that solves it. We could inline timeout value because it is null only in tests, but I think it is more handy for testing purposes and more future-proof to have it as a param. So it is up to you.

dimitris-athanasiou

LGTM

Jan-Kazlouski-elastic

Thanks for the fixes @jonathan-buttner
This PR looks good to me.

jonathan-buttner · 2026-05-01T20:39:11Z

💚 All backports created successfully

Status	Branch	Result
✅	9.4
✅	9.3
✅	9.2

Questions ?

Please refer to the Backport tool documentation

…) (#148151) * Passing tests for init refactor * Adding try/catch changes and thread tests * [CI] Auto commit changes from spotless * Fixing the tests * Using static helper to start synchronously for tests * Removing unneeded start call * More refactoring to remove unneeded threads * Adding more test coverage * Adding default timeout * [CI] Auto commit changes from spotless * Update docs/changelog/147063.yaml * Fixing merge and clean up * [CI] Auto commit changes from spotless * Addressing coderabbit feedback * Fixing tests * Fixing tests * Address first feedback * Addressing second round of feedback --------- Co-authored-by: elasticsearchmachine <infra-root+elasticsearchmachine@elastic.co> (cherry picked from commit dbfda06) # Conflicts: # x-pack/plugin/inference/src/main/java/org/elasticsearch/xpack/inference/services/SenderService.java # x-pack/plugin/inference/src/test/java/org/elasticsearch/xpack/inference/services/contextualai/action/ContextualAiActionCreatorTests.java # x-pack/plugin/inference/src/test/java/org/elasticsearch/xpack/inference/services/jinaai/action/JinaAIActionCreatorTests.java # x-pack/plugin/inference/src/test/java/org/elasticsearch/xpack/inference/services/mistral/action/MistralActionCreatorTests.java

…147063) (#148153) * [Inference API] Fix inference initialization thread exhaustion (#147063) * Passing tests for init refactor * Adding try/catch changes and thread tests * [CI] Auto commit changes from spotless * Fixing the tests * Using static helper to start synchronously for tests * Removing unneeded start call * More refactoring to remove unneeded threads * Adding more test coverage * Adding default timeout * [CI] Auto commit changes from spotless * Update docs/changelog/147063.yaml * Fixing merge and clean up * [CI] Auto commit changes from spotless * Addressing coderabbit feedback * Fixing tests * Fixing tests * Address first feedback * Addressing second round of feedback --------- Co-authored-by: elasticsearchmachine <infra-root+elasticsearchmachine@elastic.co> (cherry picked from commit dbfda06) # Conflicts: # x-pack/plugin/inference/src/main/java/org/elasticsearch/xpack/inference/external/http/sender/HttpRequestSender.java # x-pack/plugin/inference/src/main/java/org/elasticsearch/xpack/inference/services/SenderService.java # x-pack/plugin/inference/src/test/java/org/elasticsearch/xpack/inference/external/http/sender/HttpRequestSenderTests.java # x-pack/plugin/inference/src/test/java/org/elasticsearch/xpack/inference/services/contextualai/action/ContextualAiActionCreatorTests.java # x-pack/plugin/inference/src/test/java/org/elasticsearch/xpack/inference/services/jinaai/action/JinaAIActionCreatorTests.java # x-pack/plugin/inference/src/test/java/org/elasticsearch/xpack/inference/services/mistral/action/MistralActionCreatorTests.java # x-pack/plugin/inference/src/test/java/org/elasticsearch/xpack/inference/services/openai/OpenAiServiceTests.java * Fixing test * Skipping failing test

jonathan-buttner added 2 commits April 21, 2026 14:09

Passing tests for init refactor

eb76ab5

Adding try/catch changes and thread tests

2909882

jonathan-buttner added >bug :SearchOrg/Inference Label for the Search Inference team Team:Search - Inference branch:9.2 v9.5.0 branch:9.4 labels Apr 21, 2026

elasticsearchmachine added v9.2.9 v9.3.4 v9.4.1 and removed branch:9.2 branch:9.4 labels Apr 21, 2026

macroscopeapp Bot reviewed Apr 21, 2026

View reviewed changes

Comment thread .../src/main/java/org/elasticsearch/xpack/inference/external/http/sender/HttpRequestSender.java Outdated

Comment thread .../src/main/java/org/elasticsearch/xpack/inference/external/http/sender/HttpRequestSender.java Outdated

elasticsearchmachine and others added 3 commits April 21, 2026 21:24

[CI] Auto commit changes from spotless

10b3a4e

Merge branch 'main' of github.com:elastic/elasticsearch into ia-http-…

9dcaf5c

…sender-startup

Merge branch 'ia-http-sender-startup' of github.com:jonathan-buttner/…

e20eb47

…elasticsearch into ia-http-sender-startup

macroscopeapp Bot reviewed Apr 21, 2026

View reviewed changes

Comment thread .../src/main/java/org/elasticsearch/xpack/inference/external/http/sender/HttpRequestSender.java Outdated

Fixing the tests

ad1b2f6

elasticsearchmachine added v9.3.5 and removed v9.3.4 labels Apr 23, 2026

jonathan-buttner linked an issue Apr 23, 2026 that may be closed by this pull request

[Inference API] Http sender startup failed #147290

Closed

jonathan-buttner added 6 commits April 23, 2026 14:46

Using static helper to start synchronously for tests

2e540fe

Removing unneeded start call

6504fc4

More refactoring to remove unneeded threads

599cef4

Adding more test coverage

67bb7a6

Merge branch 'main' of github.com:elastic/elasticsearch into ia-http-…

7a9ff9b

…sender-startup

Adding default timeout

f2179a9

jonathan-buttner and others added 4 commits April 27, 2026 13:48

Merge branch 'main' into ia-http-sender-startup

f924921

Fixing tests

3a97ec4

Merge branch 'main' of github.com:elastic/elasticsearch into ia-http-…

fecf684

…sender-startup

Merge branch 'ia-http-sender-startup' of github.com:jonathan-buttner/…

92e9142

…elasticsearch into ia-http-sender-startup

DonalEvans reviewed Apr 28, 2026

View reviewed changes

dimitris-athanasiou reviewed Apr 28, 2026

View reviewed changes

Merge branch 'main' of github.com:elastic/elasticsearch into ia-http-…

877b2ae

…sender-startup

Jan-Kazlouski-elastic reviewed Apr 28, 2026

View reviewed changes

Address first feedback

7108af7

jonathan-buttner mentioned this pull request Apr 28, 2026

[Inference API] Use error logging for initialization failures #147731

Open

jonathan-buttner and others added 2 commits April 28, 2026 11:44

Addressing second round of feedback

3f603e9

Merge branch 'main' into ia-http-sender-startup

65c933e

jonathan-buttner requested review from DonalEvans, Jan-Kazlouski-elastic and dimitris-athanasiou April 28, 2026 15:58

DonalEvans approved these changes Apr 28, 2026

View reviewed changes

dimitris-athanasiou approved these changes Apr 28, 2026

View reviewed changes

Merge branch 'main' into ia-http-sender-startup

4386f07

jonathan-buttner enabled auto-merge (squash) April 28, 2026 18:25

Merge branch 'main' into ia-http-sender-startup

af93d5f

Jan-Kazlouski-elastic approved these changes Apr 28, 2026

View reviewed changes

jonathan-buttner merged commit dbfda06 into elastic:main Apr 28, 2026
38 checks passed

jonathan-buttner deleted the ia-http-sender-startup branch April 28, 2026 20:55

		@@ -419,24 +427,20 @@ protected boolean supportsChunkedInfer() {
		}

		public void start(Model model, ActionListener<Boolean> listener) {

Conversation

jonathan-buttner commented Apr 21, 2026 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

macroscopeapp Bot commented Apr 27, 2026 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Approvability

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

jonathan-buttner Apr 28, 2026 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Jan-Kazlouski-elastic left a comment

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Jan-Kazlouski-elastic Apr 28, 2026 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

dimitris-athanasiou left a comment

Choose a reason for hiding this comment

jonathan-buttner commented Apr 21, 2026 •

edited

Loading

macroscopeapp Bot commented Apr 27, 2026 •

edited

Loading

jonathan-buttner Apr 28, 2026 •

edited

Loading

Jan-Kazlouski-elastic Apr 28, 2026 •

edited

Loading