build: upgrade kafka client (#620)

Author: venikkin

.github/workflows/ci.yml

@@ -12,15 +12,16 @@ jobs:
     runs-on: ubuntu-latest
 
     steps:
-      - uses: actions/checkout@v2
+      - uses: actions/checkout@v4
 
       - name: Set up JDK 11
-        uses: actions/setup-java@v1
+        uses: actions/setup-java@v4
         with:
+          distribution: temurin
           java-version: 11
 
       - name: Cache Maven packages
-        uses: actions/cache@v1
+        uses: actions/cache@v4
         with:
           path: ~/.m2
           key: ${{ runner.os }}-m2-${{ hashFiles('**/pom.xml') }}

common/src/main/kotlin/streams/extensions/CommonExtensions.kt

@@ -1,17 +1,10 @@
 package streams.extensions
 
-import org.apache.kafka.clients.consumer.ConsumerRecord
-import org.apache.kafka.clients.consumer.OffsetAndMetadata
-import org.apache.kafka.common.TopicPartition
 import org.neo4j.driver.types.Node
 import org.neo4j.driver.types.Relationship
-import streams.service.StreamsSinkEntity
-import streams.utils.JSONUtils
-import java.nio.ByteBuffer
 import java.util.*
 import javax.lang.model.SourceVersion
 
-fun Map<String,String>.getInt(name:String, defaultValue: Int) = this.get(name)?.toInt() ?: defaultValue
 fun Map<*, *>.asProperties() = this.let {
     val properties = Properties()
     properties.putAll(it)
@@ -34,10 +27,6 @@ fun Relationship.asStreamsMap(): Map<String, Any?> {
     return relMap
 }
 
-fun String.toPointCase(): String {
-    return this.split("(?<=[a-z])(?=[A-Z])".toRegex()).joinToString(separator = ".").toLowerCase()
-}
-
 fun String.quote(): String = if (SourceVersion.isIdentifier(this)) this else "`$this`"
 
 fun Map<String, Any?>.flatten(map: Map<String, Any?> = this, prefix: String = ""): Map<String, Any?> {
@@ -51,33 +40,4 @@ fun Map<String, Any?>.flatten(map: Map<String, Any?> = this, prefix: String = ""
             listOf(newKey to value)
         }
     }.toMap()
-}
-
-fun ConsumerRecord<*, *>.topicPartition() = TopicPartition(this.topic(), this.partition())
-fun ConsumerRecord<*, *>.offsetAndMetadata(metadata: String = "") = OffsetAndMetadata(this.offset() + 1, metadata)
-
-private fun convertAvroData(rawValue: Any?): Any? = when (rawValue) {
-    is Collection<*> -> rawValue.map(::convertAvroData)
-    is Array<*> -> if (rawValue.javaClass.componentType.isPrimitive) rawValue else rawValue.map(::convertAvroData)
-    is Map<*, *> -> rawValue
-            .mapKeys { it.key.toString() }
-            .mapValues { convertAvroData(it.value) }
-    is ByteBuffer -> rawValue.array()
-    is CharSequence -> rawValue.toString()
-    else -> rawValue
-}
-
-
-
-private fun convertData(data: Any?, stringWhenFailure: Boolean = false): Any? {
-    return when (data) {
-        null -> null
-        is ByteArray -> JSONUtils.readValue<Any>(data, stringWhenFailure)
-        else -> if (stringWhenFailure) data.toString() else throw RuntimeException("Unsupported type ${data::class.java.name}")
-    }
-}
-fun ConsumerRecord<*, *>.toStreamsSinkEntity(): StreamsSinkEntity {
-    val key = convertData(this.key(), true)
-    val value = convertData(this.value())
-    return StreamsSinkEntity(key, value)
 }

common/src/main/kotlin/streams/service/errors/ErrorService.kt

@@ -1,10 +1,7 @@
 package streams.service.errors
 
-import org.apache.kafka.clients.consumer.ConsumerRecord
-import org.apache.kafka.common.record.RecordBatch
 import java.util.Properties
 
-
 data class ErrorData(val originalTopic: String,
                      val timestamp: Long,
                      val key: ByteArray?,
@@ -17,21 +14,11 @@ data class ErrorData(val originalTopic: String,
 
     constructor(originalTopic: String, timestamp: Long?, key: Any?, value: Any?,
                 partition: Int, offset: Long, executingClass: Class<*>?, databaseName: String?, exception: Exception?) :
-            this(originalTopic, timestamp ?: RecordBatch.NO_TIMESTAMP, toByteArray(key), toByteArray(value), partition.toString(),offset.toString(), executingClass, databaseName, exception)
+            this(originalTopic, timestamp ?: NO_TIMESTAMP, toByteArray(key), toByteArray(value), partition.toString(),offset.toString(), executingClass, databaseName, exception)
 
     companion object {
 
-        fun from(consumerRecord: ConsumerRecord<out Any, out Any>, exception: Exception?, executingClass: Class<*>?, databaseName: String?): ErrorData {
-            return ErrorData(offset = consumerRecord.offset().toString(),
-                    originalTopic = consumerRecord.topic(),
-                    partition = consumerRecord.partition().toString(),
-                    timestamp = consumerRecord.timestamp(),
-                    exception = exception,
-                    executingClass = executingClass,
-                    key = toByteArray(consumerRecord.key()),
-                    value = toByteArray(consumerRecord.value()),
-                    databaseName = databaseName)
-        }
+        private const val NO_TIMESTAMP: Long = -1L
 
         fun toByteArray(v:Any?) = try {
             when (v) {

common/src/main/kotlin/streams/utils/KafkaValidationUtils.kt

@@ -6,7 +6,6 @@ import org.apache.kafka.clients.producer.ProducerRecord
 import org.apache.kafka.clients.producer.internals.FutureRecordMetadata
 import org.apache.kafka.common.record.RecordBatch
 import org.apache.kafka.common.utils.SystemTime
-import org.apache.kafka.common.utils.Time
 import org.junit.Test
 import org.mockito.ArgumentMatchers
 import org.mockito.Mockito
@@ -25,7 +24,7 @@ class KafkaErrorServiceTest {
         val counter = AtomicInteger(0)
         Mockito.`when`(producer.send(ArgumentMatchers.any<ProducerRecord<ByteArray, ByteArray>>())).then {
             counter.incrementAndGet()
-            FutureRecordMetadata(null, 0, RecordBatch.NO_TIMESTAMP, 0L, 0, 0, SystemTime())
+            FutureRecordMetadata(null, 0, RecordBatch.NO_TIMESTAMP, 0, 0, SystemTime())
         }
         val dlqService = KafkaErrorService(producer, ErrorService.ErrorConfig(fail=false,dlqTopic = "dlqTopic"), { s, e -> })
         dlqService.report(listOf(dlqData()))

common/src/test/kotlin/streams/service/sink/errors/KafkaErrorServiceTest.kt

@@ -52,7 +52,8 @@
         <kotlin.version>1.9.20</kotlin.version>
         <kotlin.coroutines.version>1.7.3</kotlin.coroutines.version>
         <neo4j.version>4.4.27</neo4j.version>
-        <kafka.version>2.6.3</kafka.version>
+        <!-- Version 3.4 is the minimal version introducing version for CVE-2023-25194 -->
+        <kafka.version>3.4.1</kafka.version>
         <jackson.version>2.15.2</jackson.version>
         <kotlin.compiler.incremental>true</kotlin.compiler.incremental>
         <neo4j.java.driver.version>4.4.12</neo4j.java.driver.version>