Merge branch 'release/1.11.5.0'

.gitignore

@@ -44,3 +44,5 @@ build/
 ### DEV
 _tmp/
 .repository/
+
+.console.sh

README.md

@@ -1,15 +1,18 @@
 # avro-kotlin
 
 Avro Kotlin provides a type- and null-safe type-system that encapsulates the apache-java avro-types and makes them 
-easily accessible from kotlin.
+easily accessible from kotlin. The mission is to become your one-stop lib when dealing with avro and kotlin.
 
 [![stable](https://img.shields.io/badge/lifecycle-STABLE-green.svg)](https://github.com/holisticon#open-source-lifecycle)
+[![Maven Central](https://maven-badges.herokuapp.com/maven-central/io.toolisticon.kotlin.avro/avro-kotlin-bom/badge.svg)](https://maven-badges.herokuapp.com/maven-central/io.toolisticon.kotlin.avro/avro-kotlin-bom)
+[![Kotlin](https://img.shields.io/badge/kotlin-2.0.0-blue.svg?logo=kotlin)](http://kotlinlang.org)
+[![Kotlinx serialization](https://img.shields.io/badge/kotlinx--serialization-1.7.1-blue?logo=kotlin)](https://github.com/Kotlin/kotlinx.serialization)
+[![Avro spec](https://img.shields.io/badge/avro%20spec-1.11.3-blue.svg?logo=apache)](https://avro.apache.org/docs/1.11.3/specification/)
 [![Build Status](https://github.com/toolisticon/avro-kotlin/workflows/Development%20branches/badge.svg)](https://github.com/toolisticon/avro-kotlin/actions)
 [![sponsored](https://img.shields.io/badge/sponsoredBy-Holisticon-RED.svg)](https://holisticon.de/)
-[![Maven Central](https://maven-badges.herokuapp.com/maven-central/io.toolisticon.kotlin.avro/avro-kotlin-bom/badge.svg)](https://maven-badges.herokuapp.com/maven-central/io.toolisticon.kotlin.avro/avro-kotlin-bom)
-
-Since Version 1.11.4.2 we use a multi module build. The GroupId changed, best use the provided BOM:
 
+> [!NOTE]
+> Since Version 1.11.4.2 we use a multi module build. The GroupId changed, best use the provided BOM:
 ```xml
 <dependency>
     <groupId>io.toolisticon.kotlin.avro</groupId>

_examples/java-example/pom.xml

@@ -5,7 +5,7 @@
   <parent>
     <groupId>io.toolisticon.kotlin.avro._</groupId>
     <artifactId>avro-kotlin-examples-root</artifactId>
-    <version>1.11.4.4</version>
+    <version>1.11.5.0</version>
   </parent>
 
   <groupId>io.toolisticon.kotlin.avro.examples</groupId>

_examples/java-example/src/test/java/io/toolisticon/kotlin/avro/example/java/BankAccountCreatedTest.java

@@ -1,12 +1,12 @@
 package io.toolisticon.kotlin.avro.example.java;
 
 import io.toolisticon.example.bank.BankAccountCreated;
+import io.toolisticon.kotlin.avro.repository.AvroSchemaResolverMap;
 import org.javamoney.moneta.Money;
 import org.junit.jupiter.api.Test;
 
 import java.util.UUID;
 
-import static io.toolisticon.kotlin.avro.AvroKotlin.avroSchemaResolver;
 import static io.toolisticon.kotlin.avro.codec.SpecificRecordCodec.specificRecordSingleObjectDecoder;
 import static io.toolisticon.kotlin.avro.codec.SpecificRecordCodec.specificRecordSingleObjectEncoder;
 import static org.assertj.core.api.Assertions.assertThat;
@@ -21,7 +21,7 @@ void encodeAndDecodeEventWithMoneyLogicalType() {
       .setCustomerId("1")
       .setInitialBalance(Money.of(100.123456, "EUR"))
       .build();
-    final var resolver = avroSchemaResolver(BankAccountCreated.getClassSchema());
+    final var resolver = new AvroSchemaResolverMap(BankAccountCreated.getClassSchema());
 
     final var encoded = specificRecordSingleObjectEncoder().encode(bankAccountCreated);
 

_examples/kotlin-example/pom.xml

@@ -5,7 +5,7 @@
   <parent>
     <groupId>io.toolisticon.kotlin.avro._</groupId>
     <artifactId>avro-kotlin-examples-root</artifactId>
-    <version>1.11.4.4</version>
+    <version>1.11.5.0</version>
   </parent>
 
   <groupId>io.toolisticon.kotlin.avro.examples</groupId>

_examples/kotlin-example/src/main/kotlin/BankAccountCreatedData.kt

@@ -1,18 +1,16 @@
 package io.toolisticon.kotlin.avro.example
 
-import com.github.avrokotlin.avro4k.AvroName
-import com.github.avrokotlin.avro4k.AvroNamespace
 import com.github.avrokotlin.avro4k.serializer.UUIDSerializer
 import io.toolisticon.kotlin.avro.example.customerid.CustomerId
 import io.toolisticon.kotlin.avro.example.customerid.CustomerIdLogicalType
 import io.toolisticon.kotlin.avro.example.money.MoneyLogicalType.MoneySerializer
+import kotlinx.serialization.SerialName
 import kotlinx.serialization.Serializable
 import org.javamoney.moneta.Money
 import java.util.*
 
 @Serializable
-@AvroName("BankAccountCreated")
-@AvroNamespace("io.toolisticon.bank")
+@SerialName("io.toolisticon.bank.BankAccountCreated")
 data class BankAccountCreatedData(
 
   @Serializable(with = UUIDSerializer::class)

_examples/kotlin-example/src/test/kotlin/BankAccountCreatedDataTest.kt

@@ -1,9 +1,9 @@
 package io.toolisticon.kotlin.avro.example
 
+import io.toolisticon.kotlin.avro.AvroKotlin
 import io.toolisticon.kotlin.avro.codec.GenericRecordCodec
 import io.toolisticon.kotlin.avro.example.customerid.CustomerId
 import io.toolisticon.kotlin.avro.example.money.MoneyLogicalType
-import io.toolisticon.kotlin.avro.repository.avroSchemaResolver
 import io.toolisticon.kotlin.avro.serialization.AvroKotlinSerialization
 import io.toolisticon.kotlin.avro.value.CanonicalName.Companion.toCanonicalName
 import io.toolisticon.kotlin.avro.value.Name.Companion.toName
@@ -17,7 +17,6 @@ internal class BankAccountCreatedDataTest {
   @Test
   fun `show schema`() {
     val schema = KotlinExample.avro.schema(BankAccountCreatedData::class)
-    println(schema)
     assertThat(schema.canonicalName).isEqualTo("io.toolisticon.bank.BankAccountCreated".toCanonicalName())
     assertThat(schema.fields).hasSize(3)
 
@@ -49,7 +48,7 @@ internal class BankAccountCreatedDataTest {
     val customerId = CustomerId.random()
 
     val event = BankAccountCreatedData(accountId, customerId, amount)
-    val resolver = avroSchemaResolver(KotlinExample.avro.schema(BankAccountCreatedData::class))
+    val resolver = AvroKotlin.avroSchemaResolver(KotlinExample.avro.schema(BankAccountCreatedData::class))
 
     val record = KotlinExample.avro.toRecord(event)
 

_examples/kotlin-example/src/test/kotlin/JavaKotlinInterOpTest.kt

@@ -1,10 +1,10 @@
 package io.toolisticon.kotlin.avro.example
 
 import io.toolisticon.example.bank.BankAccountCreated
+import io.toolisticon.kotlin.avro.AvroKotlin.avroSchemaResolver
 import io.toolisticon.kotlin.avro.codec.GenericRecordCodec
 import io.toolisticon.kotlin.avro.example.customerid.CustomerId
 import io.toolisticon.kotlin.avro.model.wrapper.AvroSchema
-import io.toolisticon.kotlin.avro.repository.avroSchemaResolver
 import io.toolisticon.kotlin.avro.serialization.AvroKotlinSerialization
 import io.toolisticon.kotlin.avro.value.SingleObjectEncodedBytes
 import org.assertj.core.api.Assertions.assertThat
@@ -54,6 +54,5 @@ internal class JavaKotlinInterOpTest {
     assertThat(decoded.accountId).isEqualTo(orig.accountId)
     assertThat(decoded.customerId).isEqualTo(orig.customerId)
     assertThat(decoded.initialBalance).isEqualTo(orig.initialBalance)
-
   }
 }

_examples/logical-type-customer-id/pom.xml

@@ -5,7 +5,7 @@
   <parent>
     <groupId>io.toolisticon.kotlin.avro._</groupId>
     <artifactId>avro-kotlin-examples-root</artifactId>
-    <version>1.11.4.4</version>
+    <version>1.11.5.0</version>
   </parent>
 
   <groupId>io.toolisticon.kotlin.avro.examples</groupId>

_examples/logical-type-money/pom.xml

@@ -5,7 +5,7 @@
   <parent>
     <groupId>io.toolisticon.kotlin.avro._</groupId>
     <artifactId>avro-kotlin-examples-root</artifactId>
-    <version>1.11.4.4</version>
+    <version>1.11.5.0</version>
   </parent>
 
   <groupId>io.toolisticon.kotlin.avro.examples</groupId>

_examples/pom.xml

@@ -5,7 +5,7 @@
   <parent>
     <groupId>io.toolisticon.kotlin.avro._</groupId>
     <artifactId>avro-kotlin-root</artifactId>
-    <version>1.11.4.4</version>
+    <version>1.11.5.0</version>
   </parent>
 
   <artifactId>avro-kotlin-examples-root</artifactId>

_mvn/bom/pom.xml

@@ -5,7 +5,7 @@
   <parent>
     <groupId>io.toolisticon.kotlin.avro._</groupId>
     <artifactId>avro-kotlin-root</artifactId>
-    <version>1.11.4.4</version>
+    <version>1.11.5.0</version>
     <relativePath>../../pom.xml</relativePath>
   </parent>
 

_mvn/coverage-aggregator/pom.xml

@@ -5,7 +5,7 @@
   <parent>
     <groupId>io.toolisticon.kotlin.avro._</groupId>
     <artifactId>avro-kotlin-root</artifactId>
-    <version>1.11.4.4</version>
+    <version>1.11.5.0</version>
     <relativePath>../../pom.xml</relativePath>
   </parent>
 

_mvn/parent/pom.xml

@@ -5,7 +5,7 @@
   <parent>
     <groupId>io.toolisticon.kotlin.avro._</groupId>
     <artifactId>avro-kotlin-root</artifactId>
-    <version>1.11.4.4</version>
+    <version>1.11.5.0</version>
     <relativePath>../../pom.xml</relativePath>
   </parent>
 

_test/pom.xml

@@ -5,7 +5,7 @@
   <parent>
     <groupId>io.toolisticon.kotlin.avro._</groupId>
     <artifactId>avro-kotlin-parent</artifactId>
-    <version>1.11.4.4</version>
+    <version>1.11.5.0</version>
     <relativePath>../_mvn/parent/pom.xml</relativePath>
   </parent>
 

avro-kotlin-serialization/pom.xml

@@ -5,7 +5,7 @@
   <parent>
     <groupId>io.toolisticon.kotlin.avro._</groupId>
     <artifactId>avro-kotlin-parent</artifactId>
-    <version>1.11.4.4</version>
+    <version>1.11.5.0</version>
     <relativePath>../_mvn/parent/pom.xml</relativePath>
   </parent>
 

avro-kotlin-serialization/src/main/kotlin/AvroKotlinSerialization.kt

@@ -1,75 +1,147 @@
 package io.toolisticon.kotlin.avro.serialization
 
-import com.github.avrokotlin.avro4k.Avro
+import com.github.avrokotlin.avro4k.*
 import io.toolisticon.kotlin.avro.AvroKotlin
+import io.toolisticon.kotlin.avro.codec.AvroCodec
 import io.toolisticon.kotlin.avro.codec.GenericRecordCodec
 import io.toolisticon.kotlin.avro.model.wrapper.AvroSchema
-import io.toolisticon.kotlin.avro.model.wrapper.AvroSchemaChecks.compatibleToReadFrom
 import io.toolisticon.kotlin.avro.repository.AvroSchemaResolver
+import io.toolisticon.kotlin.avro.repository.AvroSchemaResolverMutableMap
+import io.toolisticon.kotlin.avro.serialization.avro4k.avro4k
 import io.toolisticon.kotlin.avro.serialization.spi.AvroSerializationModuleFactoryServiceLoader
 import io.toolisticon.kotlin.avro.serialization.spi.SerializerModuleKtx.reduce
+import io.toolisticon.kotlin.avro.value.AvroSchemaCompatibilityMap
 import io.toolisticon.kotlin.avro.value.SingleObjectEncodedBytes
+import kotlinx.serialization.ExperimentalSerializationApi
 import kotlinx.serialization.KSerializer
 import kotlinx.serialization.modules.SerializersModule
+import kotlinx.serialization.serializer
 import mu.KLogging
 import org.apache.avro.generic.GenericData
 import org.apache.avro.generic.GenericRecord
+import java.lang.Runtime.Version
 import java.util.concurrent.ConcurrentHashMap
 import kotlin.reflect.KClass
+import kotlin.reflect.full.createType
 
+@OptIn(ExperimentalSerializationApi::class)
 class AvroKotlinSerialization(
-  private val avro4k: Avro
-) {
+  val avro4k: Avro,
+  private val schemaResolver: AvroSchemaResolverMutableMap = AvroSchemaResolverMutableMap.EMPTY,
+  private val genericData: GenericData = AvroKotlin.genericData
+) : AvroSchemaResolver by schemaResolver {
   companion object : KLogging() {
 
     fun configure(vararg serializersModules: SerializersModule): AvroKotlinSerialization {
       return AvroKotlinSerialization(
-        Avro(
+        Avro {
           serializersModule = serializersModules.toList().reduce()
-        )
+        }
       )
     }
   }
 
+  init {
+    /**
+     * We _need_ `kotlinx.serialization >= 1.7`. spring boot provides an outdated version (1.6.3).
+     * This is a pita to resolve. This check makes sure, any misconfigurations are found on app-start.
+     */
+    check(Version.parse(KSerializer::class.java.`package`.implementationVersion) >= Version.parse("1.7")) { "avro4k uses features that required kotlinx.serialization version >= 1.7.0. Make sure to include the correct versions, especially when you use spring-boot." }
+  }
+
+  val avro4kSingleObject = AvroSingleObject(schemaRegistry = schemaResolver.avro4k(), avro = avro4k)
+
+  val compatibilityCache = AvroSchemaCompatibilityMap()
+
   private val kserializerCache = ConcurrentHashMap<KClass<*>, KSerializer<*>>()
   private val schemaCache = ConcurrentHashMap<KClass<*>, AvroSchema>()
 
   constructor() : this(
-    Avro(AvroSerializationModuleFactoryServiceLoader())
+    Avro { serializersModule = AvroSerializationModuleFactoryServiceLoader() }
   )
 
-  fun serializer(type: KClass<*>) = kserializerCache.computeIfAbsent(type) { key ->
-    logger.trace { "add kserializer for $type." }
-    key.kserializer()
+  fun <T : Any> singleObjectEncoder(): AvroCodec.SingleObjectEncoder<T> = AvroCodec.SingleObjectEncoder { data ->
+    @Suppress("UNCHECKED_CAST")
+    val serializer = serializer(data::class) as KSerializer<T>
+    val writerSchema = schema(data::class)
+
+    val bytes = avro4kSingleObject.encodeToByteArray(writerSchema.get(), serializer, data)
+
+    SingleObjectEncodedBytes.of(bytes)
   }
 
-  fun schema(type: Class<*>): AvroSchema = schema(type.kotlin)
+  fun <T : Any> singleObjectDecoder(): AvroCodec.SingleObjectDecoder<T> = AvroCodec.SingleObjectDecoder { bytes ->
+    val writerSchema = schemaResolver[bytes.fingerprint]
+    val klass = AvroKotlin.loadClassForSchema<T>(writerSchema)
 
-  fun schema(type: KClass<*>): AvroSchema = schemaCache.computeIfAbsent(type) { key ->
-    logger.trace { "add schema for $type." }
-    AvroSchema(avro4k.schema(serializer(key)))
+    @Suppress("UNCHECKED_CAST")
+    avro4kSingleObject.decodeFromByteArray(serializer(klass), bytes.value) as T
+  }
+
+  fun <T : Any> genericRecordEncoder(): AvroCodec.GenericRecordEncoder<T> = AvroCodec.GenericRecordEncoder { data ->
+    @Suppress("UNCHECKED_CAST")
+    val serializer = serializer(data::class) as KSerializer<T>
+    val writerSchema = schema(data::class)
+
+    avro4k.encodeToGenericData(writerSchema.get(), serializer, data) as GenericRecord
+  }
+
+  /**
+   * @param klass - optional. If we do know the klass already, we can pass it to avoid a second lookup.
+   */
+  fun <T : Any> genericRecordDecoder(klass: KClass<T>? = null) = AvroCodec.GenericRecordDecoder { record ->
+    val writerSchema = AvroSchema(record.schema)
+    val readerKlass: KClass<T> = klass ?: AvroKotlin.loadClassForSchema(writerSchema)
+
+    @Suppress("UNCHECKED_CAST")
+    val kserializer = serializer(readerKlass) as KSerializer<T>
+    val readerSchema = schema(readerKlass)
+    val compatibility = compatibilityCache.compatibleToReadFrom(writerSchema, readerSchema)
+
+    require(compatibility.isCompatible) { "Reader/writer schema are incompatible." }
+
+    avro4k.decodeFromGenericData(writerSchema = writerSchema.get(), deserializer = kserializer, record)
   }
 
   @Suppress("UNCHECKED_CAST")
-  fun <T : Any> toRecord(data: T): GenericRecord {
+  fun <T : Any> toGenericRecord(data: T): GenericRecord {
     val kserializer = serializer(data::class) as KSerializer<T>
+    val schema = avro4k.schema(kserializer)
 
-    return avro4k.toRecord(kserializer, data)
+    return avro4k.encodeToGenericData(schema, kserializer, data) as GenericRecord
   }
 
-  inline fun <reified T : Any> fromRecord(record: GenericRecord): T = fromRecord(record, T::class)
+  fun <T : Any> toSingleObjectEncoded(data: T): SingleObjectEncodedBytes = singleObjectEncoder<T>().encode(data)
 
-  @Suppress("UNCHECKED_CAST")
-  fun <T : Any> fromRecord(record: GenericRecord, type: KClass<T>): T {
-    val writerSchema = AvroSchema(record.schema)
+  /**
+   * @return kotlinx-serializer for given class.
+   */
+  fun serializer(klass: KClass<*>) = kserializerCache.computeIfAbsent(klass) { key ->
+
+    require(klass.isSerializable())
+
+    // TODO: if we use SpecificRecords, we could derive the schema from the class directly
+    logger.trace { "add kserializer for $key." }
+
+    // TODO: createType takes a lot of optional args. We probably won't need them but at least we should check them.
+    val type = key.createType()
+
+    avro4k.serializersModule.serializer(type)
+  }
+
+  // TODO We would like to use the reified function from avro4k but we need to be able to dynamically load class by fqn
+  fun schema(type: KClass<*>): AvroSchema = schemaCache.computeIfAbsent(type) { key ->
+    logger.trace { "add schema for $type." }
+    AvroSchema(avro4k.schema(serializer(key))).also(this::registerSchema)
+  }
 
-    val kserializer = serializer(type) as KSerializer<T>
-    val readerSchema = schema(type)
+  @Deprecated("use toGenericRecord instead", ReplaceWith("toGenericRecord(data"))
+  fun <T : Any> toRecord(data: T): GenericRecord = toGenericRecord(data)
 
-    // TODO nicer?
-    require(readerSchema.compatibleToReadFrom(writerSchema).result.incompatibilities.isEmpty()) { "Reader/writer schema are incompatible" }
+  inline fun <reified T : Any> fromRecord(record: GenericRecord): T = fromRecord(record, T::class)
 
-    return avro4k.fromRecord(kserializer, record) as T
+  fun <T : Any> fromRecord(record: GenericRecord, type: KClass<T>): T {
+    return genericRecordDecoder(type).decode(record)
   }
 
   fun <T : Any> encodeSingleObject(
@@ -108,4 +180,8 @@ class AvroKotlinSerialization(
     readerType = T::class
   )
 
+  fun registerSchema(schema: AvroSchema): AvroKotlinSerialization = apply { schemaResolver + schema }
+
+  fun cachedSerializerClasses() = kserializerCache.keys().toList().toSet()
+  fun cachedSchemaClasses() = schemaCache.keys().toList().toSet()
 }