IGNITE-28110 Fix buffer leak in ClientInboundMessageHandler on invalid message (#7733)

Author: ptupitsyn

modules/client-handler/src/integrationTest/java/org/apache/ignite/client/handler/ItClientHandlerTest.java

@@ -580,6 +580,60 @@ public void testServerReturnsAllItsFeatures() throws IOException {
         }
     }
 
+    @Test
+    void testInvalidHandshakeStateDropsConnection() throws Exception {
+        try (var sock = new Socket("127.0.0.1", serverPort)) {
+            OutputStream out = sock.getOutputStream();
+
+            // Magic: IGNI
+            out.write(MAGIC);
+
+            // Send first handshake.
+            try (var packer1 = MessagePack.newDefaultBufferPacker()) {
+                packer1.packInt(0);
+                packer1.packInt(0);
+                packer1.packInt(0);
+                packer1.packInt(7); // Size.
+
+                packer1.packInt(3); // Major
+                packer1.packInt(0); // Minor
+                packer1.packInt(0); // Patch
+
+                packer1.packInt(2); // Client type: general purpose.
+
+                packer1.packBinaryHeader(0); // Features.
+                packer1.packInt(0); // Extensions.
+
+                out.write(packer1.toByteArray());
+            }
+
+            // Second message before handshake completes.
+            // This should trigger "Unexpected message received before handshake completion"
+            try (var packer2 = MessagePack.newDefaultBufferPacker()) {
+                packer2.packInt(0);
+                packer2.packInt(0);
+                packer2.packInt(0);
+                packer2.packInt(7); // Size.
+
+                packer2.packInt(3); // Major
+                packer2.packInt(0); // Minor
+                packer2.packInt(0); // Patch
+
+                packer2.packInt(2); // Client type: general purpose.
+
+                packer2.packBinaryHeader(0); // Features.
+                packer2.packInt(0); // Extensions.
+
+                out.write(packer2.toByteArray());
+            }
+
+            out.flush();
+
+            // Server drops the connection due to invalid message.
+            assertThrows(IOException.class, () -> writeAndFlushLoop(sock));
+        }
+    }
+
     private static void writeAndFlushLoop(Socket socket) throws Exception {
         var stop = System.currentTimeMillis() + 5000;
         var out = socket.getOutputStream();

modules/client-handler/src/main/java/org/apache/ignite/client/handler/ClientInboundMessageHandler.java

@@ -421,28 +421,34 @@ public void channelRead(ChannelHandlerContext ctx, Object msg) {
 
         // Each inbound handler in a pipeline has to release the received messages.
         var unpacker = new ClientMessageUnpacker(byteBuf);
-        metrics.bytesReceivedAdd(byteBuf.readableBytes() + ClientMessageCommon.HEADER_SIZE);
 
-        switch (state) {
-            case STATE_BEFORE_HANDSHAKE:
-                state = STATE_HANDSHAKE_REQUESTED;
-                metrics.bytesReceivedAdd(ClientMessageCommon.MAGIC_BYTES.length);
-                handshake(ctx, unpacker);
+        try {
+            metrics.bytesReceivedAdd(byteBuf.readableBytes() + ClientMessageCommon.HEADER_SIZE);
 
-                break;
+            switch (state) {
+                case STATE_BEFORE_HANDSHAKE:
+                    state = STATE_HANDSHAKE_REQUESTED;
+                    metrics.bytesReceivedAdd(ClientMessageCommon.MAGIC_BYTES.length);
+                    handshake(ctx, unpacker);
 
-            case STATE_HANDSHAKE_REQUESTED:
-                // Handshake is in progress, any messages are not allowed.
-                throw new IgniteException(PROTOCOL_ERR, "Unexpected message received before handshake completion");
+                    break;
 
-            case STATE_HANDSHAKE_RESPONSE_SENT:
-                assert clientContext != null : "Client context != null";
-                processOperation(ctx, unpacker);
+                case STATE_HANDSHAKE_REQUESTED:
+                    // Handshake is in progress, any messages are not allowed.
+                    throw new IgniteException(PROTOCOL_ERR, "Unexpected message received before handshake completion");
 
-                break;
+                case STATE_HANDSHAKE_RESPONSE_SENT:
+                    assert clientContext != null : "Client context != null";
+                    processOperation(ctx, unpacker);
 
-            default:
-                throw new IllegalStateException("Unexpected state: " + state);
+                    break;
+
+                default:
+                    throw new IllegalStateException("Unexpected state: " + state);
+            }
+        } catch (Throwable t) {
+            unpacker.close();
+            throw t;
         }
     }