MQTT unknown message type isn't handled as decoding error

Motivation:

MQTT unknown message type isn't handled as decoding error

Modification:

Catching exception during the MQTT decoding of the fixed header
Adding a unit test for unknown MQTT message type

Result:

Fixes #6984.

codec-mqtt/src/main/java/io/netty/handler/codec/mqtt/MqttDecoder.java

@@ -71,11 +71,15 @@ public final class MqttDecoder extends ReplayingDecoder<DecoderState> {
     @Override
     protected void decode(ChannelHandlerContext ctx, ByteBuf buffer, List<Object> out) throws Exception {
         switch (state()) {
-            case READ_FIXED_HEADER:
+            case READ_FIXED_HEADER: try {
                 mqttFixedHeader = decodeFixedHeader(buffer);
                 bytesRemainingInVariablePart = mqttFixedHeader.remainingLength();
                 checkpoint(DecoderState.READ_VARIABLE_HEADER);
                 // fall through
+            } catch (Exception cause) {
+                out.add(invalidMessage(cause));
+                return;
+            }
 
             case READ_VARIABLE_HEADER:  try {
                 if (bytesRemainingInVariablePart > maxBytesInMessage) {

codec-mqtt/src/test/java/io/netty/handler/codec/mqtt/MqttCodecTest.java

@@ -239,6 +239,29 @@ public class MqttCodecTest {
         testMessageWithOnlyFixedHeader(MqttMessageType.DISCONNECT);
     }
 
+    @Test
+    public void testUnknownMessageType() throws Exception {
+
+        final MqttMessage message = createMessageWithFixedHeader(MqttMessageType.PINGREQ);
+        ByteBuf byteBuf = MqttEncoder.doEncode(ALLOCATOR, message);
+        try {
+            // setting an invalid message type (15, reserved and forbidden by MQTT 3.1.1 spec)
+            byteBuf.setByte(0, 0xF0);
+            final List<Object> out = new LinkedList<Object>();
+            mqttDecoder.decode(ctx, byteBuf, out);
+
+            assertEquals("Expected one object but got " + out.size(), 1, out.size());
+
+            final MqttMessage decodedMessage = (MqttMessage) out.get(0);
+            assertTrue(decodedMessage.decoderResult().isFailure());
+            Throwable cause = decodedMessage.decoderResult().cause();
+            assertTrue(cause instanceof IllegalArgumentException);
+            assertEquals("unknown message type: 15", cause.getMessage());
+        } finally {
+            byteBuf.release();
+        }
+    }
+
     private void testMessageWithOnlyFixedHeader(MqttMessageType messageType) throws Exception {
         MqttMessage message = createMessageWithFixedHeader(messageType);
         ByteBuf byteBuf = MqttEncoder.doEncode(ALLOCATOR, message);