Remove unneeded Throwable handling in nio (#27412)

This is related to #27260. In the nio transport work we do not catch or
handle `Throwable`. There are a few places where we have exception
handlers that accept `Throwable`. This commit removes those cases.

test/framework/src/main/java/org/elasticsearch/transport/nio/NioTransport.java

@@ -184,9 +184,7 @@ protected SocketEventHandler getSocketEventHandler() {
         return new SocketEventHandler(logger, this::exceptionCaught, openChannels);
     }
 
-    final void exceptionCaught(NioSocketChannel channel, Throwable cause) {
-        final Throwable unwrapped = ExceptionsHelper.unwrap(cause, ElasticsearchException.class);
-        final Throwable t = unwrapped != null ? unwrapped : cause;
-        onException(channel, t instanceof Exception ? (Exception) t : new ElasticsearchException(t));
+    final void exceptionCaught(NioSocketChannel channel, Exception exception) {
+        onException(channel, exception);
     }
 }

test/framework/src/main/java/org/elasticsearch/transport/nio/SocketEventHandler.java

@@ -34,10 +34,10 @@
  */
 public class SocketEventHandler extends EventHandler {
 
-    private final BiConsumer<NioSocketChannel, Throwable> exceptionHandler;
+    private final BiConsumer<NioSocketChannel, Exception> exceptionHandler;
     private final Logger logger;
 
-    public SocketEventHandler(Logger logger, BiConsumer<NioSocketChannel, Throwable> exceptionHandler, OpenChannels openChannels) {
+    public SocketEventHandler(Logger logger, BiConsumer<NioSocketChannel, Exception> exceptionHandler, OpenChannels openChannels) {
         super(logger, openChannels);
         this.exceptionHandler = exceptionHandler;
         this.logger = logger;

test/framework/src/main/java/org/elasticsearch/transport/nio/TcpReadHandler.java

@@ -32,10 +32,9 @@ public TcpReadHandler(NioTransport transport) {
         this.transport = transport;
     }
 
-    public void handleMessage(BytesReference reference, NioSocketChannel channel, String profileName,
-                              int messageBytesLength) {
+    public void handleMessage(BytesReference reference, NioSocketChannel channel, int messageBytesLength) {
         try {
-            transport.messageReceived(reference, channel, profileName, channel.getRemoteAddress(), messageBytesLength);
+            transport.messageReceived(reference, channel, channel.getProfile(), channel.getRemoteAddress(), messageBytesLength);
         } catch (IOException e) {
             handleException(channel, e);
         }

test/framework/src/main/java/org/elasticsearch/transport/nio/channel/TcpReadContext.java

@@ -80,7 +80,7 @@ public int read() throws IOException {
 
                 // A message length of 6 bytes it is just a ping. Ignore for now.
                 if (messageLengthWithHeader != 6) {
-                    handler.handleMessage(messageWithoutHeader, channel, channel.getProfile(), messageWithoutHeader.length());
+                    handler.handleMessage(messageWithoutHeader, channel, messageWithoutHeader.length());
                 }
             } catch (Exception e) {
                 handler.handleException(channel, e);

test/framework/src/test/java/org/elasticsearch/transport/nio/SocketEventHandlerTests.java

@@ -43,7 +43,7 @@
 
 public class SocketEventHandlerTests extends ESTestCase {
 
-    private BiConsumer<NioSocketChannel, Throwable> exceptionHandler;
+    private BiConsumer<NioSocketChannel, Exception> exceptionHandler;
 
     private SocketEventHandler handler;
     private NioSocketChannel channel;

test/framework/src/test/java/org/elasticsearch/transport/nio/TestingSocketEventHandler.java

@@ -30,11 +30,8 @@
 
 public class TestingSocketEventHandler extends SocketEventHandler {
 
-    private final Logger logger;
-
-    public TestingSocketEventHandler(Logger logger, BiConsumer<NioSocketChannel, Throwable> exceptionHandler, OpenChannels openChannels) {
+    public TestingSocketEventHandler(Logger logger, BiConsumer<NioSocketChannel, Exception> exceptionHandler, OpenChannels openChannels) {
         super(logger, exceptionHandler, openChannels);
-        this.logger = logger;
     }
 
     private Set<NioSocketChannel> hasConnectedMap = Collections.newSetFromMap(new WeakHashMap<>());

test/framework/src/test/java/org/elasticsearch/transport/nio/channel/TcpReadContextTests.java

@@ -72,7 +72,7 @@ public void testSuccessfulRead() throws IOException {
 
         readContext.read();
 
-        verify(handler).handleMessage(new BytesArray(bytes), channel, PROFILE, messageLength);
+        verify(handler).handleMessage(new BytesArray(bytes), channel, messageLength);
         assertEquals(1024 * 16, bufferCapacity.get());
 
         BytesArray bytesArray = new BytesArray(new byte[10]);
@@ -110,7 +110,7 @@ public void testPartialRead() throws IOException {
         assertEquals(1024 * 16 - fullPart1.length, bufferCapacity.get());
 
         CompositeBytesReference reference = new CompositeBytesReference(new BytesArray(part1), new BytesArray(part2));
-        verify(handler).handleMessage(reference, channel, PROFILE, messageLength + messageLength);
+        verify(handler).handleMessage(reference, channel, messageLength + messageLength);
     }
 
     public void testReadThrowsIOException() throws IOException {