Merge pull request #147 from OSGP/feature/FDP-2161-allowing-kafka-tombstones

FDP-2161 - Enabling Kafka tombstones for compact topics

Author: thomasmoelker

.github/dependabot.yml

@@ -22,7 +22,7 @@ updates:
   - package-ecosystem: github-actions
     directory: /
     schedule:
-      interval: weekly
+      interval: "weekly"
     groups:
       all:
         patterns:

gradle/libs.versions.toml

@@ -12,7 +12,7 @@ mockk = "1.14.9"
 msal4j = "1.24.0"
 sonarqube = "7.2.3.7755"
 spotless = "8.3.0"
-springBoot = "4.0.3"
+springBoot = "4.0.5"
 springmockk = "5.0.1"
 
 [libraries]

kafka-avro/build.gradle.kts

@@ -5,9 +5,50 @@ plugins {
     alias(libs.plugins.avro)
 }
 
+buildscript {
+    configurations.all { resolutionStrategy { force("org.kohsuke:github-api:1.330") } }
+    dependencies {
+        // override the avro dependencies (1.12.1) of the avro plugin (that contain vulnerabilities)
+        classpath(libs.avro)
+    }
+
+    configurations {
+        classpath {
+            resolutionStrategy {
+                // Temporary: check regularly if still necessary to override plugin classpath dependencies, and remove
+                // when no longer needed
+                eachDependency {
+                    if (requested.group == "com.fasterxml.jackson.core" && requested.name == "jackson-core") {
+                        useVersion("2.21.2")
+                        because(
+                            "Override plugin classpath to use non-vulnerable com.fasterxml.jackson.core:jackson-core 2.21.2",
+                        )
+                    }
+                }
+            }
+        }
+    }
+}
+
+configurations.all {
+    resolutionStrategy {
+        // Temporary: check regularly if still necessary to override dependency versions, and remove when no longer
+        // needed
+        eachDependency {
+            if (requested.group == "com.fasterxml.jackson.core" && requested.name == "jackson-core") {
+                useVersion("2.21.2")
+                because(
+                    "Override BOM and all constraints to use non-vulnerable com.fasterxml.jackson.core:jackson-core 2.21.2",
+                )
+            }
+        }
+    }
+}
+
 dependencies {
-    implementation(libs.kafkaClients)
     implementation(libs.avro)
+    implementation(libs.kafkaClients)
+    implementation(libs.kotlinLoggingJvm)
 
     implementation(libs.slf4jApi)
 

kafka-avro/src/main/kotlin/com/gxf/utilities/kafka/avro/AvroDeserializer.kt

@@ -3,17 +3,17 @@
 // SPDX-License-Identifier: Apache-2.0
 package com.gxf.utilities.kafka.avro
 
+import io.github.oshai.kotlinlogging.KotlinLogging
 import org.apache.avro.Schema
 import org.apache.avro.message.BinaryMessageDecoder
 import org.apache.avro.specific.SpecificData
 import org.apache.avro.specific.SpecificRecordBase
 import org.apache.kafka.common.errors.SerializationException
 import org.apache.kafka.common.serialization.Deserializer
-import org.slf4j.LoggerFactory
 
 class AvroDeserializer(deserializerSchemas: List<Schema>) : Deserializer<SpecificRecordBase> {
     companion object {
-        private val logger = LoggerFactory.getLogger(AvroDeserializer::class.java)
+        private val logger = KotlinLogging.logger {}
     }
 
     private val decoder = BinaryMessageDecoder<SpecificRecordBase>(SpecificData(), null)
@@ -26,7 +26,7 @@ class AvroDeserializer(deserializerSchemas: List<Schema>) : Deserializer<Specifi
     /** Deserializes a Byte Array to an Avro SpecificRecordBase */
     override fun deserialize(topic: String, payload: ByteArray): SpecificRecordBase {
         try {
-            logger.trace("Deserializing for {}", topic)
+            logger.trace { "Deserializing for $topic" }
             return decoder.decode(payload)
         } catch (ex: Exception) {
             throw SerializationException("Error deserializing Avro message for topic: $topic", ex)

kafka-avro/src/main/kotlin/com/gxf/utilities/kafka/avro/AvroEncoder.kt

@@ -3,18 +3,18 @@
 // SPDX-License-Identifier: Apache-2.0
 package com.gxf.utilities.kafka.avro
 
+import io.github.oshai.kotlinlogging.KotlinLogging
 import org.apache.avro.message.BinaryMessageEncoder
 import org.apache.avro.specific.SpecificData
 import org.apache.avro.specific.SpecificRecordBase
-import org.slf4j.LoggerFactory
 import java.io.IOException
 import java.io.OutputStream
 import kotlin.reflect.KClass
 
 object AvroEncoder {
     val encoders: HashMap<KClass<out SpecificRecordBase>, BinaryMessageEncoder<SpecificRecordBase>> = HashMap()
 
-    private val logger = LoggerFactory.getLogger(AvroEncoder::class.java)
+    private val logger = KotlinLogging.logger {}
 
     @Throws(IOException::class)
     fun encode(message: SpecificRecordBase): ByteArray {
@@ -38,7 +38,7 @@ object AvroEncoder {
             return existingEncoder
         }
 
-        logger.info("New encoder created for Avro schema {}", message::class)
+        logger.info { "New encoder created for Avro schema ${message::class}" }
         val newEncoder = BinaryMessageEncoder<SpecificRecordBase>(SpecificData(), message.schema)
         encoders[message::class] = newEncoder
         return newEncoder

kafka-avro/src/main/kotlin/com/gxf/utilities/kafka/avro/AvroSerializer.kt

@@ -3,24 +3,26 @@
 // SPDX-License-Identifier: Apache-2.0
 package com.gxf.utilities.kafka.avro
 
+import io.github.oshai.kotlinlogging.KotlinLogging
 import org.apache.avro.specific.SpecificRecordBase
 import org.apache.kafka.common.errors.SerializationException
 import org.apache.kafka.common.serialization.Serializer
-import org.slf4j.LoggerFactory
 import java.io.ByteArrayOutputStream
 
 class AvroSerializer : Serializer<SpecificRecordBase> {
-    companion object {
-        private val logger = LoggerFactory.getLogger(AvroSerializer::class.java)
-    }
+    private val logger = KotlinLogging.logger {}
 
-    /** Serializes a Byte Array to an Avro specific record */
-    override fun serialize(topic: String?, data: SpecificRecordBase): ByteArray {
+    /** Serializes an Avro specific record to a ByteArray, returns null if data is null (Kafka Tombstone) */
+    override fun serialize(topic: String?, data: SpecificRecordBase?): ByteArray? {
         try {
-            logger.trace("Serializing for {}", topic)
-            val outputStream = ByteArrayOutputStream()
-            AvroEncoder.encode(data, outputStream)
-            return outputStream.toByteArray()
+            return if (data == null) {
+                null
+            } else {
+                logger.trace { "Serializing for $topic" }
+                val outputStream = ByteArrayOutputStream()
+                AvroEncoder.encode(data, outputStream)
+                outputStream.toByteArray()
+            }
         } catch (ex: Exception) {
             throw SerializationException("Error serializing Avro message for topic: $topic", ex)
         }

kafka-avro/src/test/kotlin/com/gxf/utilities/kafka/avro/AvroSerializerTest.kt

@@ -9,17 +9,28 @@ import org.assertj.core.api.Assertions.assertThat
 import org.junit.jupiter.api.Test
 
 class AvroSerializerTest {
+    val topicName = "topic-name"
+
     @Test
-    fun testEncodersCache() {
+    fun `should fill encoder cache`() {
         val message1 = AvroSchema1("field no 1", "field no 2")
         val message2 = AvroSchema2("message in a bottle")
         val serializer = AvroSerializer()
 
-        serializer.serialize("", message1)
-        serializer.serialize("", message2)
+        serializer.serialize(topicName, message1)
+        serializer.serialize(topicName, message2)
 
         assertThat(AvroEncoder.encoders).containsKeys(AvroSchema1::class)
         assertThat(AvroEncoder.encoders).containsKeys(AvroSchema2::class)
         assertThat(AvroEncoder.encoders.size).isEqualTo(2)
     }
+
+    @Test
+    fun `should allow a tombstone message`() {
+        val serializer = AvroSerializer()
+
+        val result = serializer.serialize(topicName, null)
+
+        assertThat(result).isNull()
+    }
 }