Repository: avro-kotlin/avro4k (kotlin)

Issue:
Severe Parallel Deserialization Performance Degradation Due to synchronized in Avro.RecordResolver.fieldCache[WeakIdentityHashmap].reap()
Describe the bug
When deserializing data in parallel using AvroSingleObject, there's a significant performance drop due to the synchronized usage in the Avro.RecordResolver cache `reap()` method. This becomes a major bottleneck in high-concurrency environments. Profiling indicates heavy thread contention on this synchronized block, which severely limits scalability.

Stack trace
No exception is thrown, but profiling shows blocked threads waiting on a monitor inside WeakIdentityHashmap.reap()

To Reproduce
Run the following test case. It demonstrates degraded performance when running parallel deserialization tasks:

```
import com.github.avrokotlin.avro4k.Avro
import com.github.avrokotlin.avro4k.AvroSingleObject
import com.github.avrokotlin.avro4k.encodeToByteArray
import io.github.oshai.kotlinlogging.KotlinLogging
import kotlinx.coroutines.async
import kotlinx.coroutines.newFixedThreadPoolContext
import kotlinx.coroutines.runBlocking
import kotlinx.serialization.Serializable
import kotlin.random.Random
import kotlin.test.Test
import kotlin.time.measureTime

@Serializable
private data class TestContainer(val list: List<TestData>) {
    constructor() : this((0..100_000).map { TestData() })
}

@Serializable
private data class TestData(
    val a: Long, val b: Long, val c: Long, val d: Long, val e: Long,
) {
    constructor() : this(
        Random.nextLong(), Random.nextLong(), Random.nextLong(),
        Random.nextLong(), Random.nextLong()
    )
}

class ConcurrencyTest {
    private val data = (0..100).map { TestContainer() }
    private val serializer = TestContainerSerializer()
    private val serializedData: List<ByteArray> = data.map { serializer.serialize(it) }

    @Test
    fun run(): Unit = runBlocking {
        val context = newFixedThreadPoolContext(20, "test-pool")
        serializedData.chunked(5)
            .map {
                async(context) {
                    measured {
                        val serializer = TestContainerSerializer()
                        it.map { serializer.deserialize(it) }
                    }
                }
            }.forEach { it.await() }
    }
}

private class TestContainerSerializer {
    val serializer = TestContainer.serializer()
    val schema = Avro.schema(serializer.descriptor)

    val avroSingleObject = AvroSingleObject(
        schemaRegistry = { _ -> schema },
        // Commenting out the following line causes performance degradation:
        // avro = Avro {}
    )

    fun serialize(data: TestContainer): ByteArray =
        avroSingleObject.encodeToByteArray(schema, serializer, data)

    fun deserialize(data: ByteArray): TestContainer =
        avroSingleObject.decodeFromByteArray(serializer, data)
}

private fun <T> measured(block: () -> T): T {
    var result: T? = null
    val time = measureTime { result = block() }
    logger.info { "time: ${time.inWholeMilliseconds} ms" }
    return result!!
}

private val logger = KotlinLogging.logger {}
```

Expected behavior
Parallel deserialization should scale with the number of threads without unnecessary contention. There should be no need to manually create a separate Avro instance for each thread to avoid the bottleneck.

Desktop (please complete the following information):

Kotlin version: 2.1.0

Avro4k version: 2.2.0

kotlinx.serialization version: 1.7.0

Additional context
As a workaround, instantiating a new Avro instance per thread mitigates the issue. However, this is not ideal and introduces unnecessary overhead and lifecycle management complexity.

I add sampler results when running with / without reinitialization of Avro. When running with single Avro across all threads (single RecordResolver with single cache) tasks are blocked f.e. 3000 ms in method reap() while processing takes only 200 ms. Using more threads causes ~exponential performance drops. 

