Adding serde for config change event key and value (#104)

config-service-change-event-api/build.gradle.kts

@@ -23,4 +23,5 @@ sourceSets {
 
 dependencies {
   api(libs.protobuf.java)
+  api(libs.kafka.clients)
 }

config-service-change-event-api/src/main/java/org/hypertrace/config/change/event/v1/ConfigChangeEventKeyDeserializer.java

@@ -0,0 +1,14 @@
+package org.hypertrace.config.change.event.v1;
+
+import com.google.protobuf.InvalidProtocolBufferException;
+import org.apache.kafka.common.serialization.Deserializer;
+
+public class ConfigChangeEventKeyDeserializer implements Deserializer<ConfigChangeEventKey> {
+  public ConfigChangeEventKey deserialize(String topic, byte[] data) {
+    try {
+      return ConfigChangeEventKey.parseFrom(data);
+    } catch (InvalidProtocolBufferException e) {
+      throw new RuntimeException(e);
+    }
+  }
+}

config-service-change-event-api/src/main/java/org/hypertrace/config/change/event/v1/ConfigChangeEventKeySerde.java

@@ -0,0 +1,24 @@
+package org.hypertrace.config.change.event.v1;
+
+import java.util.Map;
+import org.apache.kafka.common.serialization.Deserializer;
+import org.apache.kafka.common.serialization.Serde;
+import org.apache.kafka.common.serialization.Serializer;
+
+public class ConfigChangeEventKeySerde implements Serde<ConfigChangeEventKey> {
+  @Override
+  public void configure(Map<String, ?> configs, boolean isKey) {}
+
+  @Override
+  public void close() {}
+
+  @Override
+  public Serializer<ConfigChangeEventKey> serializer() {
+    return new ConfigChangeEventKeySerializer();
+  }
+
+  @Override
+  public Deserializer<ConfigChangeEventKey> deserializer() {
+    return new ConfigChangeEventKeyDeserializer();
+  }
+}

config-service-change-event-api/src/main/java/org/hypertrace/config/change/event/v1/ConfigChangeEventKeySerializer.java

@@ -0,0 +1,10 @@
+package org.hypertrace.config.change.event.v1;
+
+import org.apache.kafka.common.serialization.Serializer;
+
+public class ConfigChangeEventKeySerializer implements Serializer<ConfigChangeEventKey> {
+  @Override
+  public byte[] serialize(String topic, ConfigChangeEventKey data) {
+    return data.toByteArray();
+  }
+}

config-service-change-event-api/src/main/java/org/hypertrace/config/change/event/v1/ConfigChangeEventValueDeserializer.java

@@ -0,0 +1,16 @@
+package org.hypertrace.config.change.event.v1;
+
+import com.google.protobuf.InvalidProtocolBufferException;
+import org.apache.kafka.common.serialization.Deserializer;
+
+public class ConfigChangeEventValueDeserializer implements Deserializer<ConfigChangeEventValue> {
+
+  @Override
+  public ConfigChangeEventValue deserialize(String topic, byte[] data) {
+    try {
+      return ConfigChangeEventValue.parseFrom(data);
+    } catch (InvalidProtocolBufferException e) {
+      throw new RuntimeException(e);
+    }
+  }
+}

config-service-change-event-api/src/main/java/org/hypertrace/config/change/event/v1/ConfigChangeEventValueSerde.java

@@ -0,0 +1,24 @@
+package org.hypertrace.config.change.event.v1;
+
+import java.util.Map;
+import org.apache.kafka.common.serialization.Deserializer;
+import org.apache.kafka.common.serialization.Serde;
+import org.apache.kafka.common.serialization.Serializer;
+
+public class ConfigChangeEventValueSerde implements Serde<ConfigChangeEventValue> {
+  @Override
+  public void configure(Map<String, ?> configs, boolean isKey) {}
+
+  @Override
+  public void close() {}
+
+  @Override
+  public Serializer<ConfigChangeEventValue> serializer() {
+    return new ConfigChangeEventValueSerializer();
+  }
+
+  @Override
+  public Deserializer<ConfigChangeEventValue> deserializer() {
+    return new ConfigChangeEventValueDeserializer();
+  }
+}

config-service-change-event-api/src/main/java/org/hypertrace/config/change/event/v1/ConfigChangeEventValueSerializer.java

@@ -0,0 +1,11 @@
+package org.hypertrace.config.change.event.v1;
+
+import org.apache.kafka.common.serialization.Serializer;
+
+public class ConfigChangeEventValueSerializer implements Serializer<ConfigChangeEventValue> {
+
+  @Override
+  public byte[] serialize(String topic, ConfigChangeEventValue data) {
+    return data.toByteArray();
+  }
+}

config-service-change-event-generator/build.gradle.kts

@@ -17,7 +17,6 @@ dependencies {
   annotationProcessor(libs.lombok)
   compileOnly(libs.lombok)
 
-  runtimeOnly(libs.kafka.protobuf.serializer)
   constraints {
     implementation(libs.jersey.common) {
       because("https://snyk.io/vuln/SNYK-JAVA-ORGGLASSFISHJERSEYCORE-1255637")

config-service/src/main/resources/configs/common/application.conf

@@ -21,8 +21,8 @@ event.store {
   config.change.events.producer {
     topic.name = config-change-events
     bootstrap.servers = "localhost:9092"
-    key.serializer = io.confluent.kafka.serializers.protobuf.KafkaProtobufSerializer
-    value.serializer = io.confluent.kafka.serializers.protobuf.KafkaProtobufSerializer
+    key.serializer = org.hypertrace.config.change.event.v1.ConfigChangeEventKeySerializer
+    value.serializer = org.hypertrace.config.change.event.v1.ConfigChangeEventValueSerializer
     schema.registry.url = "http://localhost:8081"
   }
 }

gradle/libs.versions.toml

@@ -6,7 +6,7 @@ hypertrace-grpcutils = "0.6.2"
 hypertrace-framework = "0.1.29"
 lombok = "1.18.20"
 jackson = "2.12.4"
-confluent = "6.0.5"
+kafka = "6.0.1-ccs"
 netty = "4.1.71.Final"
 kotlin = "1.4.32"
 
@@ -48,7 +48,7 @@ gson = { module = "com.google.code.gson:gson", version.ref = "gson" }
 netty-codecHttp2 = { module = "io.netty:netty-codec-http2", version.ref = "netty" }
 netty-handlerProxy = { module = "io.netty:netty-handler-proxy", version.ref = "netty" }
 
-kafka-protobuf-serializer = { module = "io.confluent:kafka-protobuf-serializer", version.ref = "confluent" }
+kafka-clients = { module = "org.apache.kafka:kafka-clients", version.ref = "kafka" }
 kotlin-stdlib = { module = "org.jetbrains.kotlin:kotlin-stdlib", version.ref = "kotlin" }
 kotlin-stdlibJdk7 = { module = "org.jetbrains.kotlin:kotlin-stdlib-jdk7", version.ref = "kotlin" }
 kotlin-stdlibJdk8 = { module = "org.jetbrains.kotlin:kotlin-stdlib-jdk8", version.ref = "kotlin" }