Fix auth issue

.env

@@ -1 +1 @@
-KAFKA_CONFLUENT_VERSION=7.5.0
+KAFKA_CONFLUENT_VERSION=7.6.0

buildSrc/src/main/kotlin/Versions.kt

@@ -1,15 +1,15 @@
 @Suppress("ConstPropertyName")
 object Versions {
-    const val project = "0.7.1-SNAPSHOT"
+    const val project = "0.7.2-SNAPSHOT"
 
     const val java = 17
     const val kotlin = "1.9.22"
     const val dockerCompose = "0.17.6"
 
     const val ktor = "2.3.10"
     const val radarJersey = "0.11.1"
-    const val radarCommons = "1.1.2"
-    const val radarSchemas = "0.8.7"
+    const val radarCommons = "1.1.3-SNAPSHOT"
+    const val radarSchemas = "0.8.8"
     const val jackson = "2.15.3"
     const val slf4j = "2.0.13"
     const val log4j2 = "2.23.1"
@@ -22,6 +22,4 @@ object Versions {
     const val mockitoKotlin = "5.3.1"
     const val grizzly = "4.0.2"
     const val hamcrest = "2.2"
-
-    const val wrapper = "8.4"
 }

radar-gateway/gateway.yml

@@ -0,0 +1,55 @@
+# Resource config class
+#resourceConfig: org.radarbase.gateway.inject.ManagementPortalEnhancerFactory
+
+server:
+  # URI to serve data to
+  baseUri: http://0.0.0.0:8090/radar-gateway/
+  # Maximum number of simultaneous requests to Kafka.
+  #maxRequests: 200
+  # Maximum request content length, also when decompressed.
+  # This protects against memory overflows.
+  #maxRequestSize: 25165824
+  # Whether JMX should be enabled. Disable if not needed, for higher performance.
+  #isJmxEnabled: true
+
+kafka:
+  # Number of Kafka brokers to keep in a pool for reuse in multiple requests.
+  # poolSize: 20
+  # Kafka producer settings. Read from https://kafka.apache.org/documentation/#producerconfigs.
+  producer:
+    bootstrap.servers: kafka-1:9092
+    security.protocol: PLAINTEXT
+    # Kafka Admin Client settings. Read from https://kafka.apache.org/documentation/#adminclientconfigs.
+    #admin:
+    # bootstrap server property is copied from the producer settings if none is provided.
+    #bootstrap.servers: kafka-1:9092
+  # Kafka serialization settings, used in KafkaAvroSerializer. Read from [io.confluent.kafka.serializers.AbstractKafkaSchemaSerDeConfig].
+  serialization:
+    schema.registry.url: http://schema-registry:8081
+
+# Authorization settings
+auth:
+  # ManagementPortal URL. If available, this is used to read the public key from
+  # ManagementPortal directly. This is the recommended method of getting public key.
+  managementPortalUrl: http://managementportal:8080/managementportal
+    # Whether to check that the user that submits data has the reported source ID registered
+    # in the ManagementPortal.
+    #checkSourceId: true
+    # OAuth 2.0 resource name.
+    #resourceName: res_gateway
+    # OAuth 2.0 token issuer. If null, this is not checked.
+    #issuer: null
+    # Key store for checking the digital signature of OAuth 2.0 JWTs.
+    #keyStore:
+    # Path to the p12 key store.
+    #path: null
+    # Alias in the key store to use
+    #alias: null
+    # Password of the key store
+    #password: null
+    # Plain-text PEM public keys
+    #publicKeys:
+    # ECDSA public keys
+    #ecdsa: []
+    # RSA public keys
+  #rsa: []

radar-gateway/src/main/kotlin/org/radarbase/gateway/inject/SchemaRetrieverFactory.kt

@@ -3,15 +3,20 @@ package org.radarbase.gateway.inject
 import io.confluent.kafka.schemaregistry.client.SchemaRegistryClientConfig.USER_INFO_CONFIG
 import io.confluent.kafka.serializers.AbstractKafkaSchemaSerDeConfig.SCHEMA_REGISTRY_URL_CONFIG
 import io.confluent.kafka.serializers.AbstractKafkaSchemaSerDeConfig.SCHEMA_REGISTRY_USER_INFO_CONFIG
+import io.ktor.client.*
+import io.ktor.client.engine.cio.*
 import io.ktor.client.plugins.*
 import io.ktor.client.plugins.auth.*
 import io.ktor.client.plugins.auth.providers.*
+import io.ktor.client.request.*
 import jakarta.ws.rs.core.Context
 import org.radarbase.config.ServerConfig
 import org.radarbase.gateway.config.GatewayConfig
 import org.radarbase.producer.io.timeout
+import org.radarbase.producer.schema.SchemaRestClient
 import org.radarbase.producer.schema.SchemaRetriever
 import org.radarbase.producer.schema.SchemaRetriever.Companion.schemaRetriever
+import org.slf4j.LoggerFactory
 import java.util.function.Supplier
 import kotlin.time.Duration.Companion.seconds
 
@@ -32,26 +37,25 @@ class SchemaRetrieverFactory(
                 ?: config.kafka.serialization[USER_INFO_CONFIG].asNonEmptyString()
 
         return schemaRetriever(baseUrl = server.urlString) {
-            httpClient {
-                if (basicCredentials != null && basicCredentials.contains(':')) {
-                    install(Auth) {
-                        basic {
-                            credentials {
-                                val (username, password) = basicCredentials.split(':', limit = 2)
-                                BasicAuthCredentials(
-                                    username = username,
-                                    password = password,
-                                )
+            if (basicCredentials != null && basicCredentials.contains(':')) {
+                val (apiKey, apiSecret) = basicCredentials.split(':', limit = 2)
+                httpClient = HttpClient(CIO) {
+                    timeout(30.seconds)
+                        install(Auth) {
+                            basic {
+                                sendWithoutRequest { true }
+                                credentials {
+                                    BasicAuthCredentials(username = apiKey, password = apiSecret)
+                                }
                             }
                         }
                     }
-                    timeout(30.seconds)
                 }
-            }
         }
     }
 
     companion object {
         private fun Any?.asNonEmptyString(): String? = (this as? String)?.takeIf { it.isNotEmpty() }
+        private val logger = LoggerFactory.getLogger(SchemaRetrieverFactory::class.java)
     }
 }

radar-gateway/src/main/kotlin/org/radarbase/gateway/io/AvroProcessor.kt

@@ -132,7 +132,7 @@ class AvroProcessor(
                                     "Schema ID not found in subject",
                                 )
                             } else {
-                                throw HttpBadGatewayException("cannot get data from schema registry: ${ex.javaClass.simpleName}")
+                                throw HttpBadGatewayException("cannot get data from schema registry: ${ex}")
                             }
                         }
                         createMapping(topic, ofValue, parsedSchema.schema)