Fix Apache HttpClient 5 reactor recovery

[sharp-pixel]

Fixes #1969

Problem

ApacheHttpClient5Transport could enter an unrecoverable state after Apache HttpClient 5 reported IOReactorShutdownException. Once the underlying I/O reactor was shut down, the transport continued using the same dead CloseableHttpAsyncClient, so every later request failed until the application recreated the transport.

One common trigger is memory pressure while buffering large or highly concurrent responses. The existing per-response buffer limit capped a single response, but it did not cap aggregate heap usage across concurrent responses, so many in-flight responses could still exhaust the heap and kill the reactor.

Solution

This change makes ApacheHttpClient5Transport recover from reactor shutdowns when it owns the client lifecycle:

• Rebuilds and starts a fresh Apache HC5 async client when IOReactorShutdownException is detected.
• Handles both synchronous execute(...) failures and callback-delivered reactor shutdown failures.
• Retries the affected request once on the rebuilt client.
• Adds rebuild backoff to avoid rebuild storms if the reactor keeps dying.
• Prevents recovery after the transport has been explicitly closed.
• Keeps externally supplied clients unchanged; recovery is disabled when the transport does not own the client lifecycle.

It also reduces the chance of reactor-killing memory failures:

• Adds an optional aggregate response-buffer memory budget.
• Adds bounded response-consumer capacity increments when a budget is active.
• Converts OutOfMemoryError during response buffering into a request-level IOException so the reactor thread is not killed.
• Fails fast without node denylisting/retry for deterministic client-side failures such as shared-budget exhaustion, oversized single responses, and OOM-caused buffering failures.
• Resets leaked response-buffer budget reservations when the client is rebuilt, including request-specific budgets.

Testing

Ran:

./gradlew :java-client:test \
  --tests org.opensearch.client.transport.httpclient5.ApacheHttpClient5TransportRecoveryTest \
  --tests org.opensearch.client.transport.httpclient5.internal.ResponseMemoryBudgetTest \
  --tests org.opensearch.client.transport.httpclient5.internal.HeapBufferedAsyncEntityConsumerTest

./gradlew :java-client:test \
  --tests 'org.opensearch.client.transport.httpclient5.*' \
  --tests 'org.opensearch.client.transport.httpclient5.internal.*'

./gradlew :java-client:spotlessJavaCheck

git diff --check

Also verified the locally published modified client in a standalone sample app using the normal existing transport builder path against OpenSearch 3.7.0 in Docker:

• ping and info/version
• create index
• index and search a document
• concurrent indexing/search smoke test with one client instance

[andrross]

I'm not sure it is safe to do this. The contract of java.lang.Error generally is that it "indicates serious problems that a reasonable application should not try to catch". I don't think you can assume the JVM is in a workable state after observing an error and the right thing to do is let the process crash. In this case, you don't actually know if the response buffer in question here caused the OOM, versus something else in the application unrelated to this client that consumed all the heap.

[andrross]

Would it make sense to break this into 2 PRs? One with the fix to allow recovery from reactor shutdowns, and another PR with the changes to reduce the chance of reactor-killing memory failures.

[andrross]

Looks good to me. @reta can you take a look?

@sharp-pixel The PR description should be updated since this no longer has all the original changes.

[andrross]

Nitpick, but easier to follow if you invert the logic:

Suggested change

	if (isRecoverableReactorFailure(runtimeFailure, client) == false) {
	throw runtimeFailure;
	}
	// The I/O reactor has been shut down. Try to recover by rebuilding the client, then retry this request once
	// on the fresh client.
	retryAfterReactorFailure(nodeTuple, options, request, warningsHandler, listener, node, client, runtimeFailure, allowRecovery);
	return;
	if (isRecoverableReactorFailure(runtimeFailure, client)) {
	// The I/O reactor has been shut down. Try to recover by rebuilding the client, then retry this request once
	// on the fresh client.
	retryAfterReactorFailure(nodeTuple, options, request, warningsHandler, listener, node, client, runtimeFailure, allowRecovery);
	} else {
	throw runtimeFailure;
	}

[reta]

Looks good to me. @reta can you take a look?

Yeah I will take a look in a few days, thanks @andrross and @sharp-pixel

[reta]

