debuggability improvements to the CDK (#37746)

A few changes to improve debuggability:
1) We now print the stack trace that created an orphan thread, in addition to the current stack trace.
2) we add the ability for a connector to tell the integration runner to not care about certain threads (by adding a thread filter)
3) We change the log of connector containers to include the name of the test that created it (usually each test will create and close a container)
4) We also log when the timeout timer gets triggered, to try to understand why some tests are still taking way more time than their timeout should allow

Another change that was initially in there but has been removed and might be worth thinking about:
Today, only Write checks for live threads on shutdown. We might want to do that for other operations as well

Author: stephane-airbyte

airbyte-cdk/java/airbyte-cdk/core/src/main/kotlin/io/airbyte/cdk/integrations/base/IntegrationRunner.kt

@@ -25,13 +25,13 @@ import io.airbyte.protocol.models.v0.AirbyteMessage
 import io.airbyte.protocol.models.v0.ConfiguredAirbyteCatalog
 import io.airbyte.validation.json.JsonSchemaValidator
 import java.io.*
+import java.lang.reflect.Method
 import java.nio.charset.StandardCharsets
 import java.nio.file.Path
+import java.time.Instant
 import java.util.*
 import java.util.concurrent.*
 import java.util.function.Consumer
-import java.util.function.Predicate
-import java.util.stream.Collectors
 import org.apache.commons.lang3.ThreadUtils
 import org.apache.commons.lang3.concurrent.BasicThreadFactory
 import org.slf4j.Logger
@@ -84,6 +84,7 @@ internal constructor(
             (destination != null) xor (source != null),
             "can only pass in a destination or a source"
         )
+        threadCreationInfo.set(ThreadCreationInfo())
         this.cliParser = cliParser
         this.outputRecordCollector = outputRecordCollector
         // integration iface covers the commands that are the same for both source and destination.
