Add MessagePack Support

pom.xml

@@ -111,6 +111,11 @@
             <artifactId>avro</artifactId>
             <version>1.9.1</version>
         </dependency>
+        <dependency>
+            <groupId>org.msgpack</groupId>
+            <artifactId>msgpack-core</artifactId>
+            <version>0.8.20</version>
+        </dependency>
 
         <!-- Spring Boot -->
         <dependency>

src/main/java/kafdrop/controller/MessageController.java

@@ -192,6 +192,8 @@ private MessageFormat getSelectedMessageFormat(String format) {
       return MessageFormat.AVRO;
     } else if ("PROTOBUF".equalsIgnoreCase(format)) {
       return MessageFormat.PROTOBUF;
+    } else if ("MSGPACK".equalsIgnoreCase(format)){
+      return MessageFormat.MSGPACK;
     } else {
       return MessageFormat.DEFAULT;
     }
@@ -268,6 +270,8 @@ private MessageDeserializer getDeserializer(String topicName, MessageFormat form
           .replaceAll("/", "");
       final var fullDescFile = protobufProperties.getDirectory() + File.separator + descFileName + ".desc";
       deserializer = new ProtobufMessageDeserializer(topicName, fullDescFile, msgTypeName);
+    } else if (format == MessageFormat.MSGPACK) {
+      deserializer = new MsgPackMessageDeserializer();
     } else {
       deserializer = new DefaultMessageDeserializer();
     }

src/main/java/kafdrop/util/MessageFormat.java

@@ -1,5 +1,5 @@
 package kafdrop.util;
 
 public enum MessageFormat {
-  DEFAULT, AVRO, PROTOBUF
+  DEFAULT, AVRO, PROTOBUF, MSGPACK
 }

src/main/java/kafdrop/util/MsgPackMessageDeserializer.java

@@ -0,0 +1,27 @@
+package kafdrop.util;
+
+import java.io.IOException;
+import java.nio.ByteBuffer;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import org.msgpack.core.MessagePack;
+import org.msgpack.core.MessageUnpacker;
+
+public class MsgPackMessageDeserializer implements MessageDeserializer {
+
+  private static final Logger LOG = LoggerFactory.getLogger(MsgPackMessageDeserializer.class);
+
+  @Override
+  public String deserializeMessage(ByteBuffer buffer) {
+    MessageUnpacker unpacker = MessagePack.newDefaultUnpacker(buffer);
+    try {
+      return unpacker.unpackValue().toJson();
+    } catch (IOException e) {
+      final String errorMsg = "Unable to unpack msgpack message";
+      LOG.error(errorMsg, e);
+      throw new DeserializationException(errorMsg);
+    }
+  }
+}