BE: support MessagePackSerde

api/build.gradle

@@ -59,6 +59,8 @@ dependencies {
     implementation libs.confluent.protobuf.serializer
     implementation libs.confluent.json.schema.serializer
 
+    implementation libs.msgpack.core
+
     implementation libs.aws.msk.auth
     implementation(libs.azure.identity) {
         exclude group: 'io.netty', module: 'netty-tcnative-boringssl-static'

api/src/main/java/io/kafbat/ui/serdes/SerdesInitializer.java

@@ -14,6 +14,7 @@
 import io.kafbat.ui.serdes.builtin.HexSerde;
 import io.kafbat.ui.serdes.builtin.Int32Serde;
 import io.kafbat.ui.serdes.builtin.Int64Serde;
+import io.kafbat.ui.serdes.builtin.MessagePackSerde;
 import io.kafbat.ui.serdes.builtin.ProtobufFileSerde;
 import io.kafbat.ui.serdes.builtin.ProtobufRawSerde;
 import io.kafbat.ui.serdes.builtin.StringSerde;
@@ -52,6 +53,7 @@ public SerdesInitializer() {
             .put(AvroEmbeddedSerde.NAME, AvroEmbeddedSerde.class)
             .put(Base64Serde.NAME, Base64Serde.class)
             .put(HexSerde.NAME, HexSerde.class)
+            .put(MessagePackSerde.NAME, MessagePackSerde.class)
             .put(UuidBinarySerde.NAME, UuidBinarySerde.class)
             .put(ProtobufRawSerde.NAME, ProtobufRawSerde.class)
 

api/src/main/java/io/kafbat/ui/serdes/builtin/MessagePackSerde.java

@@ -0,0 +1,55 @@
+package io.kafbat.ui.serdes.builtin;
+
+import io.kafbat.ui.serde.api.DeserializeResult;
+import io.kafbat.ui.serde.api.Serde;
+import io.kafbat.ui.serdes.BuiltInSerde;
+import java.io.IOException;
+import java.util.Map;
+import org.msgpack.core.MessageBufferPacker;
+import org.msgpack.core.MessagePack;
+import org.msgpack.core.MessageUnpacker;
+import org.msgpack.value.Value;
+
+
+public class MessagePackSerde implements BuiltInSerde {
+  public static final String NAME = "MessagePack";
+
+  @Override
+  public boolean canDeserialize(String topic, Serde.Target type) {
+    return true;
+  }
+
+  @Override
+  public boolean canSerialize(String topic, Serde.Target type) {
+    return true;
+  }
+
+  @Override
+  public Serde.Serializer serializer(String topic, Serde.Target type) {
+    return inputString -> {
+      inputString = inputString.trim();
+      // it is actually a hack to provide ability to sent empty array as a key/value
+      if (inputString.isEmpty()) {
+        return new byte[] {};
+      }
+      try (MessageBufferPacker packer = MessagePack.newDefaultBufferPacker()) {
+        packer.packString(inputString);
+        return packer.toByteArray();
+      } catch (IOException e) {
+        throw new IllegalArgumentException("Failed to serialize MessagePack payload", e);
+      }
+    };
+  }
+
+  @Override
+  public Serde.Deserializer deserializer(String topic, Serde.Target type) {
+    return (headers, data) -> {
+      try (MessageUnpacker unpacker = MessagePack.newDefaultUnpacker(data)) {
+        Value value = unpacker.unpackValue();
+        return new DeserializeResult(value.toString(), DeserializeResult.Type.STRING, Map.of());
+      } catch (IOException e) {
+        throw new IllegalArgumentException("Failed to deserialize MessagePack payload", e);
+      }
+    };
+  }
+}

gradle/libs.versions.toml

@@ -16,6 +16,7 @@ avro = '1.12.1'
 byte-buddy = '1.18.1'
 confluent = '7.9.5'
 confluent-ccs = '7.9.5-ccs'
+msgpack = "0.9.11"
 
 mapstruct = '1.6.2'
 lombok = '1.18.42'
@@ -90,6 +91,8 @@ confluent-avro-serializer = { module = 'io.confluent:kafka-avro-serializer', ver
 confluent-protobuf-serializer = { module = 'io.confluent:kafka-protobuf-serializer', version.ref = 'confluent' }
 confluent-json-schema-serializer = { module = 'io.confluent:kafka-json-schema-serializer', version.ref = 'confluent' }
 
+msgpack-core = { group = "org.msgpack", name = "msgpack-core", version.ref = "msgpack" }
+
 aws-msk-auth = { module = 'software.amazon.msk:aws-msk-iam-auth', version.ref = 'aws-msk-auth' }
 azure-identity = { module = 'com.azure:azure-identity', version.ref = 'azure-identity' }
 reactor-test = { module = 'io.projectreactor:reactor-test' }