Merge pull request #164 from RADAR-base/fix_schemaretriever

Fix schemaretriever

.github/workflows/snyk.yaml

@@ -3,6 +3,7 @@ on:
   pull_request:
     branches:
       - master
+      - dev
 
 jobs:
   security:

buildSrc/src/main/kotlin/Versions.kt

@@ -14,17 +14,17 @@ object Versions {
     }
 
     const val java = 17
-    const val slf4j = "2.0.9"
-    const val confluent = "7.5.0"
-    const val kafka = "7.5.0-ce"
+    const val slf4j = "2.0.13"
+    const val confluent = "7.6.0"
+    const val kafka = "${confluent}-ce"
     const val avro = "1.11.3"
-    const val jackson = "2.15.2"
-    const val okhttp = "4.11.0"
+    const val jackson = "2.15.3"
+    const val okhttp = "4.12.0"
     const val junit = "5.10.0"
     const val mockito = "5.5.0"
     const val mockitoKotlin = "5.1.0"
     const val hamcrest = "2.2"
-    const val radarSchemas = "0.8.4"
+    const val radarSchemas = "0.8.8"
     const val opencsv = "5.8"
     const val ktor = "2.3.4"
     const val coroutines = "1.7.3"

radar-commons-gradle/build.gradle.kts

@@ -190,17 +190,17 @@ object Versions {
     }
 
     const val java = 17
-    const val slf4j = "2.0.9"
-    const val confluent = "7.5.0"
-    const val kafka = "7.5.0-ce"
+    const val slf4j = "2.0.13"
+    const val confluent = "7.6.0"
+    const val kafka = "${confluent}-ce"
     const val avro = "1.11.3"
-    const val jackson = "2.15.2"
-    const val okhttp = "4.11.0"
-    const val junit = "5.10.0"
+    const val jackson = "2.15.3"
+    const val okhttp = "4.12.0"
+    const val junit = "5.10.3"
     const val mockito = "5.5.0"
     const val mockitoKotlin = "5.1.0"
     const val hamcrest = "2.2"
-    const val radarSchemas = "0.8.4"
+    const val radarSchemas = "0.8.8"
     const val opencsv = "5.8"
     const val ktor = "2.3.4"
     const val coroutines = "1.7.3"
@@ -210,4 +210,3 @@ object Versions {
     const val gradleVersionsPlugin = "0.50.0"
     const val ktlint = "12.0.3"
 }
-

radar-commons/src/main/java/org/radarbase/producer/schema/SchemaRestClient.kt

@@ -2,13 +2,10 @@ package org.radarbase.producer.schema
 
 import io.ktor.client.HttpClient
 import io.ktor.client.call.body
+import io.ktor.client.plugins.auth.*
+import io.ktor.client.plugins.auth.providers.*
 import io.ktor.client.plugins.contentnegotiation.ContentNegotiation
-import io.ktor.client.plugins.defaultRequest
-import io.ktor.client.request.HttpRequestBuilder
-import io.ktor.client.request.accept
-import io.ktor.client.request.request
-import io.ktor.client.request.setBody
-import io.ktor.client.request.url
+import io.ktor.client.request.*
 import io.ktor.http.ContentType
 import io.ktor.http.HttpMethod
 import io.ktor.http.contentType
@@ -21,13 +18,15 @@ import kotlinx.coroutines.withContext
 import kotlinx.serialization.json.Json
 import org.apache.avro.Schema
 import org.radarbase.producer.rest.RestException.Companion.toRestException
+import org.slf4j.LoggerFactory
 import java.io.IOException
+import java.net.URI
 import kotlin.coroutines.CoroutineContext
 
 /** REST client for Confluent schema registry.  */
 class SchemaRestClient(
     httpClient: HttpClient,
-    baseUrl: String,
+    private val baseUrl: String,
     private val ioContext: CoroutineContext = Dispatchers.IO,
 ) {
     private val httpClient: HttpClient = httpClient.config {
@@ -39,10 +38,6 @@ class SchemaRestClient(
                 },
             )
         }
-        defaultRequest {
-            url(baseUrl)
-            accept(ContentType.Application.Json)
-        }
     }
 
     suspend inline fun <reified T> request(
@@ -88,7 +83,7 @@ class SchemaRestClient(
     @Throws(IOException::class)
     suspend fun schemaGet(path: String): SchemaMetadata = request {
         method = HttpMethod.Get
-        url(path)
+        url(URI(baseUrl).resolve(path).toString())
     }
 
     @Throws(IOException::class)
@@ -97,7 +92,7 @@ class SchemaRestClient(
         schema: Schema,
     ): SchemaMetadata = request {
         method = HttpMethod.Post
-        url(path)
+        url(URI(baseUrl).resolve(path).toString())
         contentType(ContentType.Application.Json)
         setBody(SchemaMetadata(schema = schema.toString()))
     }
@@ -132,4 +127,8 @@ class SchemaRestClient(
         schemaGet("/schemas/ids/$id")
             .toParsedSchemaMetadata(id)
             .schema
+
+    companion object {
+        private val logger = LoggerFactory.getLogger(SchemaRestClient::class.java)
+    }
 }

radar-commons/src/main/java/org/radarbase/producer/schema/SchemaRetriever.kt

@@ -205,8 +205,8 @@ open class SchemaRetriever(config: Config) {
     @RadarProducerDsl
     class Config(
         val baseUrl: String,
+        var httpClient: HttpClient? = null,
     ) {
-        var httpClient: HttpClient? = null
         var schemaTimeout: CacheConfig = DEFAULT_SCHEMA_TIMEOUT_CONFIG
         var ioContext: CoroutineContext = Dispatchers.IO
         fun httpClient(config: HttpClientConfig<*>.() -> Unit) {