resolved comments

Author: singh-viikram

kafka-bom/build.gradle.kts

@@ -15,7 +15,7 @@ dependencies {
       because("[https://nvd.nist.gov/vuln/detail/CVE-2023-34455] in 'org.apache.kafka:kafka-clients:*'")
       because("[https://nvd.nist.gov/vuln/detail/CVE-2023-43642]")
     }
-    api("com.google.protobuf:protobuf-java-util:3.21.7") {
+    api("com.google.protobuf:protobuf-java-util:$protobufVersion") {
       because("https://nvd.nist.gov/vuln/detail/CVE-2022-3171")
     }
     api("com.squareup.okio:okio:3.4.0") {

kafka-streams-serdes/build.gradle.kts

@@ -1,7 +1,6 @@
 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")
@@ -16,7 +15,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
+  api("com.google.protobuf:protobuf-java-util")
 
   testImplementation("org.junit.jupiter:junit-jupiter:5.8.2")
 }

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

@@ -5,6 +5,43 @@
 import com.google.protobuf.Parser;
 import org.apache.kafka.common.serialization.Deserializer;
 
+/**
+ * Custom Proto Deserializer for Kafka.
+ *
+ * <p>This class provides a deserialization mechanism for Kafka messages using Protocol Buffers
+ * without schema validation. It extends the Kafka Deserializer interface and allows for direct
+ * deserialization of byte arrays into Proto message objects by utilizing the provided Parser for
+ * the specific Proto message type.
+ *
+ * <p>Motivation: In setups where both producers and consumers use the same Proto schemas, the need
+ * for schema validation becomes redundant. The built-in {@code kafkaProtoSerdes} from Confluent
+ * performs schema validation via the schema registry service, which introduces overhead. This
+ * custom deserializer eliminates that overhead, simplifying the processing flow by bypassing schema
+ * validation.
+ *
+ *
+ * <p>Usage: To use this class, create a subclass specifying the Proto message type, pass the
+ * corresponding Parser to the superclass constructor, and configure Kafka to use the custom
+ * deserializer.
+ *
+ * <p>Example:
+ *
+ * <pre>{@code
+ * public class MyProtoMessageDeserializer extends ProtoDeserializer<MyProtoMessage> {
+ *     public MyProtoMessageDeserializer() {
+ *         super(MyProtoMessage.parser());
+ *     }
+ * }
+ * }</pre>
+ *
+ * Then, configure Kafka to use this deserializer:
+ *
+ * <pre>{@code
+ * key.deserializer=com.example.MyProtoMessageDeserializer
+ * }</pre>
+ *
+ * @param <T> The Proto message type to be deserialized.
+ */
 public class ProtoDeserializer<T extends Message> implements Deserializer<T> {
 
   private final Parser<T> parser;