Handle protocol version mismatch in Netty IO handler

Author: acogoluegnes

pom.xml

@@ -779,6 +779,7 @@
               <include>src/main/java/com/rabbitmq/client/observation/**/*.java</include>
               <include>src/test/java/com/rabbitmq/client/test/functional/MicrometerObservationCollectorMetrics.java</include>
               <include>src/test/java/com/rabbitmq/client/test/NettyTest.java</include>
+              <include>src/test/java/com/rabbitmq/client/test/ProtocolVersionMismatch.java</include>
             </includes>
             <googleJavaFormat>
               <version>${google-java-format.version}</version>

src/main/java/com/rabbitmq/client/ConnectionFactory.java

@@ -813,7 +813,7 @@ public ConnectionFactory setExceptionHandler(ExceptionHandler exceptionHandler)
   public boolean isSSL() {
     return getSocketFactory() instanceof SSLSocketFactory
         || sslContextFactory != null
-        || this.netty().isTls();
+        || this.nettyConf.isTls();
   }
 
   /**

src/main/java/com/rabbitmq/client/impl/AMQConnection.java

@@ -493,6 +493,7 @@ public void start()
 
         // We can now respond to errors having finished tailoring the connection
         this._inConnectionNegotiation = false;
+        this._frameHandler.finishConnectionNegotiation();
     }
 
     protected ChannelManager instantiateChannelManager(int channelMax, ThreadFactory threadFactory) {

src/main/java/com/rabbitmq/client/impl/Frame.java

@@ -90,7 +90,6 @@ public static Frame readFrom(DataInputStream is, int maxPayloadSize) throws IOEx
         try {
             type = is.readUnsignedByte();
         } catch (SocketTimeoutException ste) {
-            // System.err.println("Timed out waiting for a frame.");
             return null; // failed
         }
 

src/main/java/com/rabbitmq/client/impl/FrameHandler.java

@@ -54,6 +54,10 @@ default boolean internalHearbeat() {
 
     void initialize(AMQConnection connection);
 
+    default void finishConnectionNegotiation() {
+
+    }
+
     /**
      * Read a {@link Frame} from the underlying data connection.
      * @return an incoming Frame, or null if there is none
@@ -77,4 +81,5 @@ default boolean internalHearbeat() {
 
     /** Close the underlying data connection (complaint not permitted). */
     void close();
+
 }

src/main/java/com/rabbitmq/client/impl/NettyFrameHandlerFactory.java

@@ -44,6 +44,8 @@
 import io.netty.handler.timeout.IdleStateEvent;
 import io.netty.handler.timeout.IdleStateHandler;
 import io.netty.handler.timeout.ReadTimeoutHandler;
+import java.io.ByteArrayInputStream;
+import java.io.DataInputStream;
 import java.io.IOException;
 import java.net.InetAddress;
 import java.net.InetSocketAddress;
