Describe closing channel in exception message (#133632)

Today we report having disconnected from a node along with the root
cause, typically a `SocketException`, but for troubleshooting purposes
we may also need to identify the exact TCP channel which was affected by
the exception. This commit adds a `NodeDisconnectedException` wrapper to
add the additional details.

Author: DaveCTurner

modules/transport-netty4/src/internalClusterTest/java/org/elasticsearch/transport/netty4/ESLoggingHandlerIT.java

@@ -10,11 +10,13 @@
 package org.elasticsearch.transport.netty4;
 
 import org.apache.logging.log4j.Level;
+import org.apache.logging.log4j.core.LogEvent;
 import org.elasticsearch.ESNetty4IntegTestCase;
 import org.elasticsearch.core.TimeValue;
 import org.elasticsearch.test.ESIntegTestCase;
 import org.elasticsearch.test.MockLog;
 import org.elasticsearch.test.junit.annotations.TestLogging;
+import org.elasticsearch.transport.NodeDisconnectedException;
 import org.elasticsearch.transport.TcpTransport;
 import org.elasticsearch.transport.TransportLogger;
 
@@ -117,7 +119,15 @@ public void testExceptionalDisconnectLogging() throws Exception {
                 TcpTransport.class.getCanonicalName(),
                 Level.DEBUG,
                 ".*closed transport connection \\[[1-9][0-9]*\\] to .* with age \\[[0-9]+ms\\], exception:.*"
-            )
+            ) {
+                @Override
+                public void match(LogEvent event) {
+                    if (event.getThrown() instanceof NodeDisconnectedException nodeDisconnectedException
+                        && nodeDisconnectedException.getMessage().contains("closed exceptionally: Netty4TcpChannel{")) {
+                        super.match(event);
+                    }
+                }
+            }
         );
 
         final String nodeName = internalCluster().startNode();

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

@@ -1134,15 +1134,15 @@ public void onResponse(Void v) {
                         nodeChannels.channels.forEach(ch -> {
                             // Mark the channel init time
                             ch.getChannelStats().markAccessed(relativeMillisTime);
-                            ch.addCloseListener(new ActionListener<Void>() {
+                            ch.addCloseListener(new ActionListener<>() {
                                 @Override
                                 public void onResponse(Void ignored) {
                                     nodeChannels.close();
                                 }
 
                                 @Override
                                 public void onFailure(Exception e) {
-                                    nodeChannels.closeAndFail(e);
+                                    nodeChannels.closeAndFail(new NodeDisconnectedException(node, "closed exceptionally: " + ch, null, e));
                                 }
                             });
                         });