Adding custom proto serdes

Author: singh-viikram

kafka-streams-serdes/build.gradle.kts

@@ -1,6 +1,7 @@
 plugins {
   `java-library`
   jacoco
+  id("com.google.protobuf") version "0.9.4"
   id("org.hypertrace.avro-plugin")
   id("org.hypertrace.publish-plugin")
   id("org.hypertrace.jacoco-report-plugin")
@@ -15,6 +16,7 @@ dependencies {
 
   api("org.apache.kafka:kafka-clients")
   api("org.apache.avro:avro")
+  implementation("com.google.protobuf:protobuf-java:3.23.0") // Adjust the version as needed
 
   testImplementation("org.junit.jupiter:junit-jupiter:5.8.2")
 }

kafka-streams-serdes/src/main/java/org/hypertrace/core/kafkastreams/framework/serdes/proto/ProtoDeserializer.java

@@ -0,0 +1,24 @@
+package org.hypertrace.core.kafkastreams.framework.serdes.proto;
+
+import com.google.protobuf.InvalidProtocolBufferException;
+import com.google.protobuf.Message;
+import com.google.protobuf.Parser;
+import org.apache.kafka.common.serialization.Deserializer;
+
+public class ProtoDeserializer<T extends Message> implements Deserializer<T> {
+
+  private final Parser<T> parser;
+
+  public ProtoDeserializer(Parser<T> parser) {
+    this.parser = parser;
+  }
+
+  @Override
+  public T deserialize(String s, byte[] bytes) {
+    try {
+      return parser.parseFrom(bytes);
+    } catch (InvalidProtocolBufferException e) {
+      throw new RuntimeException(e);
+    }
+  }
+}

kafka-streams-serdes/src/main/java/org/hypertrace/core/kafkastreams/framework/serdes/proto/ProtoSerializer.java

@@ -0,0 +1,11 @@
+package org.hypertrace.core.kafkastreams.framework.serdes.proto;
+
+import com.google.protobuf.Message;
+import org.apache.kafka.common.serialization.Serializer;
+
+public class ProtoSerializer<T extends Message> implements Serializer<T> {
+  @Override
+  public byte[] serialize(String topic, T data) {
+    return data.toByteArray();
+  }
+}