@@ -189,17 +190,20 @@ internal constructor(
                     }
                 }
                 Command.WRITE -> {
-                    val config = parseConfig(parsed.getConfigPath())
-                    validateConfig(integration.spec().connectionSpecification, config, "WRITE")
-                    // save config to singleton
-                    DestinationConfig.Companion.initialize(
-                        config,
-                        (integration as Destination).isV2Destination
-                    )
-                    val catalog =
-                        parseConfig(parsed.getCatalogPath(), ConfiguredAirbyteCatalog::class.java)!!
-
                     try {
+                        val config = parseConfig(parsed.getConfigPath())
+                        validateConfig(integration.spec().connectionSpecification, config, "WRITE")
+                        // save config to singleton
+                        DestinationConfig.Companion.initialize(
+                            config,
+                            (integration as Destination).isV2Destination
+                        )
+                        val catalog =
+                            parseConfig(
+                                parsed.getCatalogPath(),
+                                ConfiguredAirbyteCatalog::class.java
+                            )!!
+
                         destination!!
                             .getSerializedMessageConsumer(config, catalog, outputRecordCollector)
                             .use { consumer -> consumeWriteStream(consumer!!) }
@@ -339,11 +343,37 @@ internal constructor(
         }
     }
 
+    class ThreadCreationInfo {
+        val stack: List<StackTraceElement> = Thread.currentThread().stackTrace.asList()
+        val time: Instant = Instant.now()
+        override fun toString(): String {
+            return "creationStack=${stack.joinToString("\n  ")}\ncreationTime=$time"
+        }
+    }
+
     companion object {
         private val LOGGER: Logger = LoggerFactory.getLogger(IntegrationRunner::class.java)
+        private val threadCreationInfo: InheritableThreadLocal<ThreadCreationInfo> =
+            object : InheritableThreadLocal<ThreadCreationInfo>() {
+                override fun childValue(parentValue: ThreadCreationInfo): ThreadCreationInfo {
+                    return ThreadCreationInfo()
+                }
+            }
 
         const val TYPE_AND_DEDUPE_THREAD_NAME: String = "type-and-dedupe"
 
+        // ThreadLocal.get(Thread) is private. So we open it and keep a reference to the
+        // opened method
+        private val getMethod: Method =
+            ThreadLocal::class.java.getDeclaredMethod("get", Thread::class.java).also {
+                it.isAccessible = true
+            }
+
+        @JvmStatic
+        fun getThreadCreationInfo(thread: Thread): ThreadCreationInfo {
+            return getMethod.invoke(threadCreationInfo, thread) as ThreadCreationInfo
+        }
+
         /**
          * Filters threads that should not be considered when looking for orphaned threads at
          * shutdown of the integration runner.
@@ -353,11 +383,12 @@ internal constructor(
          * active so long as the database connection pool is open.
          */
         @VisibleForTesting
-        val ORPHANED_THREAD_FILTER: Predicate<Thread> = Predicate { runningThread: Thread ->
-            (runningThread.name != Thread.currentThread().name &&
-                !runningThread.isDaemon &&
-                TYPE_AND_DEDUPE_THREAD_NAME != runningThread.name)
-        }
+        private val orphanedThreadPredicates: MutableList<(Thread) -> Boolean> =
+            mutableListOf({ runningThread: Thread ->
+                (runningThread.name != Thread.currentThread().name &&
+                    !runningThread.isDaemon &&
+                    TYPE_AND_DEDUPE_THREAD_NAME != runningThread.name)
+            })
 
         const val INTERRUPT_THREAD_DELAY_MINUTES: Int = 1
         const val EXIT_THREAD_DELAY_MINUTES: Int = 2
@@ -398,6 +429,15 @@ internal constructor(
             LOGGER.info("Finished buffered read of input stream")
         }
 
+        @JvmStatic
+        fun addOrphanedThreadFilter(predicate: (Thread) -> (Boolean)) {
+            orphanedThreadPredicates.add(predicate)
+        }
+
+        fun filterOrphanedThread(thread: Thread): Boolean {
+            return orphanedThreadPredicates.all { it(thread) }
+        }
+
         /**
          * Stops any non-daemon threads that could block the JVM from exiting when the main thread
          * is done.
@@ -425,11 +465,7 @@ internal constructor(
         ) {
             val currentThread = Thread.currentThread()
 
-            val runningThreads =
-                ThreadUtils.getAllThreads()
-                    .stream()
-                    .filter(ORPHANED_THREAD_FILTER)
-                    .collect(Collectors.toList())
+            val runningThreads = ThreadUtils.getAllThreads().filter(::filterOrphanedThread).toList()
             if (runningThreads.isNotEmpty()) {
                 LOGGER.warn(
                     """
@@ -450,7 +486,10 @@ internal constructor(
                             .build()
                     )
                 for (runningThread in runningThreads) {
-                    val str = "Active non-daemon thread: " + dumpThread(runningThread)
+                    val str =
+                        "Active non-daemon thread: " +
+                            dumpThread(runningThread) +
+                            "\ncreationStack=${getThreadCreationInfo(runningThread)}"
                     LOGGER.warn(str)
                     // even though the main thread is already shutting down, we still leave some
                     // chances to the children

airbyte-cdk/java/airbyte-cdk/core/src/main/resources/version.properties

@@ -1 +1 @@
-version=0.31.5
+version=0.31.6

airbyte-cdk/java/airbyte-cdk/core/src/test/kotlin/io/airbyte/cdk/integrations/base/IntegrationRunnerTest.kt

@@ -477,7 +477,7 @@ ${Jsons.serialize(message2)}""".toByteArray(
         val runningThreads =
             ThreadUtils.getAllThreads()
                 .stream()
-                .filter(IntegrationRunner.ORPHANED_THREAD_FILTER)
+                .filter(IntegrationRunner::filterOrphanedThread)
                 .collect(Collectors.toList())
         // all threads should be interrupted
         Assertions.assertEquals(listOf<Any>(), runningThreads)
@@ -505,7 +505,7 @@ ${Jsons.serialize(message2)}""".toByteArray(
         val runningThreads =
             ThreadUtils.getAllThreads()
                 .stream()
-                .filter(IntegrationRunner.ORPHANED_THREAD_FILTER)
+                .filter(IntegrationRunner::filterOrphanedThread)
                 .collect(Collectors.toList())
         // a thread that refuses to be interrupted should remain
         Assertions.assertEquals(1, runningThreads.size)

airbyte-cdk/java/airbyte-cdk/core/src/testFixtures/kotlin/io/airbyte/cdk/extensions/LoggingInvocationInterceptor.kt

@@ -46,18 +46,18 @@ class LoggingInvocationInterceptor : InvocationInterceptor {
                         ExtensionContext::class.java
                     ) == null
             ) {
-                LOGGER!!.error(
+                LOGGER.error(
                     "Junit LoggingInvocationInterceptor executing unknown interception point {}",
                     method.name
                 )
-                return method.invoke(proxy, *(args!!))
+                return method.invoke(proxy, *(args))
             }
-            val invocation = args!![0] as InvocationInterceptor.Invocation<*>?
-            val invocationContext = args[1] as ReflectiveInvocationContext<*>?
+            val invocation = args[0] as InvocationInterceptor.Invocation<*>?
+            val invocationContext = args[1] as ReflectiveInvocationContext<*>
             val extensionContext = args[2] as ExtensionContext?
             val methodName = method.name
-            val logLineSuffix: String?
-            val methodMatcher = methodPattern!!.matcher(methodName)
+            val logLineSuffix: String
+            val methodMatcher = methodPattern.matcher(methodName)
             if (methodName == "interceptDynamicTest") {
                 logLineSuffix =
                     "execution of DynamicTest %s".formatted(extensionContext!!.displayName)
@@ -66,12 +66,19 @@ class LoggingInvocationInterceptor : InvocationInterceptor {
                     "instance creation for %s".formatted(invocationContext!!.targetClass)
             } else if (methodMatcher.matches()) {
                 val interceptedEvent = methodMatcher.group(1)
+                val methodRealClassName = invocationContext!!.executable!!.declaringClass.simpleName
+                val methodName = invocationContext.executable!!.name
+                val targetClassName = invocationContext!!.targetClass.simpleName
+                val methodDisplayName =
+                    if (targetClassName == methodRealClassName) methodName
+                    else "$methodName($methodRealClassName)"
                 logLineSuffix =
                     "execution of @%s method %s.%s".formatted(
                         interceptedEvent,
-                        invocationContext!!.executable!!.declaringClass.simpleName,
-                        invocationContext.executable!!.name
+                        targetClassName,
+                        methodDisplayName
                     )
+                TestContext.CURRENT_TEST_NAME.set("$targetClassName.$methodName")
             } else {
                 logLineSuffix = "execution of unknown intercepted call %s".formatted(methodName)
             }
@@ -81,15 +88,15 @@ class LoggingInvocationInterceptor : InvocationInterceptor {
             try {
                 val timeout = getTimeout(invocationContext)
                 if (timeout != null) {
-                    LOGGER!!.info(
+                    LOGGER.info(
                         "Junit starting {} with timeout of {}",
                         logLineSuffix,
                         DurationFormatUtils.formatDurationWords(timeout.toMillis(), true, true)
                     )
                     Timer("TimeoutTimer-" + currentThread.name, true)
                         .schedule(timeoutTask, timeout.toMillis())
                 } else {
-                    LOGGER!!.warn("Junit starting {} with no timeout", logLineSuffix)
+                    LOGGER.warn("Junit starting {} with no timeout", logLineSuffix)
                 }
                 val retVal = invocation!!.proceed()
                 val elapsedMs = Duration.between(start, Instant.now()).toMillis()
@@ -136,7 +143,7 @@ class LoggingInvocationInterceptor : InvocationInterceptor {
                     }
                 }
                 val stackTrace = StringUtils.join(stackToDisplay, "\n    ")
-                LOGGER!!.error(
+                LOGGER.error(
                     "Junit exception throw during {} after {}:\n{}",
                     logLineSuffix,
                     DurationFormatUtils.formatDurationWords(elapsedMs, true, true),
@@ -145,24 +152,29 @@ class LoggingInvocationInterceptor : InvocationInterceptor {
                 throw t1
             } finally {
                 timeoutTask.cancel()
+                TestContext.CURRENT_TEST_NAME.set(null)
             }
         }
 
-        private class TimeoutInteruptor(private val parentThread: Thread?) : TimerTask() {
+        private class TimeoutInteruptor(private val parentThread: Thread) : TimerTask() {
             @Volatile var wasTriggered: Boolean = false
 
             override fun run() {
+                LOGGER.info(
+                    "interrupting running task on ${parentThread.name}. Current Stacktrace is ${parentThread.stackTrace.asList()}"
+                )
                 wasTriggered = true
-                parentThread!!.interrupt()
+                parentThread.interrupt()
             }
 
             override fun cancel(): Boolean {
+                LOGGER.info("cancelling timer task on ${parentThread.name}")
                 return super.cancel()
             }
         }
 
         companion object {
-            private val methodPattern: Pattern? = Pattern.compile("intercept(.*)Method")
+            private val methodPattern: Pattern = Pattern.compile("intercept(.*)Method")
 
             private val PATTERN: Pattern =
                 Pattern.compile(
@@ -201,11 +213,11 @@ class LoggingInvocationInterceptor : InvocationInterceptor {
                 )
             }
 
-            private fun getTimeout(invocationContext: ReflectiveInvocationContext<*>?): Duration? {
+            private fun getTimeout(invocationContext: ReflectiveInvocationContext<*>): Duration {
                 var timeout: Duration? = null
-                var m = invocationContext!!.executable
+                var m = invocationContext.executable
                 if (m is Method) {
-                    var timeoutAnnotation: Timeout? = m.getAnnotation<Timeout?>(Timeout::class.java)
+                    var timeoutAnnotation: Timeout? = m.getAnnotation(Timeout::class.java)
                     if (timeoutAnnotation == null) {
                         timeoutAnnotation =
                             invocationContext.targetClass.getAnnotation(Timeout::class.java)
@@ -328,9 +340,9 @@ class LoggingInvocationInterceptor : InvocationInterceptor {
     }
 
     companion object {
-        private val LOGGER: Logger? =
+        private val LOGGER: Logger =
             LoggerFactory.getLogger(LoggingInvocationInterceptor::class.java)
-        private val JUNIT_METHOD_EXECUTION_TIMEOUT_PROPERTY_NAME: String? =
+        private val JUNIT_METHOD_EXECUTION_TIMEOUT_PROPERTY_NAME: String =
             "JunitMethodExecutionTimeout"
     }
 }

airbyte-cdk/java/airbyte-cdk/core/src/testFixtures/kotlin/io/airbyte/cdk/extensions/TestContext.kt

@@ -0,0 +1,9 @@
+/*
+ * Copyright (c) 2024 Airbyte, Inc., all rights reserved.
+ */
+
+package io.airbyte.cdk.extensions
+
+object TestContext {
+    val CURRENT_TEST_NAME: ThreadLocal<String?> = ThreadLocal()
+}

airbyte-cdk/java/airbyte-cdk/db-destinations/src/testFixtures/kotlin/io/airbyte/cdk/integrations/standardtest/destination/DestinationAcceptanceTest.kt

@@ -1469,7 +1469,7 @@ abstract class DestinationAcceptanceTest {
     }
 
     /** Whether the destination should be tested against different namespaces. */
-    protected open fun supportNamespaceTest(): Boolean {
+    open protected fun supportNamespaceTest(): Boolean {
         return false
     }
 
@@ -1571,19 +1571,21 @@ abstract class DestinationAcceptanceTest {
     }
 
     protected val destination: AirbyteDestination
-        get() =
-            DefaultAirbyteDestination(
-                AirbyteIntegrationLauncher(
-                    JOB_ID,
-                    JOB_ATTEMPT,
-                    imageName,
-                    processFactory,
-                    null,
-                    null,
-                    false,
-                    EnvVariableFeatureFlags()
-                )
+        get() {
+            return DefaultAirbyteDestination(
+                integrationLauncher =
+                    AirbyteIntegrationLauncher(
+                        JOB_ID,
+                        JOB_ATTEMPT,
+                        imageName,
+                        processFactory,
+                        null,
+                        null,
+                        false,
+                        EnvVariableFeatureFlags()
+                    )
             )
+        }
 
     @Throws(Exception::class)
     protected fun runSyncAndVerifyStateOutput(