![Image](https://github.com/user-attachments/assets/aa3300e3-6eb1-4fc0-a2f9-ae0302a335bc)

![Image](https://github.com/user-attachments/assets/678923ca-a6de-4f42-84f2-0232bbfe64f7)

Pull request description:
Solves an issue of performance where WeakIdentityHashMap uses too early a synchronised block. Closes #303

Patch:
diff --git a/benchmark/README.md b/benchmark/README.md
index 66a95698..ecf850a0 100644
--- a/benchmark/README.md
+++ b/benchmark/README.md
@@ -26,34 +26,34 @@ Computer: Macbook air M2
 
 ```
 Benchmark                                                       Mode   Cnt     Score       Error   Units
-c.g.a.b.complex.Avro4kBenchmark.read                            thrpt    5  27070.886 ±  2562.606  ops/s
-c.g.a.b.complex.ApacheAvroReflectBenchmark.read                 thrpt    5  26239.615 ± 15033.290  ops/s
-c.g.a.b.complex.Avro4kGenericWithApacheAvroBenchmark.read       thrpt    5  16140.138 ±    90.523  ops/s
-
-c.g.a.b.complex.Avro4kBenchmark.write                           thrpt    5  54488.317 ±  719.412   ops/s
-c.g.a.b.complex.ApacheAvroReflectBenchmark.write                thrpt    5  47510.885 ± 2467.348   ops/s
-c.g.a.b.complex.JacksonAvroBenchmark.write                      thrpt    5  33936.765 ± 2139.528   ops/s
-c.g.a.b.complex.Avro4kGenericWithApacheAvroBenchmark.write      thrpt    5  24072.277 ±  697.493   ops/s
-
-c.g.a.b.simple.Avro4kSimpleBenchmark.read                       thrpt    5  221277.895 ±  3945.928  ops/s
-c.g.a.b.simple.ApacheAvroReflectSimpleBenchmark.read            thrpt    5  230744.377 ± 30164.628  ops/s
-c.g.a.b.simple.Avro4kGenericWithApacheAvroSimpleBenchmark.read  thrpt    5  138394.796 ±  5130.421  ops/s
-c.g.a.b.simple.JacksonAvroSimpleBenchmark.read                  thrpt    5   69615.099 ±  4047.717  ops/s
-
-c.g.a.b.simple.Avro4kSimpleBenchmark.write                      thrpt    5  446673.090 ± 15520.264  ops/s
-c.g.a.b.simple.ApacheAvroReflectSimpleBenchmark.write           thrpt    5  320367.673 ± 33394.537  ops/s
-c.g.a.b.simple.Avro4kGenericWithApacheAvroSimpleBenchmark.write thrpt    5  168702.542 ±  5553.797  ops/s
-c.g.a.b.simple.JacksonAvroSimpleBenchmark.write                 thrpt    5  138898.312 ±  9156.715  ops/s
-
-JacksonAvroListsBenchmark.read                  thrpt    5  7.441 ± 1.575  ops/s
-Avro4kListsBenchmark.read                       thrpt    5  6.443 ± 1.822  ops/s
-Avro4kGenericWithApacheAvroListsBenchmark.read  thrpt    5  5.166 ± 0.120  ops/s
-ApacheAvroReflectListsBenchmark.read            thrpt    5  4.461 ± 0.520  ops/s
-
-ApacheAvroReflectListsBenchmark.write            thrpt    5  7.199 ± 1.409  ops/s
-Avro4kListsBenchmark.write                       thrpt    5  6.140 ± 0.200  ops/s
-JacksonAvroListsBenchmark.write                  thrpt    5  5.247 ± 0.258  ops/s
-Avro4kGenericWithApacheAvroListsBenchmark.write  thrpt    5  4.520 ± 1.391  ops/s
+ApacheAvroReflectBenchmark.read            thrpt    5  28819.641 ± 1048.236  ops/s
+Avro4kBenchmark.read                       thrpt    5  27259.164 ±  418.829  ops/s
+Avro4kGenericWithApacheAvroBenchmark.read  thrpt    5  16242.380 ±  123.109  ops/s
+
+Avro4kBenchmark.write                       thrpt    5  55760.544 ±  394.596  ops/s
+ApacheAvroReflectBenchmark.write            thrpt    5  48015.770 ± 4021.317  ops/s
+JacksonAvroBenchmark.write                  thrpt    5  34701.945 ±  719.865  ops/s
+Avro4kGenericWithApacheAvroBenchmark.write  thrpt    5  23859.446 ± 3207.023  ops/s
+
+Avro4kSimpleBenchmark.read                       thrpt    5  224766.920 ± 21004.000  ops/s
+ApacheAvroReflectSimpleBenchmark.read            thrpt    5  214986.578 ± 23138.102  ops/s
+Avro4kGenericWithApacheAvroSimpleBenchmark.read  thrpt    5  146503.504 ±  2613.945  ops/s
+JacksonAvroSimpleBenchmark.read                  thrpt    5   66071.428 ±  1571.426  ops/s
+
+Avro4kSimpleBenchmark.write                       thrpt    5  448433.701 ±  2086.119  ops/s
+ApacheAvroReflectSimpleBenchmark.write            thrpt    5  245640.080 ±  3835.344  ops/s
+Avro4kGenericWithApacheAvroSimpleBenchmark.write  thrpt    5  158308.148 ± 12314.688  ops/s
+JacksonAvroSimpleBenchmark.write                  thrpt    5  125374.072 ±  5240.116  ops/s
+
+JacksonAvroListsBenchmark.read                  thrpt    5  6.329 ± 0.631  ops/s
+Avro4kListsBenchmark.read                       thrpt    5  5.074 ± 1.624  ops/s
+ApacheAvroReflectListsBenchmark.read            thrpt    5  4.048 ± 0.366  ops/s
+Avro4kGenericWithApacheAvroListsBenchmark.read  thrpt    5  3.709 ± 1.694  ops/s
+
+ApacheAvroReflectListsBenchmark.write            thrpt    5  7.332 ± 0.532  ops/s
+Avro4kListsBenchmark.write                       thrpt    5  6.074 ± 0.087  ops/s
+JacksonAvroListsBenchmark.write                  thrpt    5  3.590 ± 1.973  ops/s
+Avro4kGenericWithApacheAvroListsBenchmark.write  thrpt    5  3.406 ± 0.900  ops/s
 
 ```
 
diff --git a/benchmark/build.gradle.kts b/benchmark/build.gradle.kts
index 270f76fb..524eea2b 100644
--- a/benchmark/build.gradle.kts
+++ b/benchmark/build.gradle.kts
@@ -66,6 +66,7 @@ dependencies {
     implementation("com.fasterxml.jackson.module:jackson-module-kotlin:$jacksonVersion")
     implementation("com.fasterxml.jackson.datatype:jackson-datatype-jsr310:$jacksonVersion")
     implementation("com.fasterxml.jackson.dataformat:jackson-dataformat-avro:$jacksonVersion")
+    implementation("org.slf4j:slf4j-simple:2.0.17")
 
     implementation(project(":"))
 }
diff --git a/src/main/kotlin/com/github/avrokotlin/avro4k/Avro.kt b/src/main/kotlin/com/github/avrokotlin/avro4k/Avro.kt
index 84ddfdaf..bc067870 100644
--- a/src/main/kotlin/com/github/avrokotlin/avro4k/Avro.kt
+++ b/src/main/kotlin/com/github/avrokotlin/avro4k/Avro.kt
@@ -22,7 +22,7 @@ import kotlinx.serialization.serializer
 import okio.Buffer
 import org.apache.avro.Schema
 import org.apache.avro.io.DecoderFactory
-import org.apache.avro.util.WeakIdentityHashMap
+import java.util.WeakHashMap
 
 /**
  * The goal of this class is to serialize and deserialize in avro binary format, not in GenericRecords.
@@ -31,10 +31,7 @@ public sealed class Avro(
     public val configuration: AvroConfiguration,
     public final override val serializersModule: SerializersModule,
 ) : BinaryFormat {
-    // We use the identity hash map because we could have multiple descriptors with the same name, especially
-    // when having 2 different version of the schema for the same name. kotlinx-serialization is instantiating the descriptors
-    // only once, so we are safe in the main use cases. Combined with weak references to avoid memory leaks.
-    private val schemaCache: MutableMap<SerialDescriptor, Schema> = WeakIdentityHashMap()
+    private val schemaCache: MutableMap<SerialDescriptor, Schema> = WeakHashMap()
 
     internal val recordResolver = RecordResolver(this)
     internal val polymorphicResolver = PolymorphicResolver(serializersModule)
diff --git a/src/main/kotlin/com/github/avrokotlin/avro4k/internal/EnumResolver.kt b/src/main/kotlin/com/github/avrokotlin/avro4k/internal/EnumResolver.kt
index 4f21a9a3..3e754d63 100644
--- a/src/main/kotlin/com/github/avrokotlin/avro4k/internal/EnumResolver.kt
+++ b/src/main/kotlin/com/github/avrokotlin/avro4k/internal/EnumResolver.kt
@@ -2,10 +2,10 @@ package com.github.avrokotlin.avro4k.internal
 
 import com.github.avrokotlin.avro4k.AvroEnumDefault
 import kotlinx.serialization.descriptors.SerialDescriptor
-import org.apache.avro.util.WeakIdentityHashMap
+import java.util.WeakHashMap
 
 internal class EnumResolver {
-    private val defaultIndexCache: MutableMap<SerialDescriptor, EnumDefault> = WeakIdentityHashMap()
+    private val defaultIndexCache: MutableMap<SerialDescriptor, EnumDefault> = WeakHashMap()
 
     private data class EnumDefault(val index: Int?)
 
diff --git a/src/main/kotlin/com/github/avrokotlin/avro4k/internal/PolymorphicResolver.kt b/src/main/kotlin/com/github/avrokotlin/avro4k/internal/PolymorphicResolver.kt
index 20b1b661..5f26b746 100644
--- a/src/main/kotlin/com/github/avrokotlin/avro4k/internal/PolymorphicResolver.kt
+++ b/src/main/kotlin/com/github/avrokotlin/avro4k/internal/PolymorphicResolver.kt
@@ -3,10 +3,10 @@ package com.github.avrokotlin.avro4k.internal
 import com.github.avrokotlin.avro4k.AvroAlias
 import kotlinx.serialization.descriptors.SerialDescriptor
 import kotlinx.serialization.modules.SerializersModule
-import org.apache.avro.util.WeakIdentityHashMap
+import java.util.WeakHashMap
 
 internal class PolymorphicResolver(private val serializersModule: SerializersModule) {
-    private val cache = WeakIdentityHashMap<SerialDescriptor, Map<String, String>>()
+    private val cache = WeakHashMap<SerialDescriptor, Map<String, String>>()
 
     fun getFullNamesAndAliasesToSerialName(descriptor: SerialDescriptor): Map<String, String> {
         return cache.getOrPut(descriptor) {
diff --git a/src/main/kotlin/com/github/avrokotlin/avro4k/internal/RecordResolver.kt b/src/main/kotlin/com/github/avrokotlin/avro4k/internal/RecordResolver.kt
index 95269416..f49cc499 100644
--- a/src/main/kotlin/com/github/avrokotlin/avro4k/internal/RecordResolver.kt
+++ b/src/main/kotlin/com/github/avrokotlin/avro4k/internal/RecordResolver.kt
@@ -21,7 +21,6 @@ import kotlinx.serialization.json.int
 import kotlinx.serialization.json.long
 import org.apache.avro.Schema
 import org.apache.avro.generic.GenericData
-import org.apache.avro.util.WeakIdentityHashMap
 import java.util.WeakHashMap
 
 internal class RecordResolver(
@@ -32,7 +31,7 @@ internal class RecordResolver(
      *
      * Note: We use the descriptor in the key as we could have multiple descriptors for the same record schema, and multiple record schemas for the same descriptor.
      */
-    private val fieldCache: MutableMap<SerialDescriptor, MutableMap<Schema, SerializationWorkflow>> = WeakIdentityHashMap()
+    private val fieldCache: MutableMap<Pair<SerialDescriptor, Schema>, SerializationWorkflow> = WeakHashMap()
 
     /**
      * Maps the class fields to the schema fields.
@@ -53,7 +52,7 @@ internal class RecordResolver(
         if (classDescriptor.elementsCount == 0 && writerSchema.fields.isEmpty()) {
             return SerializationWorkflow.EMPTY
         }
-        return fieldCache.getOrPut(classDescriptor) { WeakHashMap() }.getOrPut(writerSchema) {
+        return fieldCache.getOrPut(classDescriptor to writerSchema) {
             loadCache(classDescriptor, writerSchema)
         }
     }


Test patch:
diff --git a/src/test/kotlin/com/github/avrokotlin/avro4k/encoding/AvroFixedEncodingTest.kt b/src/test/kotlin/com/github/avrokotlin/avro4k/encoding/AvroFixedEncodingTest.kt
index 30b05a65..db60972c 100644
--- a/src/test/kotlin/com/github/avrokotlin/avro4k/encoding/AvroFixedEncodingTest.kt
+++ b/src/test/kotlin/com/github/avrokotlin/avro4k/encoding/AvroFixedEncodingTest.kt
@@ -126,7 +126,7 @@ internal class AvroFixedEncodingTest : StringSpec({
     )
 
     @Serializable
-    @SerialName("Fixed")
+    @SerialName("FieldPriorToValueClass")
     private data class FieldPriorToValueClass(
         @AvroFixed(5) val mystring: FixedStringValueClass,
     )
diff --git a/src/test/resources/fixed_string_5.json b/src/test/resources/fixed_string_5.json
index b0144ce2..ecbd65cf 100644
--- a/src/test/resources/fixed_string_5.json
+++ b/src/test/resources/fixed_string_5.json
@@ -1,6 +1,6 @@
 {
   "type": "record",
-  "name": "Fixed",
+  "name": "FieldPriorToValueClass",
   "fields": [
     {
       "name": "mystring",