@@ -137,6 +139,8 @@ private static final class NettyFrameHandler implements FrameHandler {
         LengthFieldBasedFrameDecoder.class.getSimpleName();
     private static final String HANDLER_READ_TIMEOUT = ReadTimeoutHandler.class.getSimpleName();
     private static final String HANDLER_IDLE_STATE = IdleStateHandler.class.getSimpleName();
+    private static final String HANDLER_PROTOCOL_VERSION_MISMATCH =
+        ProtocolVersionMismatchHandler.class.getSimpleName();
     private static final byte[] HEADER =
         new byte[] {
           'A', 'M', 'Q', 'P', 0, AMQP.PROTOCOL.MAJOR, AMQP.PROTOCOL.MINOR, AMQP.PROTOCOL.REVISION
@@ -193,6 +197,8 @@ public void initChannel(SocketChannel ch) {
                       HANDLER_FLUSH_CONSOLIDATION,
                       new FlushConsolidationHandler(
                           FlushConsolidationHandler.DEFAULT_EXPLICIT_FLUSH_AFTER_FLUSHES, true));
+              ch.pipeline()
+                  .addLast(HANDLER_PROTOCOL_VERSION_MISMATCH, new ProtocolVersionMismatchHandler());
               ch.pipeline()
                   .addLast(
                       HANDLER_FRAME_DECODER,
@@ -284,6 +290,11 @@ public void initialize(AMQConnection connection) {
       this.handler.connection = connection;
     }
 
+    @Override
+    public void finishConnectionNegotiation() {
+      maybeRemoveHandler(HANDLER_PROTOCOL_VERSION_MISMATCH);
+    }
+
     @Override
     public Frame readFrame() {
       throw new UnsupportedOperationException();
@@ -412,11 +423,7 @@ public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception
         if (noProblem
             && (!this.connection.isRunning() || this.connection.hasBrokerInitiatedShutdown())) {
           // looks like the frame was Close-Ok or Close
-          ctx.executor()
-              .submit(
-                  () -> {
-                    this.connection.doFinalShutdown();
-                  });
+          ctx.executor().submit(() -> this.connection.doFinalShutdown());
         }
       } finally {
         m.release();
@@ -503,4 +510,23 @@ private CountDownLatch writableLatch() {
       return this.writableLatch.get();
     }
   }
+
+  private static final class ProtocolVersionMismatchHandler extends ChannelInboundHandlerAdapter {
+
+    @Override
+    public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
+      ByteBuf b = (ByteBuf) msg;
+      if (b.readByte() == 'A') {
+        // likely an AMQP header that indicates a protocol version mismatch
+        // the header is small, we read everything in memory and use the Frame class
+        int toRead = Math.min(b.readableBytes(), NettyFrameHandler.HEADER.length - 1);
+        byte[] header = new byte[toRead];
+        b.readBytes(header);
+        Frame.protocolVersionMismatch(new DataInputStream(new ByteArrayInputStream(header)));
+      } else {
+        b.readerIndex(0);
+        ctx.fireChannelRead(msg);
+      }
+    }
+  }
 }

src/test/java/com/rabbitmq/client/test/ClientTestSuite.java

@@ -75,7 +75,8 @@
     AMQConnectionRefreshCredentialsTest.class,
     ValueWriterTest.class,
     BlockedConnectionTest.class,
-    NettyTest.class
+    NettyTest.class,
+    ProtocolVersionMismatch.class
 })
 public class ClientTestSuite {
 

src/test/java/com/rabbitmq/client/test/FrameTest.java

@@ -72,12 +72,12 @@ private void checkWrittenChunks(int totalFrameSize, AccumulatorWritableByteChann
 
     private static class AccumulatorWritableByteChannel implements WritableByteChannel {
 
-        List<byte[]> chunks = new ArrayList<byte[]>();
+        List<byte[]> chunks = new ArrayList<>();
 
         Random random = new Random();
 
         @Override
-        public int write(ByteBuffer src) throws IOException {
+        public int write(ByteBuffer src) {
             int remaining = src.remaining();
             if(remaining > 0) {
                 int toRead = random.nextInt(remaining) + 1;
@@ -88,7 +88,6 @@ public int write(ByteBuffer src) throws IOException {
             } else {
                 return remaining;
             }
-
         }
 
         @Override
@@ -97,9 +96,7 @@ public boolean isOpen() {
         }
 
         @Override
-        public void close() throws IOException {
-
-        }
+        public void close() { }
     }
 
     public static void drain(WritableByteChannel channel, ByteBuffer buffer) throws IOException {

src/test/java/com/rabbitmq/client/test/ProtocolVersionMismatch.java

@@ -0,0 +1,92 @@
+// Copyright (c) 2025 Broadcom. All Rights Reserved.
+// The term "Broadcom" refers to Broadcom Inc. and/or its subsidiaries.
+//
+// This software, the RabbitMQ Java client library, is triple-licensed under the
+// Mozilla Public License 2.0 ("MPL"), the GNU General Public License version 2
+// ("GPL") and the Apache License version 2 ("ASL"). For the MPL, please see
+// LICENSE-MPL-RabbitMQ. For the GPL, please see LICENSE-GPL2.  For the ASL,
+// please see LICENSE-APACHE2.
+//
+// This software is distributed on an "AS IS" basis, WITHOUT WARRANTY OF ANY KIND,
+// either express or implied. See the LICENSE file for specific language governing
+// rights and limitations of this software.
+//
+// If you have any questions regarding licensing, please contact us at
+// [email protected].
+package com.rabbitmq.client.test;
+
+import static org.assertj.core.api.Assertions.assertThatThrownBy;
+
+import com.rabbitmq.client.ConnectionFactory;
+import com.rabbitmq.client.MalformedFrameException;
+import io.netty.bootstrap.ServerBootstrap;
+import io.netty.buffer.ByteBuf;
+import io.netty.channel.ChannelFutureListener;
+import io.netty.channel.ChannelHandlerContext;
+import io.netty.channel.ChannelInboundHandlerAdapter;
+import io.netty.channel.ChannelInitializer;
+import io.netty.channel.EventLoopGroup;
+import io.netty.channel.MultiThreadIoEventLoopGroup;
+import io.netty.channel.nio.NioIoHandler;
+import io.netty.channel.socket.SocketChannel;
+import io.netty.channel.socket.nio.NioServerSocketChannel;
+import io.netty.util.ReferenceCountUtil;
+import java.util.concurrent.TimeUnit;
+import org.junit.jupiter.params.ParameterizedTest;
+import org.junit.jupiter.params.provider.MethodSource;
+
+public class ProtocolVersionMismatch {
+
+  @ParameterizedTest
+  @MethodSource("com.rabbitmq.client.test.TestUtils#ioLayers")
+  void connectionShouldFailWithProtocolVersionMismatch(String ioLayer) throws Exception {
+    int port = TestUtils.randomNetworkPort();
+    try (SimpleServer ignored = new SimpleServer(port)) {
+      ConnectionFactory cf = TestUtils.connectionFactory();
+      TestUtils.setIoLayer(cf, ioLayer);
+      cf.setPort(port);
+      assertThatThrownBy(cf::newConnection).hasRootCauseInstanceOf(MalformedFrameException.class);
+    }
+  }
+
+  private static class SimpleServer implements AutoCloseable {
+
+    private final EventLoopGroup elp =
+        new MultiThreadIoEventLoopGroup(1, NioIoHandler.newFactory());
+
+    private static final byte[] AMQP_HEADER = new byte[] {'A', 'M', 'Q', 'P', 0, 1, 0, 0};
+
+    private SimpleServer(int port) throws InterruptedException {
+      ServerBootstrap b = new ServerBootstrap();
+      b.group(elp);
+      b.channel(NioServerSocketChannel.class);
+      b.childHandler(
+              new ChannelInitializer<SocketChannel>() {
+                @Override
+                protected void initChannel(SocketChannel ch) {
+                  ch.pipeline()
+                      .addLast(
+                          new ChannelInboundHandlerAdapter() {
+                            @Override
+                            public void channelRead(ChannelHandlerContext ctx, Object msg) {
+                              // discard the data
+                              ReferenceCountUtil.release(msg);
+                              ByteBuf b = ctx.alloc().buffer(AMQP_HEADER.length);
+                              b.writeBytes(AMQP_HEADER);
+                              ctx.channel()
+                                  .writeAndFlush(b)
+                                  .addListener(ChannelFutureListener.CLOSE);
+                            }
+                          });
+                }
+              })
+          .validate();
+      b.bind(port).sync();
+    }
+
+    @Override
+    public void close() {
+      this.elp.shutdownGracefully(0, 0, TimeUnit.SECONDS);
+    }
+  }
+}

src/test/java/com/rabbitmq/client/test/TestUtils.java

@@ -62,9 +62,6 @@ public class TestUtils {
     public static ConnectionFactory connectionFactory() {
         ConnectionFactory connectionFactory = new ConnectionFactory();
         setIoLayer(connectionFactory);
-        if (isNetty()) {
-            connectionFactory.netty().eventLoopGroup(EVENT_LOOP_GROUP.get());
-        }
         return connectionFactory;
     }
 
@@ -98,6 +95,9 @@ private static boolean isNetty(String layer) {
 
     public static void setIoLayer(ConnectionFactory cf) {
         setIoLayer(cf, IO_LAYER);
+        if (isNetty()) {
+            cf.netty().eventLoopGroup(EVENT_LOOP_GROUP.get());
+        }
     }
 
     public static void setIoLayer(ConnectionFactory cf, String layer) {