First draft of serializer and deserializer along with tests for the serializer

Author: calvinlfer

.github/workflows/scala.yml

@@ -0,0 +1,20 @@
+name: CI
+
+on:
+  push:
+    branches: [ main ]
+  pull_request:
+    branches: [ main ]
+
+jobs:
+  build:
+    runs-on: ubuntu-latest
+    steps:
+      - uses: actions/checkout@v2
+      - name: Set up JDK 11
+        uses: actions/setup-java@v2
+        with:
+          java-version: '11'
+          distribution: 'adopt'
+      - name: Run tests
+        run: sbt test

.gitignore

@@ -0,0 +1,20 @@
+### SBT ###
+dist/*
+target/
+lib_managed/
+src_managed/
+project/boot/
+project/plugins/project/
+.history
+.cache
+.lib/
+
+### Scala ###
+*.class
+*.log
+
+### IntelliJ ###
+.idea
+
+# virtual machine crash logs, see http://www.java.com/en/download/help/error_hotspot.xml
+hs_err_pid*

.scalafmt.conf

@@ -0,0 +1,11 @@
+version = 2.7.5
+maxColumn = 120
+align = most
+continuationIndent.defnSite = 2
+assumeStandardLibraryStripMargin = true
+docstrings = JavaDoc
+lineEndings = preserve
+includeCurlyBraceInSelectChains = false
+danglingParentheses = true
+optIn.annotationNewlines = true
+rewrite.rules = [RedundantBraces]

build.sbt

@@ -0,0 +1,45 @@
+organization := "io.kaizensolutions"
+
+name := "fs2-kafka-jsonschema-support"
+
+version := "0.1"
+
+scalaVersion := "2.13.6"
+
+scalacOptions ++= Seq(
+  "-deprecation",
+  "-encoding",
+  "UTF-8",
+  "-feature",
+  "-language:implicitConversions",
+  "-unchecked",
+  "-language:higherKinds",
+  "-Xlint",
+  "-Ywarn-dead-code",
+  "-Ywarn-numeric-widen",
+  "-Ywarn-value-discard",
+  "-Ywarn-unused",
+  "-Xfatal-warnings"
+)
+
+resolvers ++= Seq(
+  "confluent" at "https://packages.confluent.io/maven/",
+  "jitpack" at "https://jitpack.io"
+)
+
+libraryDependencies ++= {
+  val circe     = "io.circe"
+  val fd4s      = "com.github.fd4s"
+  val fs2KafkaV = "2.1.0"
+  Seq(
+    fd4s                  %% "fs2-kafka"                    % fs2KafkaV,
+    fd4s                  %% "fs2-kafka-vulcan"             % fs2KafkaV,
+    "com.github.andyglow" %% "scala-jsonschema"             % "0.7.2",
+    circe                 %% "circe-jackson212"             % "0.14.0",
+    circe                 %% "circe-generic"                % "0.14.1",
+    "org.typelevel"       %% "munit-cats-effect-3"          % "1.0.0"  % Test,
+    "com.dimafeng"        %% "testcontainers-scala-munit"   % "0.39.5" % Test,
+    "ch.qos.logback"       % "logback-classic"              % "1.2.3"  % Test,
+    "io.confluent"         % "kafka-json-schema-serializer" % "6.2.0"
+  )
+}

docker-compose.yaml

@@ -0,0 +1,47 @@
+version: '2.1'
+
+services:
+  zoo1:
+    image: zookeeper:3.4.9
+    restart: unless-stopped
+    hostname: zoo1
+    ports:
+      - "2181:2181"
+    environment:
+      ZOO_MY_ID: 1
+      ZOO_PORT: 2181
+      ZOO_SERVERS: server.1=zoo1:2888:3888
+
+  kafka1:
+    image: confluentinc/cp-kafka:6.2.0
+    hostname: kafka1
+    ports:
+      - "9092:9092"
+      - "9999:9999"
+    environment:
+      KAFKA_ADVERTISED_LISTENERS: LISTENER_DOCKER_INTERNAL://kafka1:19092,LISTENER_DOCKER_EXTERNAL://${DOCKER_HOST_IP:-127.0.0.1}:9092
+      KAFKA_LISTENER_SECURITY_PROTOCOL_MAP: LISTENER_DOCKER_INTERNAL:PLAINTEXT,LISTENER_DOCKER_EXTERNAL:PLAINTEXT
+      KAFKA_INTER_BROKER_LISTENER_NAME: LISTENER_DOCKER_INTERNAL
+      KAFKA_ZOOKEEPER_CONNECT: "zoo1:2181"
+      KAFKA_BROKER_ID: 1
+      KAFKA_LOG4J_LOGGERS: "kafka.controller=INFO,kafka.producer.async.DefaultEventHandler=INFO,state.change.logger=INFO"
+      KAFKA_OFFSETS_TOPIC_REPLICATION_FACTOR: 1
+      KAFKA_TRANSACTION_STATE_LOG_REPLICATION_FACTOR: 1
+      KAFKA_TRANSACTION_STATE_LOG_MIN_ISR: 1
+      KAFKA_JMX_PORT: 9999
+      KAFKA_JMX_HOSTNAME: ${DOCKER_HOST_IP:-127.0.0.1}
+    depends_on:
+      - zoo1
+
+  kafka-schema-registry:
+    image: confluentinc/cp-schema-registry:6.2.0
+    hostname: kafka-schema-registry
+    ports:
+      - "8081:8081"
+    environment:
+      SCHEMA_REGISTRY_KAFKASTORE_BOOTSTRAP_SERVERS: PLAINTEXT://kafka1:19092
+      SCHEMA_REGISTRY_HOST_NAME: kafka-schema-registry
+      SCHEMA_REGISTRY_LISTENERS: http://0.0.0.0:8081
+    depends_on:
+      - zoo1
+      - kafka1

project/build.properties

@@ -0,0 +1 @@
+sbt.version = 1.5.4

src/main/scala/io/kaizensolutions/jsonschema/JsonSchemaDeserializer.scala

@@ -0,0 +1,119 @@
+package io.kaizensolutions.jsonschema
+
+import cats.effect.{Ref, Sync}
+import cats.syntax.all._
+import com.github.andyglow.jsonschema._
+import com.fasterxml.jackson.databind.{DeserializationFeature, JsonNode, ObjectMapper}
+import fs2.kafka.Deserializer
+import io.circe.Decoder
+import io.circe.jackson.jacksonToCirce
+import io.confluent.kafka.schemaregistry.client.SchemaRegistryClient
+import io.confluent.kafka.schemaregistry.json.JsonSchema
+import io.confluent.kafka.schemaregistry.json.jackson.Jackson
+import org.apache.kafka.common.errors.SerializationException
+
+import java.io.{ByteArrayInputStream, IOException}
+import java.nio.ByteBuffer
+import scala.reflect.ClassTag
+
+// See AbstractKafkaJsonSchemaDeserializer
+object JsonSchemaDeserializer   {
+  def make[F[_]: Sync, A: Decoder](
+    settings: JsonSchemaDeserializerSettings,
+    client: SchemaRegistryClient
+  )(implicit jsonSchema: json.Schema[A], tag: ClassTag[A]): F[Deserializer[F, A]] = {
+    val fObjectMapper: F[ObjectMapper]   =
+      Sync[F].delay {
+        val instance = Jackson.newObjectMapper()
+        if (settings.failOnUnknownKeys)
+          instance.configure(
+            DeserializationFeature.FAIL_ON_UNKNOWN_PROPERTIES,
+            settings.failOnUnknownKeys
+          )
+        instance
+      }
+    val fClientJsonSchema: F[JsonSchema] =
+      Sync[F].delay {
+        val schema =
+          new JsonSchema(
+            jsonSchema.draft07(
+              settings.jsonSchemaId
+                .getOrElse(tag.runtimeClass.getSimpleName.toLowerCase + "schema.json")
+            )
+          )
+        schema.validate()
+        schema
+      }
+
+    (fObjectMapper, fClientJsonSchema, Ref.of[F, Set[Int]](Set.empty[Int])).mapN {
+      case (objectMapper, clientSchema, cache) =>
+        new JsonSchemaDeserializer[F, A](settings, clientSchema, objectMapper, cache, client).jsonSchemaDeserializer
+    }
+  }
+}
+private[jsonschema] class JsonSchemaDeserializer[F[_]: Sync, A] private (
+  settings: JsonSchemaDeserializerSettings,
+  clientSchema: JsonSchema,
+  objectMapper: ObjectMapper,
+  compatSubjectIdCache: Ref[F, Set[Int]],
+  client: SchemaRegistryClient
+)(implicit decoder: Decoder[A]) {
+  private val MagicByte: Byte = 0x0
+  private val IdSize: Int     = 4
+
+  def jsonSchemaDeserializer: Deserializer[F, A] =
+    Deserializer.instance { (_, _, bytes) =>
+      Sync[F].delay {
+        val buffer             = getByteBuffer(bytes)
+        val id                 = buffer.getInt()
+        val serverSchema       = client.getSchemaById(id).asInstanceOf[JsonSchema]
+        val bufferLength       = buffer.limit() - 1 - IdSize
+        val start              = buffer.position() + buffer.arrayOffset()
+        val jsonNode: JsonNode =
+          objectMapper.readTree(new ByteArrayInputStream(buffer.array, start, bufferLength))
+
+        if (settings.validatePayloadAgainstServerSchema) {
+          serverSchema.validate(jsonNode)
+        }
+
+        if (settings.validatePayloadAgainstClientSchema) {
+          clientSchema.validate(jsonNode)
+        }
+
+        (id, serverSchema, jsonNode)
+      }.flatMap { case (serverId, serverSchema, jsonNode) =>
+        val check =
+          if (settings.validateClientSchemaAgainstServer)
+            checkSchemaCompatibility(serverId, serverSchema)
+          else Sync[F].unit
+
+        check.as(jacksonToCirce(jsonNode))
+      }
+        .map(decoder.decodeJson)
+        .rethrow
+    }
+
+  private def getByteBuffer(payload: Array[Byte]): ByteBuffer = {
+    val buffer = ByteBuffer.wrap(payload)
+    if (buffer.get() != MagicByte)
+      throw new SerializationException("Unknown magic byte when deserializing from Kafka")
+    buffer
+  }
+
+  private def checkSchemaCompatibility(serverSubjectId: Int, serverSchema: JsonSchema): F[Unit] = {
+    val checkSchemaUpdateCache =
+      Sync[F].delay {
+        val incompatibilities = clientSchema.isBackwardCompatible(serverSchema)
+        if (!incompatibilities.isEmpty)
+          throw new IOException(
+            s"Incompatible consumer schema with server schema: ${incompatibilities.toArray.mkString(", ")}"
+          )
+        else ()
+      } *> compatSubjectIdCache.update(_ + serverSubjectId)
+
+    for {
+      existing <- compatSubjectIdCache.get
+      _        <- if (existing.contains(serverSubjectId)) Sync[F].unit else checkSchemaUpdateCache
+    } yield ()
+  }
+}

src/main/scala/io/kaizensolutions/jsonschema/JsonSchemaDeserializerSettings.scala

@@ -0,0 +1,35 @@
+package io.kaizensolutions.jsonschema
+
+object JsonSchemaDeserializerSettings {
+  val default: JsonSchemaDeserializerSettings = JsonSchemaDeserializerSettings()
+}
+final case class JsonSchemaDeserializerSettings(
+  validatePayloadAgainstServerSchema: Boolean = false,
+  validatePayloadAgainstClientSchema: Boolean = false,
+  validateClientSchemaAgainstServer: Boolean = false,
+  failOnUnknownKeys: Boolean = false,
+  jsonSchemaId: Option[String] = None
+)                                     { self =>
+  def withPayloadValidationAgainstServerSchema(b: Boolean): JsonSchemaDeserializerSettings =
+    self.copy(validatePayloadAgainstServerSchema = b)
+
+  def withPayloadValidationAgainstClientSchema(b: Boolean): JsonSchemaDeserializerSettings =
+    self.copy(validatePayloadAgainstClientSchema = b)
+
+  def withAggressiveSchemaValidation(b: Boolean): JsonSchemaDeserializerSettings =
+    self.copy(validateClientSchemaAgainstServer = b)
+
+  def withFailOnUnknownKeys(b: Boolean): JsonSchemaDeserializerSettings =
+    self.copy(failOnUnknownKeys = b)
+
+  def withAggressiveValidation(b: Boolean): JsonSchemaDeserializerSettings =
+    self.copy(
+      validatePayloadAgainstServerSchema = b,
+      validatePayloadAgainstClientSchema = b,
+      validateClientSchemaAgainstServer = b,
+      failOnUnknownKeys = b
+    )
+
+  def withJsonSchemaId(id: String): JsonSchemaDeserializerSettings =
+    self.copy(jsonSchemaId = Some(id))
+}