fix: EXPOSED-1097 Make JDBC transaction context registry thread-safe

exposed-jdbc/src/main/kotlin/org/jetbrains/exposed/v1/jdbc/transactions/TransactionManager.kt

@@ -11,6 +11,7 @@ import org.jetbrains.exposed.v1.core.transactions.suspend.TransactionContextHold
 import org.jetbrains.exposed.v1.jdbc.Database
 import org.jetbrains.exposed.v1.jdbc.JdbcTransaction
 import org.jetbrains.exposed.v1.jdbc.statements.api.ExposedConnection
+import java.util.concurrent.ConcurrentHashMap
 import kotlin.coroutines.CoroutineContext
 
 /**
@@ -106,7 +107,7 @@ class TransactionManager(
          * Each transaction manager gets a unique context key for transaction isolation.
          */
         private val contextKeys =
-            mutableMapOf<JdbcTransactionManager, CoroutineContext.Key<TransactionContextHolder>>()
+            ConcurrentHashMap<JdbcTransactionManager, CoroutineContext.Key<TransactionContextHolder>>()
 
         /**
          * Returns the context key for the given transaction manager.

exposed-tests/src/test/kotlin/org/jetbrains/exposed/v1/tests/shared/CustomTransactionManagerTest.kt

@@ -13,11 +13,17 @@ import org.jetbrains.exposed.v1.tests.DatabaseTestsBase
 import org.jetbrains.exposed.v1.tests.TestDB
 import org.junit.jupiter.api.Assumptions
 import java.sql.Connection.TRANSACTION_READ_COMMITTED
+import java.util.concurrent.CountDownLatch
+import java.util.concurrent.Executors
+import java.util.concurrent.TimeUnit
+import java.util.concurrent.atomic.AtomicBoolean
 import kotlin.test.BeforeTest
 import kotlin.test.Test
 import kotlin.test.assertEquals
 import kotlin.test.assertFailsWith
 import kotlin.test.assertNotNull
+import kotlin.test.assertSame
+import kotlin.test.assertTrue
 
 class CustomTransactionManagerTest : DatabaseTestsBase() {
 
@@ -216,4 +222,47 @@ class CustomTransactionManagerTest : DatabaseTestsBase() {
             TransactionManager.getContextKey(customManager)
         }
     }
+
+    @OptIn(InternalApi::class)
+    @Test
+    fun testContextKeysRemainAvailableDuringConcurrentDatabaseRegistration() {
+        val databases = mutableListOf<Database>()
+        val readers = Executors.newFixedThreadPool(4)
+        val ready = CountDownLatch(4)
+        val start = CountDownLatch(1)
+        val finished = AtomicBoolean(false)
+
+        try {
+            val registeredManagers = List(1024) {
+                val database = createMockDatabase().also(databases::add)
+                val manager = TransactionManager.managerFor(database)
+                manager to TransactionManager.getContextKey(manager)
+            }
+            val results = List(4) {
+                readers.submit {
+                    ready.countDown()
+                    start.await()
+                    while (!finished.get()) {
+                        registeredManagers.forEach { (manager, key) ->
+                            assertSame(key, TransactionManager.getContextKey(manager))
+                        }
+                    }
+                }
+            }
+
+            assertTrue(ready.await(30, TimeUnit.SECONDS))
+            start.countDown()
+            repeat(16384) {
+                databases.add(createMockDatabase())
+            }
+            finished.set(true)
+            results.forEach { it.get(30, TimeUnit.SECONDS) }
+        } finally {
+            finished.set(true)
+            start.countDown()
+            readers.shutdownNow()
+            assertTrue(readers.awaitTermination(30, TimeUnit.SECONDS))
+            databases.asReversed().forEach(TransactionManager::closeAndUnregister)
+        }
+    }
 }