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