[BTA] Run in-process build operations in a thread pool

This way, ThreadLocal values would not be attached to the build system threads allowing the classloader used for BTA to be collected by GC earlier.
^KT-81414 Verification Pending

Author: ALikhachev

compiler/build-tools/kotlin-build-tools-impl/src/main/kotlin/org/jetbrains/kotlin/buildtools/internal/KotlinToolchainsImpl.kt

@@ -16,7 +16,12 @@ import org.jetbrains.kotlin.cli.jvm.compiler.KotlinCoreEnvironment
 import org.jetbrains.kotlin.cli.jvm.modules.CoreJrtFileSystem
 import org.jetbrains.kotlin.config.KotlinCompilerVersion
 import java.io.File
+import java.util.concurrent.Callable
 import java.util.concurrent.ConcurrentHashMap
+import java.util.concurrent.ExecutionException
+import java.util.concurrent.Executors
+import java.util.concurrent.Future
+import kotlin.lazy
 
 internal class KotlinToolchainsImpl() : KotlinToolchains {
     private val buildIdToSessionFlagFile: MutableMap<ProjectId, File> = ConcurrentHashMap()
@@ -47,6 +52,11 @@ internal class KotlinToolchainsImpl() : KotlinToolchains {
         override val projectId: ProjectId,
         private val buildIdToSessionFlagFile: MutableMap<ProjectId, File>,
     ) : KotlinToolchains.BuildSession {
+        private val executorDelegate = lazy {
+            Executors.newCachedThreadPool()
+        }
+        private val executor by executorDelegate
+
         override fun <R> executeOperation(operation: BuildOperation<R>): R {
             return executeOperation(operation, logger = null)
         }
@@ -57,11 +67,31 @@ internal class KotlinToolchainsImpl() : KotlinToolchains {
             logger: KotlinLogger?,
         ): R {
             check(operation is BuildOperationImpl<R>) { "Unknown operation type: ${operation::class.qualifiedName}" }
-            return operation.execute(projectId, executionPolicy, logger)
+            val operationBody: Callable<R> = { operation.execute(projectId, executionPolicy, logger) }
+            return if (executionPolicy is ExecutionPolicy.InProcess) {
+                unwrapExecutionException(executor.submit(operationBody))
+            } else {
+                operationBody.call()
+            }
+        }
+
+        /**
+         * Attempts to retrieve the result of the computation from the given `Future` instance.
+         * If the computation threw an exception, unwraps and rethrows the underlying cause of the exception.
+         */
+        private fun <R> unwrapExecutionException(result: Future<R>): R {
+            return try {
+                result.get()
+            } catch (e: ExecutionException) {
+                throw e.cause ?: e
+            }
         }
 
         override fun close() {
             clearJarCaches()
+            if (executorDelegate.isInitialized()) {
+                executor.shutdown()
+            }
             val file = buildIdToSessionFlagFile.remove(projectId) ?: return
             file.delete()
         }

libraries/tools/kotlin-maven-plugin-test/src/it/test-multimodule-in-process/invoker.properties

@@ -1,2 +1 @@
-invoker.environmentVariables.MAVEN_OPTS = -XX:MaxMetaspaceSize=150M
-invoker.buildResult = failure
+invoker.environmentVariables.MAVEN_OPTS = -XX:MaxMetaspaceSize=150M