@sharp-pixel few things from my side (thanks for working on it first of all):

• I was under impression that much smaller scoped change will be ported (from Safe response consumer for rest5client elastic/elasticsearch-java#1049), I think that one helps to eliminate the cause of reactor shutdowns
• Having the same behavior across both clients is really helpful (since other projects could support OpenSearch and Elasticsearch with minimal efforts)

In general, I think there is nothing fundamentally wrong to ask the caller to recreate the client in case of fatal failures (like I/O reactor one). The current fix is changing a lot of internals of the ApacheHttpClient5Transport, but we should not forget that we also have AwsSdk2Transport (that internally uses Apache5HttpClient) and it would be reasonable to expect the same client recovery there (since it is the same AHC5 client under the hood, with all the same issues).

May be we should take a step back and:

• port the fix from elasticsearch-java
• have a wrapper around OpenSearchTransport (something like RecoverableOpenSearchTransport) that will accept a builder and would recreate the underlying transport in case of recoverable failures? (that may require some additional hooks / listeners but that should not be too difficult I believe)

Thank you.

[andrross]

I was under impression that much smaller scoped change will be ported (from elastic/elasticsearch-java#1049), I think that one helps to eliminate the cause of reactor shutdowns

@reta That change causes certain code paths to convert OutOfMemoryErrors into RuntimeExceptions and allow the JVM to continue operating. Maybe I'm paranoid but that seems like a bad idea? If I'm wrong about that then it's probably fine to take the smaller-scoped change.

[reta]

I was under impression that much smaller scoped change will be ported (from elastic/elasticsearch-java#1049), I think that one helps to eliminate the cause of reactor shutdowns

@reta That change causes certain code paths to convert OutOfMemoryErrors into RuntimeExceptions and allow the JVM to continue operating. Maybe I'm paranoid but that seems like a bad idea? If I'm wrong about that then it's probably fine to take the smaller-scoped change.

@andrross It does indeed, but it has been applied to widely used software (elasticsearch-java), so may be we both are overly paranoid?

[andrross]

I was under impression that much smaller scoped change will be ported (from elastic/elasticsearch-java#1049), I think that one helps to eliminate the cause of reactor shutdowns

@reta That change causes certain code paths to convert OutOfMemoryErrors into RuntimeExceptions and allow the JVM to continue operating. Maybe I'm paranoid but that seems like a bad idea? If I'm wrong about that then it's probably fine to take the smaller-scoped change.

@andrross It does indeed, but it has been applied to widely used software (elasticsearch-java), so may be we both are overly paranoid?

@reta Without the fix, does the OOM get swallowed anyway, just with the side effect that the netty reactor is shut down?

[reta]

I was under impression that much smaller scoped change will be ported (from elastic/elasticsearch-java#1049), I think that one helps to eliminate the cause of reactor shutdowns

@reta That change causes certain code paths to convert OutOfMemoryErrors into RuntimeExceptions and allow the JVM to continue operating. Maybe I'm paranoid but that seems like a bad idea? If I'm wrong about that then it's probably fine to take the smaller-scoped change.

@andrross It does indeed, but it has been applied to widely used software (elasticsearch-java), so may be we both are overly paranoid?

@reta Without the fix, does the OOM get swallowed anyway, just with the side effect that the netty reactor is shut down?

I am not 100% sure, I will look closely over the weekend, I believe that either a) reactor is shutdown cleanly b) reactor is not shutdown, but the thread(s) dies?

[andrross]

@reta If the existing behavior is that Netty prevents the OOM exception from propagating to the application, then the fix to prevent OOMs from shutting down the reactor seems much less risky.

[reta]

@reta If the existing behavior is that Netty prevents the OOM exception from propagating to the application, then the fix to prevent OOMs from shutting down the reactor seems much less risky.

Thanks @andrross , I will double check that, thanks!

[reta]

@reta If the existing behavior is that Netty prevents the OOM exception from propagating to the application, then the fix to prevent OOMs from shutting down the reactor seems much less risky.

Thanks @andrross , I will double check that, thanks!

@andrross so I did check the XxxError handling that the patch applies (from elastic/elasticsearch-java#1049) prevents the reactor shutdown - basically, there is no need to recreate the client