Addressed review feedback:

- renamed channelError and onException
- removed dead channel error logging code
- used expectThrows pattern in tests
- added closeListener test
- added assert to onFailure branch of netty channel future/listener adapter
- de-duplicated throwable/exception adapting code
- log transport errors at debug

Author: schase-es

modules/transport-netty4/src/main/java/org/elasticsearch/transport/netty4/Netty4TcpChannel.java

@@ -34,7 +34,8 @@ public class Netty4TcpChannel implements TcpChannel {
     private final ListenableFuture<Void> closeContext = new ListenableFuture<>();
     private final ChannelStats stats = new ChannelStats();
     private final boolean rstOnClose;
-    private volatile Exception channelError = null;
+    // Exception causing a close, reported to the closeContext listener
+    private volatile Exception closeException = null;
 
     Netty4TcpChannel(Channel channel, boolean isServer, String profile, boolean rstOnClose, ChannelFuture connectFuture) {
         this.channel = channel;
@@ -43,16 +44,21 @@ public class Netty4TcpChannel implements TcpChannel {
         this.connectContext = new ListenableFuture<>();
         this.rstOnClose = rstOnClose;
         addListener(connectFuture, connectContext);
-        // The netty closeFuture is programmed to never return an exception.
-        // This listener takes close-causing exceptions reported during the
-        // channel lifetime, and reports them to the closeListener.
-        addListener(this.channel.closeFuture(), ActionListener.running(() -> {
-            if (channelError != null) {
-                closeContext.onFailure(channelError);
-            } else {
-                closeContext.onResponse(null);
+        addListener(this.channel.closeFuture(), new ActionListener<>() {
+            @Override
+            public void onResponse(Void ignored) {
+                if (closeException != null) {
+                    closeContext.onFailure(closeException);
+                } else {
+                    closeContext.onResponse(null);
+                }
+            }
+
+            @Override
+            public void onFailure(Exception e) {
+                assert false : new AssertionError("netty channel closeFuture should never report a failure");
             }
-        }));
+        });
     }
 
     @Override
@@ -106,8 +112,8 @@ public void addConnectListener(ActionListener<Void> listener) {
     }
 
     @Override
-    public void onException(Exception e) {
-        channelError = e;
+    public void setCloseException(Exception e) {
+        closeException = e;
     }
 
     @Override

modules/transport-netty4/src/main/java/org/elasticsearch/transport/netty4/Netty4Transport.java

@@ -28,7 +28,6 @@
 import org.apache.lucene.util.BytesRef;
 import org.elasticsearch.ExceptionsHelper;
 import org.elasticsearch.TransportVersion;
-import org.elasticsearch.action.ActionListener;
 import org.elasticsearch.cluster.node.DiscoveryNode;
 import org.elasticsearch.common.io.stream.NamedWriteableRegistry;
 import org.elasticsearch.common.network.NetworkService;
@@ -55,7 +54,6 @@
 import java.util.Map;
 
 import static org.elasticsearch.common.util.concurrent.ConcurrentCollections.newConcurrentMap;
-import static org.elasticsearch.core.Strings.format;
 import static org.elasticsearch.transport.RemoteClusterPortSettings.REMOTE_CLUSTER_PROFILE;
 import static org.elasticsearch.transport.RemoteClusterPortSettings.REMOTE_CLUSTER_SERVER_ENABLED;
 
@@ -282,7 +280,6 @@ protected Netty4TcpChannel initiateChannel(DiscoveryNode node, ConnectionProfile
             rstOnClose,
             connectFuture
         );
-        addClosedExceptionLogger(nettyChannel);
         channel.attr(CHANNEL_KEY).set(nettyChannel);
 
         return nettyChannel;
@@ -310,6 +307,14 @@ protected void stopInternal() {
         }, serverBootstraps::clear, () -> clientBootstrap = null);
     }
 
+    private Exception exceptionFromThrowable(Throwable cause) {
+        if (cause instanceof Error) {
+            return new Exception(cause);
+        } else {
+            return (Exception) cause;
+        }
+    }
+
     protected class ClientChannelInitializer extends ChannelInitializer<Channel> {
 
         @Override
@@ -322,11 +327,7 @@ protected void initChannel(Channel ch) throws Exception {
         @Override
         public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
             Netty4TcpChannel channel = ctx.channel().attr(CHANNEL_KEY).get();
-            if (cause instanceof Error) {
-                channel.onException(new Exception(cause));
-            } else {
-                channel.onException((Exception) cause);
-            }
+            channel.setCloseException(exceptionFromThrowable(cause));
             ExceptionsHelper.maybeDieOnAnotherThread(cause);
             super.exceptionCaught(ctx, cause);
         }
@@ -347,7 +348,6 @@ protected void initChannel(Channel ch) throws Exception {
             assert ch instanceof Netty4NioSocketChannel;
             NetUtils.tryEnsureReasonableKeepAliveConfig(((Netty4NioSocketChannel) ch).javaChannel());
             Netty4TcpChannel nettyTcpChannel = new Netty4TcpChannel(ch, true, name, rstOnClose, ch.newSucceededFuture());
-            addClosedExceptionLogger(nettyTcpChannel);
             ch.attr(CHANNEL_KEY).set(nettyTcpChannel);
             setupPipeline(ch, isRemoteClusterServerChannel);
             serverAcceptedChannel(nettyTcpChannel);
@@ -356,11 +356,7 @@ protected void initChannel(Channel ch) throws Exception {
         @Override
         public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
             Netty4TcpChannel channel = ctx.channel().attr(CHANNEL_KEY).get();
-            if (cause instanceof Error) {
-                channel.onException(new Exception(cause));
-            } else {
-                channel.onException((Exception) cause);
-            }
+            channel.setCloseException(exceptionFromThrowable(cause));
             ExceptionsHelper.maybeDieOnAnotherThread(cause);
             super.exceptionCaught(ctx, cause);
         }
@@ -396,14 +392,6 @@ protected InboundPipeline getInboundPipeline(Channel ch, boolean isRemoteCluster
         );
     }
 
-    private static void addClosedExceptionLogger(Netty4TcpChannel channel) {
-        channel.addCloseListener(ActionListener.wrap((ignored) -> {}, (e) -> {
-            if (logger.isDebugEnabled()) {
-                logger.debug(format("exception while closing channel: %s", channel), e);
-            }
-        }));
-    }
-
     @ChannelHandler.Sharable
     private static class ServerChannelExceptionHandler extends ChannelInboundHandlerAdapter {
 

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

@@ -390,7 +390,7 @@ private void handleHandshakeRequest(TcpChannel channel, InboundMessage message)
                 () -> "error processing handshake version [" + header.getVersion() + "] received on [" + channel + "], closing channel",
                 e
             );
-            channel.onException(e);
+            channel.setCloseException(e);
             channel.close();
         }
     }

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

@@ -168,7 +168,7 @@ void sendResponse(
                     ),
                     ex
                 );
