feat: rework producer & serializer

Signed-off-by: Mehdi Rebiai <[email protected]>
lectra-tech · Aug 14, 2024 · a1db5c1 · a1db5c1
1 parent b48fd69
commit a1db5c1
Show file tree

Hide file tree

Showing 10 changed files with 175 additions and 216 deletions.
diff --git a/build.sbt b/build.sbt
@@ -26,7 +26,6 @@ lazy val root = (project in file("."))
       "org.scalatest" %% "scalatest" % "3.2.19",
       "com.typesafe" % "config" % "1.4.3",
       "io.gatling" % "gatling-jsonpath" % "3.11.5",
-      "com.lihaoyi" %% "requests" % "0.9.0",
       "com.lihaoyi" %% "os-lib" % "0.10.3",
       "ch.qos.logback" % "logback-classic" % "1.5.6" % Runtime,
       "dev.zio" %% "zio" % zioVersion,

diff --git a/docker-compose.yml b/docker-compose.yml
@@ -77,7 +77,7 @@ services:
     environment:
       CONTROL_CENTER_BOOTSTRAP_SERVERS: 'kafka:29092'
       CONTROL_CENTER_CONNECT_CONNECT-DEFAULT_CLUSTER: 'connect:8083'
-      CONTROL_CENTER_KSQL_KSQLDB1_URL: "http://primary-ksqldb-server:8088"
+      CONTROL_CENTER_KSQL_KSQLDB1_URL: "http://ksqldb-server:8088"
       CONTROL_CENTER_KSQL_KSQLDB1_ADVERTISED_URL: "http://localhost:8088"
       CONTROL_CENTER_SCHEMA_REGISTRY_URL: "http://schema-registry:8081"
       CONTROL_CENTER_SCHEMA_REGISTRY_BASIC_AUTH_CREDENTIALS_SOURCE: USER_INFO
@@ -87,10 +87,10 @@ services:
       CONTROL_CENTER_MONITORING_INTERCEPTOR_TOPIC_PARTITIONS: 1
       CONFLUENT_METRICS_TOPIC_REPLICATION: 1
       PORT: 9021
-  primary-ksqldb-server:
+  ksqldb-server:
     image: confluentinc/cp-ksqldb-server:${CONFLUENT_KAFKA_VERSION}
-    hostname: primary-ksqldb-server
-    container_name: primary-ksqldb-server
+    hostname: ksqldb-server
+    container_name: ksqldb-server
     depends_on:
       - kafka
       - schema-registry
@@ -107,8 +107,6 @@ services:
       KSQL_KSQL_LOGGING_PROCESSING_STREAM_AUTO_CREATE: "true"
       KSQL_KSQL_LOGGING_PROCESSING_TOPIC_AUTO_CREATE: "true"
   create-topics:
-    profiles:
-      - test
     image: confluentinc/cp-enterprise-kafka:${CONFLUENT_KAFKA_VERSION}
     volumes:
       - ./src/test/resources/scripts:/opt/kapoeira/scripts
@@ -126,8 +124,6 @@ services:
         echo Creating subjects... && \
         ./createSchemas.sh'"
   create-streams:
-    profiles:
-      - test
     image: confluentinc/cp-ksqldb-cli:${CONFLUENT_KAFKA_VERSION}
     container_name: create-streams
     volumes:
@@ -142,7 +138,7 @@ services:
       - -f
       - /opt/kapoeira/scripts/createStreams.ksql
       - --
-      - http://primary-ksqldb-server:8088
+      - http://ksqldb-server:8088
   kapoeira:
     profiles:
       - test

diff --git a/src/main/scala/com/lectra/kapoeira/domain/RecordRead.scala b/src/main/scala/com/lectra/kapoeira/domain/RecordRead.scala
@@ -21,15 +21,14 @@ package com.lectra.kapoeira.domain
 import com.lectra.kapoeira.domain
 import com.lectra.kapoeira.domain.Services.ReadHeaders
 
-import java.nio.charset.StandardCharsets
 import scala.util.Try
 
 /** The record read from the cucumber Datatable directly, or a File.
   */
 final case class RecordRead(
     topicAlias: String,
     key: String,
-    value: Array[Byte],
+    value: String,
     headers: Map[String, Any]
 )
 
