MockTransportService did not call handle exception (#102766) (#102805)

A failure to send the delayed message would previously not call
the handler logic for exceptions. This resulted in some tests
complaining about un-subtracted bytes in IndexingPressure.

Also adding some logging that helped figuring out the situation.

Fixes #92344

Author: kingherc

server/src/main/java/org/elasticsearch/index/IndexingPressure.java

@@ -10,6 +10,7 @@
 
 import org.apache.logging.log4j.LogManager;
 import org.apache.logging.log4j.Logger;
+import org.elasticsearch.common.Strings;
 import org.elasticsearch.common.settings.Setting;
 import org.elasticsearch.common.settings.Settings;
 import org.elasticsearch.common.unit.ByteSizeValue;
@@ -101,12 +102,14 @@ public Releasable markCoordinatingOperationStarted(int operations, long bytes, b
                 false
             );
         }
+        logger.trace(() -> Strings.format("adding [%d] coordinating operations and [%d] bytes", operations, bytes));
         currentCoordinatingBytes.getAndAdd(bytes);
         currentCoordinatingOps.getAndAdd(operations);
         totalCombinedCoordinatingAndPrimaryBytes.getAndAdd(bytes);
         totalCoordinatingBytes.getAndAdd(bytes);
         totalCoordinatingOps.getAndAdd(operations);
         return wrapReleasable(() -> {
+            logger.trace(() -> Strings.format("removing [%d] coordinating operations and [%d] bytes", operations, bytes));
             this.currentCombinedCoordinatingAndPrimaryBytes.getAndAdd(-bytes);
             this.currentCoordinatingBytes.getAndAdd(-bytes);
             this.currentCoordinatingOps.getAndAdd(-operations);
@@ -153,12 +156,14 @@ public Releasable markPrimaryOperationStarted(int operations, long bytes, boolea
                 false
             );
         }
+        logger.trace(() -> Strings.format("adding [%d] primary operations and [%d] bytes", operations, bytes));
         currentPrimaryBytes.getAndAdd(bytes);
         currentPrimaryOps.getAndAdd(operations);
         totalCombinedCoordinatingAndPrimaryBytes.getAndAdd(bytes);
         totalPrimaryBytes.getAndAdd(bytes);
         totalPrimaryOps.getAndAdd(operations);
         return wrapReleasable(() -> {
+            logger.trace(() -> Strings.format("removing [%d] primary operations and [%d] bytes", operations, bytes));
             this.currentCombinedCoordinatingAndPrimaryBytes.getAndAdd(-bytes);
             this.currentPrimaryBytes.getAndAdd(-bytes);
             this.currentPrimaryOps.getAndAdd(-operations);

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

@@ -107,7 +107,7 @@ interface Connection extends Closeable, RefCounted {
 
         /**
          * Sends the request to the node this connection is associated with
-         * @param requestId see {@link ResponseHandlers#add(ResponseContext)} for details
+         * @param requestId see {@link ResponseHandlers#add(TransportResponseHandler, Connection, String)} for details
          * @param action the action to execute
          * @param request the request to send
          * @param options request options to apply
@@ -163,35 +163,15 @@ default Object getCacheKey() {
     }
 
     /**
-     * This class represents a response context that encapsulates the actual response handler, the action and the connection it was
-     * executed on.
+     * This class represents a response context that encapsulates the actual response handler, the action. the connection it was
+     * executed on, and the request ID.
      */
-    final class ResponseContext<T extends TransportResponse> {
-
-        private final TransportResponseHandler<T> handler;
-
-        private final Connection connection;
-
-        private final String action;
-
-        ResponseContext(TransportResponseHandler<T> handler, Connection connection, String action) {
-            this.handler = handler;
-            this.connection = connection;
-            this.action = action;
-        }
-
-        public TransportResponseHandler<T> handler() {
-            return handler;
-        }
-
-        public Connection connection() {
-            return this.connection;
-        }
-
-        public String action() {
-            return this.action;
-        }
-    }
+    record ResponseContext<T extends TransportResponse>(
+        TransportResponseHandler<T> handler,
+        Connection connection,
+        String action,
+        long requestId
+    ) {};
 
     /**
      * This class is a registry that allows
@@ -218,14 +198,19 @@ public ResponseContext<? extends TransportResponse> remove(long requestId) {
 
         /**
          * Adds a new response context and associates it with a new request ID.
-         * @return the new request ID
+         * @return the new response context
          * @see Connection#sendRequest(long, String, TransportRequest, TransportRequestOptions)
          */
-        public long add(ResponseContext<? extends TransportResponse> holder) {
+        public ResponseContext<? extends TransportResponse> add(
+            TransportResponseHandler<? extends TransportResponse> handler,
+            Connection connection,
+            String action
+        ) {
             long requestId = newRequestId();
+            ResponseContext<? extends TransportResponse> holder = new ResponseContext<>(handler, connection, action, requestId);
             ResponseContext<? extends TransportResponse> existing = handlers.put(requestId, holder);
             assert existing == null : "request ID already in use: " + requestId;
-            return requestId;
+            return holder;
         }
 
         /**

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

@@ -358,6 +358,10 @@ protected void doStop() {
                 try {
                     final TransportResponseHandler<?> handler = holderToNotify.handler();
                     final var targetNode = holderToNotify.connection().getNode();
+                    final long requestId = holderToNotify.requestId();
+                    if (tracerLog.isTraceEnabled() && shouldTraceAction(holderToNotify.action())) {
+                        tracerLog.trace("[{}][{}] pruning request for node [{}]", requestId, holderToNotify.action(), targetNode);
+                    }
 
                     assert transport instanceof TcpTransport == false
                         /* other transports (used in tests) may not implement the proper close-connection behaviour. TODO fix this. */
@@ -921,7 +925,7 @@ private <T extends TransportResponse> void sendRequestInternal(
         Supplier<ThreadContext.StoredContext> storedContextSupplier = threadPool.getThreadContext().newRestorableContext(true);
         ContextRestoreResponseHandler<T> responseHandler = new ContextRestoreResponseHandler<>(storedContextSupplier, handler);
         // TODO we can probably fold this entire request ID dance into connection.sendRequest but it will be a bigger refactoring
-        final long requestId = responseHandlers.add(new Transport.ResponseContext<>(responseHandler, connection, action));
+        final long requestId = responseHandlers.add(responseHandler, connection, action).requestId();
         request.setRequestId(requestId);
         final TimeoutHandler timeoutHandler;
         if (options.timeout() != null) {
@@ -950,7 +954,7 @@ private <T extends TransportResponse> void sendRequestInternal(
         }
     }
 
-    private void handleInternalSendException(
+    protected void handleInternalSendException(
         String action,
         DiscoveryNode node,
         long requestId,
@@ -985,6 +989,9 @@ public void onFailure(Exception e) {
 
             @Override
             protected void doRun() {
+                if (tracerLog.isTraceEnabled() && shouldTraceAction(action)) {
+                    tracerLog.trace("[{}][{}] failed to send request to node [{}]", requestId, action, node);
+                }
                 contextToNotify.handler().handleException(sendRequestException);
             }
         });
@@ -1302,6 +1309,14 @@ public void onConnectionClosed(Transport.Connection connection) {
             @Override
             public void doRun() {
                 for (Transport.ResponseContext<?> holderToNotify : pruned) {
+                    if (tracerLog.isTraceEnabled() && shouldTraceAction(holderToNotify.action())) {
+                        tracerLog.trace(
+                            "[{}][{}] pruning request because connection to node [{}] closed",
+                            holderToNotify.requestId(),
+                            holderToNotify.action(),
+                            connection.getNode()
+                        );
+                    }
                     holderToNotify.handler().handleException(new NodeDisconnectedException(connection.getNode(), holderToNotify.action()));
                 }
             }

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

@@ -137,7 +137,7 @@ public void testRequestAndResponse() throws Exception {
         AtomicReference<Exception> exceptionCaptor = new AtomicReference<>();
         AtomicReference<TransportChannel> channelCaptor = new AtomicReference<>();
 
-        long requestId = responseHandlers.add(new Transport.ResponseContext<>(new TransportResponseHandler<TestResponse>() {
+        long requestId = responseHandlers.add(new TransportResponseHandler<TestResponse>() {
             @Override
             public Executor executor(ThreadPool threadPool) {
                 return TransportResponseHandler.TRANSPORT_WORKER;
@@ -157,7 +157,7 @@ public void handleException(TransportException exp) {
             public TestResponse read(StreamInput in) throws IOException {
                 return new TestResponse(in);
             }
-        }, null, action));
+        }, null, action).requestId();
         RequestHandlerRegistry<TestRequest> registry = new RequestHandlerRegistry<>(
             action,
             TestRequest::new,

test/framework/src/main/java/org/elasticsearch/test/transport/MockTransportService.java

@@ -31,6 +31,7 @@
 import org.elasticsearch.common.util.concurrent.RunOnce;
 import org.elasticsearch.core.IOUtils;
 import org.elasticsearch.core.Nullable;
+import org.elasticsearch.core.Strings;
 import org.elasticsearch.core.TimeValue;
 import org.elasticsearch.indices.breaker.NoneCircuitBreakerService;
 import org.elasticsearch.node.Node;
@@ -546,11 +547,23 @@ public void sendRequest(
                 final RunOnce runnable = new RunOnce(new AbstractRunnable() {
                     @Override
                     public void onFailure(Exception e) {
-                        logger.debug("failed to send delayed request", e);
+                        logger.debug(
+                            () -> Strings.format(
+                                "[%d][%s] failed to send delayed request to node [%s]",
+                                requestId,
+                                action,
+                                connection.getNode()
+                            ),
+                            e
+                        );
+                        handleInternalSendException(action, connection.getNode(), requestId, null, e);
                     }
 
                     @Override
                     protected void doRun() throws IOException {
+                        logger.debug(
+                            () -> Strings.format("[%d][%s] sending delayed request to node [%s]", requestId, action, connection.getNode())
+                        );
                         connection.sendRequest(requestId, action, clonedRequest, options);
                     }
                 });
@@ -561,6 +574,15 @@ protected void doRun() throws IOException {
                         runnable.run();
                     } else {
                         requestsToSendWhenCleared.add(runnable);
+                        logger.debug(
+                            () -> Strings.format(
+                                "[%d][%s] delaying sending request to node [%s] by [%s]",
+                                requestId,
+                                action,
+                                connection.getNode(),
+                                delay
+                            )
+                        );
                         threadPool.schedule(runnable, delay, threadPool.generic());
                     }
                 }