-                channel.onException(ex);
+                channel.setCloseException(ex);
                 channel.close();
             } else {
                 sendErrorResponse(transportVersion, channel, requestId, action, responseStatsConsumer, ex);
@@ -205,7 +205,7 @@ void sendErrorResponse(
         } catch (Exception sendException) {
             sendException.addSuppressed(error);
             logger.error(() -> format("Failed to send error response on channel [%s], closing channel", channel), sendException);
-            channel.onException(sendException);
+            channel.setCloseException(sendException);
             channel.close();
         }
     }
@@ -433,7 +433,7 @@ private void maybeLogSlowMessage(boolean success) {
                 }
             });
         } catch (RuntimeException ex) {
-            channel.onException(ex);
+            channel.setCloseException(ex);
             Releasables.closeExpectNoException(() -> listener.onFailure(ex), () -> CloseableChannel.closeChannel(channel));
             throw ex;
         }

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

@@ -67,11 +67,11 @@ public interface TcpChannel extends CloseableChannel {
     void addConnectListener(ActionListener<Void> listener);
 
     /**
-     * Report an exception on this channel
+     * Report a close-causing exception on this channel
      *
      * @param e the exception
      */
-    void onException(Exception e);
+    void setCloseException(Exception e);
 
     /**
      * Returns stats about this channel

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

@@ -773,7 +773,7 @@ static void handleException(TcpChannel channel, Exception e, Lifecycle lifecycle
             }
         } finally {
             if (closeChannel) {
-                channel.onException(e);
+                channel.setCloseException(e);
                 CloseableChannel.closeChannel(channel);
             }
         }
@@ -1206,7 +1206,7 @@ public void onResponse(Void ignored) {
         @Override
         public void onFailure(Exception e) {
             long closeTimeMillis = threadPool.relativeTimeInMillis();
-            logger.info(
+            logger.debug(
                 () -> format(
                     "closed transport connection [{}] to [{}] with age [{}ms], exception:",
                     connectionId,

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

@@ -55,6 +55,7 @@
 import static org.hamcrest.Matchers.equalTo;
 import static org.hamcrest.Matchers.instanceOf;
 import static org.hamcrest.Matchers.notNullValue;
+import static org.hamcrest.Matchers.nullValue;
 import static org.mockito.ArgumentMatchers.any;
 import static org.mockito.ArgumentMatchers.eq;
 import static org.mockito.Mockito.doAnswer;
@@ -107,6 +108,8 @@ public void onNodeDisconnected(DiscoveryNode node, Transport.Connection connecti
 
         DiscoveryNode node = DiscoveryNodeUtils.create("", new TransportAddress(InetAddress.getLoopbackAddress(), 0));
         Transport.Connection connection = new TestConnect(node);
+        PlainActionFuture<Void> closeListener = new PlainActionFuture<>();
+        connection.addCloseListener(closeListener);
         doAnswer(invocationOnMock -> {
             @SuppressWarnings("unchecked")
             ActionListener<Transport.Connection> listener = (ActionListener<Transport.Connection>) invocationOnMock.getArguments()[2];
@@ -137,6 +140,7 @@ public void onNodeDisconnected(DiscoveryNode node, Transport.Connection connecti
             connection.close();
         }
         assertTrue(connection.isClosed());
+        assertThat(closeListener.actionGet(), nullValue());
         assertEquals(0, connectionManager.size());
         assertEquals(1, nodeConnectedCount.get());
         assertEquals(1, nodeDisconnectedCount.get());

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

@@ -260,10 +260,7 @@ public void testClosesChannelOnErrorInHandshake() throws Exception {
             handler.inboundMessage(channel, requestMessage);
             assertTrue(isClosed.get());
             assertTrue(closeListener.isDone());
-            try {
-                closeListener.get();
-                assert false : "channel should have an exception reported";
-            } catch (Exception e) {}
+            expectThrows(Exception.class, () -> closeListener.get());
             assertNull(channel.getMessageCaptor().get());
             mockLog.assertAllExpectationsMatched();
         }

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

@@ -438,10 +438,7 @@ public void writeTo(StreamOutput out) {
         assertNull(channel.getListenerCaptor().get());
         assertFalse(channel.isOpen());
         assertTrue(closeListener.isDone());
-        try {
-            closeListener.get();
-            assert false : "channel should have an exception reported";
-        } catch (Exception e) {}
+        expectThrows(Exception.class, () -> closeListener.get());
     }
 
     public void testFailToSendHandshakeResponse() {
@@ -485,10 +482,7 @@ public void onResponseSent(long requestId, String action, Exception error) {
         assertTrue(response.released.get());
         assertFalse(channel.isOpen());
         assertTrue(closeListener.isDone());
-        try {
-            closeListener.get();
-            assert false : "channel should have an exception reported";
-        } catch (Exception e) {}
+        expectThrows(Exception.class, () -> closeListener.get());
     }
 
     public void testFailToSendErrorResponse() {
@@ -534,10 +528,7 @@ public void onResponseSent(long requestId, String action, Exception error) {
         assertNull(channel.getMessageCaptor().get());
         assertNull(channel.getListenerCaptor().get());
         assertTrue(closeListener.isDone());
-        try {
-            closeListener.get();
-            assert false : "channel should have an exception reported";
-        } catch (Exception e) {}
+        expectThrows(Exception.class, () -> closeListener.get());
     }
 
     /**

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

@@ -605,10 +605,7 @@ private void testExceptionHandling(
 
             if (expectClosed) {
                 assertTrue(listener.isDone());
-                try {
-                    listener.get();
-                    assert false : "channel should have an exception reported";
-                } catch (Exception e) {}
+                expectThrows(Exception.class, () -> listener.get());
             } else {
                 assertFalse(listener.isDone());
             }