@@ -54,11 +53,11 @@ trait RecordData[T] {
 /** Facilitate import of all implicits.
   */
 trait RecordReadImplicits
-    extends InterpotaleImplicits
+    extends InterpolateImplicits
     with RecordDataImplicits
     with RecordDataFromFileImplicits
 
-trait InterpotaleImplicits {
+trait InterpolateImplicits {
 
   implicit val interpolateString: Interpolate[String] =
     (t: String, ctx: BackgroundContext) => ctx.substituteVariablesIn(t)
@@ -70,8 +69,7 @@ trait InterpotaleImplicits {
         ctx
           .substituteVariablesIn(t.key),
         ctx
-          .substituteVariablesIn(new String(t.value))
-          .getBytes(StandardCharsets.UTF_8),
+          .substituteVariablesIn(t.value),
         interpolateMap(t.headers, ctx)
       )
     }
@@ -123,7 +121,7 @@ trait RecordDataFromFileImplicits {
           domain.RecordRead(
             t.topicAlias,
             columns(0),
-            columns(1).getBytes(StandardCharsets.UTF_8),
+            columns(1),
             Try(columns(2))
               .map(headersString => readHeaders.readHeaders(headersString))
               .getOrElse(Map.empty)
@@ -141,7 +139,7 @@ trait RecordDataFromFileImplicits {
         RecordRead(
           t.topicAlias,
           t.key,
-          line.getBytes(StandardCharsets.UTF_8),
+          line,
           Map.empty
         )
       })
@@ -159,7 +157,7 @@ trait RecordDataFromFileImplicits {
           RecordRead(
             t.topicAlias,
             t.key,
-            line.getBytes(StandardCharsets.UTF_8),
+            line,
             Map.empty
           )
         )
@@ -179,7 +177,7 @@ trait RecordDataImplicits {
       override def read(t: KeyValueRecord): RecordRead = RecordRead(
         t.topicAlias,
         t.key,
-        t.value.getBytes(StandardCharsets.UTF_8),
+        t.value,
         t.headers
       )
     }

diff --git a/src/main/scala/com/lectra/kapoeira/glue/package.scala b/src/main/scala/com/lectra/kapoeira/glue/package.scala
@@ -25,6 +25,7 @@ import com.lectra.kapoeira.domain._
 import com.lectra.kapoeira.kafka.KapoeiraConsumer._
 import com.lectra.kapoeira.kafka.{KapoeiraAdmin, KapoeiraConsumer}
 import com.typesafe.scalalogging.LazyLogging
+import io.confluent.kafka.schemaregistry.json.jackson.Jackson
 import org.apache.kafka.clients.admin.AdminClient
 import org.apache.kafka.clients.consumer.ConsumerRecord
 import org.apache.kafka.common.TopicPartition
@@ -123,7 +124,7 @@ package object glue extends LazyLogging with RecordReadImplicits {
     }
   }
 
-  private[glue] val objectMapper = new ObjectMapper()
+  private[glue] val objectMapper = Jackson.newObjectMapper();
 
   implicit class RecordReadOps(recordRead: RecordRead) {
     def jsonHeaders: Try[Map[String, Array[Byte]]] = Try(

diff --git a/src/main/scala/com/lectra/kapoeira/kafka/DataType.scala b/src/main/scala/com/lectra/kapoeira/kafka/DataType.scala
@@ -18,34 +18,26 @@
  */
 package com.lectra.kapoeira.kafka
 import com.fasterxml.jackson.databind.JsonNode
-import io.confluent.kafka.serializers.{KafkaAvroDeserializer, KafkaAvroSerializer}
-import io.confluent.kafka.serializers.json.{KafkaJsonSchemaDeserializer, KafkaJsonSchemaSerializer}
+import io.confluent.kafka.serializers.KafkaAvroDeserializer
+import io.confluent.kafka.serializers.json.KafkaJsonSchemaDeserializer
 import org.apache.avro.generic.GenericData
-import org.apache.kafka.common.serialization.{StringDeserializer, StringSerializer}
+import org.apache.kafka.common.serialization.StringDeserializer
 
 sealed trait DataType[A] {
   type DeserializerT
-  type SerializerT
   val classDeserializer: Class[DeserializerT]
-  val classSerializer: Class[SerializerT]
 }
 trait AvroType[T] extends DataType[T] {
   type DeserializerT = KafkaAvroDeserializer
-  type SerializerT = KafkaAvroSerializer
   val classDeserializer: Class[KafkaAvroDeserializer] = classOf[KafkaAvroDeserializer]
-  val classSerializer: Class[KafkaAvroSerializer] = classOf[KafkaAvroSerializer]
 }
 case object StringType extends DataType[String] {
   type DeserializerT = StringDeserializer
-  type SerializerT = StringSerializer
   val classDeserializer: Class[StringDeserializer] = classOf[StringDeserializer]
-  val classSerializer: Class[StringSerializer] = classOf[StringSerializer]
 }
 case object JsonType extends DataType[JsonNode] {
   type DeserializerT = KafkaJsonSchemaDeserializer[JsonNode]
-  type SerializerT = KafkaJsonSchemaSerializer[JsonNode]
   val classDeserializer: Class[KafkaJsonSchemaDeserializer[JsonNode]] = classOf[KafkaJsonSchemaDeserializer[JsonNode]]
-  val classSerializer: Class[KafkaJsonSchemaSerializer[JsonNode]] = classOf[KafkaJsonSchemaSerializer[JsonNode]]
 }
 object DataType {
   implicit val avroType: DataType[Any] = new AvroType[Any] {}