Fix OuraServiceUserRepository serialization issues

Author: mpgxvii

kafka-connect-oura-source/build.gradle.kts

@@ -5,14 +5,21 @@ dependencies {
     api(project(":oura-library"))
     api("io.confluent:kafka-connect-avro-converter:${Versions.confluent}")
     api("org.radarbase:radar-schemas-commons:${Versions.radarSchemas}")
-    implementation("org.radarbase:oauth-client-util:${Versions.managementPortal}")
+    implementation("org.radarbase:radar-commons-kotlin:${Versions.radarCommons}")
 
     implementation(platform("com.fasterxml.jackson:jackson-bom:${Versions.jackson}"))
     implementation("com.fasterxml.jackson.dataformat:jackson-dataformat-yaml")
     implementation("com.fasterxml.jackson.datatype:jackson-datatype-jsr310")
     implementation("com.google.firebase:firebase-admin:${Versions.firebaseAdmin}")
     implementation("org.jetbrains.kotlin:kotlin-stdlib-jdk8:1.8.21")
 
+    implementation("io.ktor:ktor-client-auth:${Versions.ktor}")
+    implementation("io.ktor:ktor-client-content-negotiation:${Versions.ktor}")
+    implementation("io.ktor:ktor-serialization-jackson:${Versions.ktor}")
+    implementation("io.ktor:ktor-client-cio-jvm:${Versions.ktor}")
+    implementation("io.ktor:ktor-serialization-kotlinx-json:${Versions.ktor}")
+    implementation("com.fasterxml.jackson.module:jackson-module-kotlin:${Versions.jackson}")
+
     // Included in connector runtime
     compileOnly("org.apache.kafka:connect-api:${Versions.kafka}")
     compileOnly(platform("com.fasterxml.jackson:jackson-bom:${Versions.jackson}"))

kafka-connect-oura-source/src/main/java/org/radarbase/connect/rest/oura/user/OuraServiceUserRepository.kt

@@ -16,9 +16,11 @@
  */
 package org.radarbase.connect.rest.oura.user
 
+import com.fasterxml.jackson.databind.ObjectMapper
 import com.fasterxml.jackson.datatype.jsr310.JavaTimeModule
+import com.fasterxml.jackson.module.kotlin.readValue
+import com.fasterxml.jackson.module.kotlin.registerKotlinModule
 import io.ktor.client.HttpClient
-import io.ktor.client.call.body
 import io.ktor.client.engine.cio.CIO
 import io.ktor.client.plugins.HttpTimeout
 import io.ktor.client.plugins.auth.Auth
@@ -41,9 +43,11 @@ import io.ktor.http.contentType
 import io.ktor.http.isSuccess
 import io.ktor.http.takeFrom
 import io.ktor.serialization.jackson.jackson
+import io.ktor.serialization.kotlinx.json.json
 import kotlinx.coroutines.Dispatchers
 import kotlinx.coroutines.runBlocking
 import kotlinx.coroutines.withContext
+import kotlinx.serialization.json.Json
 import org.radarbase.connect.rest.oura.OuraRestSourceConnectorConfig
 import org.radarbase.kotlin.coroutines.CacheConfig
 import org.radarbase.kotlin.coroutines.CachedSet
@@ -69,6 +73,7 @@ class OuraServiceUserRepository : UserRepository {
     private val credentialCaches = ConcurrentHashMap<String, CachedValue<OAuth2UserCredentials>>()
     private val credentialCacheConfig =
         CacheConfig(refreshDuration = 1.days, retryDuration = 1.minutes)
+    private val mapper = ObjectMapper().registerKotlinModule().registerModule(JavaTimeModule())
 
     @Throws(IOException::class)
     override fun get(key: String): User = runBlocking(Dispatchers.Default) {
@@ -86,18 +91,17 @@ class OuraServiceUserRepository : UserRepository {
             clientId = config.ouraUserRepositoryClientId,
             clientSecret = config.ouraUserRepositoryClientSecret,
         )
-        
 
         userCache = CachedSet(
             CacheConfig(refreshDuration = 1.hours, retryDuration = 1.minutes),
         ) {
             makeRequest<OuraUsers> { url("users?source-type=Oura") }
                 .users
                 .toHashSet()
-            // .filterTo(HashSet()) { u ->
-            //     u.isComplete() &&
-            //         (containedUsers.isEmpty() || u.versionedId in containedUsers)
-            // }
+                .filterTo(HashSet()) { u ->
+                    u.isComplete() &&
+                        (containedUsers.isEmpty() || u.versionedId in containedUsers)
+                }
         }
     }
 
@@ -118,6 +122,13 @@ class OuraServiceUserRepository : UserRepository {
                     baseUrl.host,
                 )
             }
+            install(ContentNegotiation) {
+                json(
+                    Json {
+                        ignoreUnknownKeys = true
+                    },
+                )
+            }
         } else if (clientId != null && clientSecret != null) {
             install(Auth) {
                 basic {
@@ -244,7 +255,7 @@ class OuraServiceUserRepository : UserRepository {
             }
             throw HttpResponseException(message, response.status.value)
         }
-        response.body<T>()
+        mapper.readValue<T>(response.bodyAsText())
     }
 
     companion object {