[fix] Handle TLS close_notify to avoid SslClosedEngineException: SSLEngine closed already (#24986)

Author: lhotari

pulsar-client/src/main/java/org/apache/pulsar/client/impl/ClientCnx.java

@@ -29,7 +29,6 @@
 import io.netty.channel.ChannelHandlerContext;
 import io.netty.channel.EventLoopGroup;
 import io.netty.channel.unix.Errors.NativeIoException;
-import io.netty.handler.ssl.SslHandshakeCompletionEvent;
 import io.netty.util.concurrent.Promise;
 import io.opentelemetry.api.common.Attributes;
 import java.net.InetSocketAddress;
@@ -1442,18 +1441,6 @@ public void close() {
        }
     }
 
-    @Override
-    public void userEventTriggered(ChannelHandlerContext ctx, Object evt) throws Exception {
-        if (evt instanceof SslHandshakeCompletionEvent) {
-            SslHandshakeCompletionEvent sslHandshakeCompletionEvent = (SslHandshakeCompletionEvent) evt;
-            if (sslHandshakeCompletionEvent.cause() != null) {
-                log.warn("{} Got ssl handshake exception {}", ctx.channel(),
-                        sslHandshakeCompletionEvent);
-            }
-        }
-        ctx.fireUserEventTriggered(evt);
-    }
-
     protected void closeWithException(Throwable e) {
        if (ctx != null) {
            connectionFuture.completeExceptionally(e);

pulsar-common/src/main/java/org/apache/pulsar/common/protocol/PulsarDecoder.java

@@ -24,6 +24,8 @@
 import io.netty.channel.ChannelInboundHandlerAdapter;
 import io.netty.channel.ChannelOutboundInvoker;
 import io.netty.handler.codec.haproxy.HAProxyMessage;
+import io.netty.handler.ssl.SslCloseCompletionEvent;
+import io.netty.handler.ssl.SslHandshakeCompletionEvent;
 import org.apache.pulsar.common.api.proto.BaseCommand;
 import org.apache.pulsar.common.api.proto.CommandAck;
 import org.apache.pulsar.common.api.proto.CommandAckResponse;
@@ -749,4 +751,26 @@ protected void handleCommandWatchTopicListClose(CommandWatchTopicListClose comma
     private void writeAndFlush(ChannelOutboundInvoker ctx, ByteBuf cmd) {
         NettyChannelUtil.writeAndFlushWithVoidPromise(ctx, cmd);
     }
+
+    public void userEventTriggered(ChannelHandlerContext ctx, Object evt) {
+        if (evt instanceof SslHandshakeCompletionEvent) {
+            // log handshake failures
+            SslHandshakeCompletionEvent sslHandshakeCompletionEvent = (SslHandshakeCompletionEvent) evt;
+            if (!sslHandshakeCompletionEvent.isSuccess()) {
+                log.warn("[{}] TLS handshake failed. {}", ctx.channel(), sslHandshakeCompletionEvent);
+            }
+        } else if (evt instanceof SslCloseCompletionEvent) {
+            // handle TLS close_notify event and immediately close the channel
+            // this is not handled by Netty by default
+            // See https://datatracker.ietf.org/doc/html/rfc8446#section-6.1 for more details
+            SslCloseCompletionEvent sslCloseCompletionEvent = (SslCloseCompletionEvent) evt;
+            if (sslCloseCompletionEvent.isSuccess() && ctx.channel().isActive()) {
+                if (log.isDebugEnabled()) {
+                    log.debug("[{}] Received a TLS close_notify, closing the channel.", ctx.channel());
+                }
+                ctx.close();
+            }
+        }
+        ctx.fireUserEventTriggered(evt);
+    }
 }