Handle InternalSendException inline for non-forking handlers (#114375)

When TransportService fails to send a transport action, it can complete
the listener's `onFailure` with the `generic` executor. If the listener
is a `PlainActionFuture` and also waits to be completed with a `generic`
thread, it will trip the `assertCompleteAllowed` assertion. 

https://github.com/elastic/elasticsearch/blob/fb482f863d5430702b19bd3dd23e9d8652f12ddd/server/src/main/java/org/elasticsearch/transport/TransportService.java#L1062-L1064

With this PR, we no longer fork to the generic thread pool and instead
just handle the exeption inline with the current thread. The expectation
is that the downstream handler should take care potential stack overflow
issues. This is similar to what is done in #109236

Author: ywangd

docs/changelog/114375.yaml

@@ -0,0 +1,5 @@
+pr: 114375
+summary: Handle `InternalSendException` inline for non-forking handlers
+area: Distributed
+type: bug
+issues: []

server/src/main/java/org/elasticsearch/transport/TransportService.java

@@ -1059,8 +1059,9 @@ private Executor getInternalSendExceptionExecutor(Executor handlerExecutor) {
         if (lifecycle.stoppedOrClosed()) {
             // too late to try and dispatch anywhere else, let's just use the calling thread
             return EsExecutors.DIRECT_EXECUTOR_SERVICE;
-        } else if (handlerExecutor == EsExecutors.DIRECT_EXECUTOR_SERVICE) {
-            // if the handler is non-forking then dispatch to GENERIC to avoid a possible stack overflow
+        } else if (handlerExecutor == EsExecutors.DIRECT_EXECUTOR_SERVICE && enableStackOverflowAvoidance) {
+            // If the handler is non-forking and stack overflow protection is enabled then dispatch to GENERIC
+            // Otherwise we let the handler deal with any potential stack overflow (this is the default)
             return threadPool.generic();
         } else {
             return handlerExecutor;

server/src/test/java/org/elasticsearch/transport/TransportServiceLifecycleTests.java

@@ -149,8 +149,13 @@ public void testInternalSendExceptionForksToHandlerExecutor() {
         }
     }
 
-    public void testInternalSendExceptionForksToGenericIfHandlerDoesNotFork() {
-        try (var nodeA = new TestNode("node-A")) {
+    public void testInternalSendExceptionForksToGenericIfHandlerDoesNotForkAndStackOverflowProtectionEnabled() {
+        try (
+            var nodeA = new TestNode(
+                "node-A",
+                Settings.builder().put(TransportService.ENABLE_STACK_OVERFLOW_AVOIDANCE.getKey(), true).build()
+            )
+        ) {
             final var future = new PlainActionFuture<TransportResponse.Empty>();
             nodeA.transportService.sendRequest(
                 nodeA.getThrowingConnection(),
@@ -165,6 +170,33 @@ public void testInternalSendExceptionForksToGenericIfHandlerDoesNotFork() {
 
             assertEquals("simulated exception in sendRequest", getSendRequestException(future, IOException.class).getMessage());
         }
+        assertWarnings(
+            "[transport.enable_stack_protection] setting was deprecated in Elasticsearch and will be removed in a future release."
+        );
+    }
+
+    public void testInternalSendExceptionWithNonForkingResponseHandlerCompletesListenerInline() {
+        try (var nodeA = new TestNode("node-A")) {
+            final Thread callingThread = Thread.currentThread();
+            assertEquals(
+                "simulated exception in sendRequest",
+                safeAwaitAndUnwrapFailure(
+                    IOException.class,
+                    TransportResponse.Empty.class,
+                    l -> nodeA.transportService.sendRequest(
+                        nodeA.getThrowingConnection(),
+                        TestNode.randomActionName(random()),
+                        new EmptyRequest(),
+                        TransportRequestOptions.EMPTY,
+                        new ActionListenerResponseHandler<>(
+                            ActionListener.runBefore(l, () -> assertSame(callingThread, Thread.currentThread())),
+                            unusedReader(),
+                            EsExecutors.DIRECT_EXECUTOR_SERVICE
+                        )
+                    )
+                ).getMessage()
+            );
+        }
     }
 
     public void testInternalSendExceptionForcesExecutionOnHandlerExecutor() {