Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Kotlin coroutines instrumentation causes IllegalStateException sometimes #4718

Closed
monosoul opened this issue Feb 11, 2023 · 2 comments
Closed
Milestone

Comments

@monosoul
Copy link
Contributor

monosoul commented Feb 11, 2023

Enabling coroutines instrumentation causes IllegalStateException in kotlinx.coroutines.flow.internal.SafeCollector sometimes.

E.g.:
Stacktrace:

java.lang.IllegalStateException: Flow invariant is violated:
		Flow was collected in [io.ktor.server.engine.DefaultUncaughtExceptionHandler@205ac27b, CoroutineName(call-handler), io.ktor.server.netty.NettyDispatcher$CurrentContext@41c6aff0, datadog.trace.instrumentation.kotlin.coroutines.ScopeStateCoroutineContext@4665ebe3, ScopeCoroutine{Active}@3a4b0a58, NettyDispatcher@6f2a5663],
		but emission happened in [io.ktor.server.engine.DefaultUncaughtExceptionHandler@205ac27b, CoroutineName(call-handler), io.ktor.server.netty.NettyDispatcher$CurrentContext@41c6aff0, datadog.trace.instrumentation.kotlin.coroutines.ScopeStateCoroutineContext@433755fa, ScopeCoroutine{Active}@377acb43, NettyDispatcher@6f2a5663].
		Please refer to 'flow' documentation or use 'flowOn' instead
	at kotlinx.coroutines.flow.internal.SafeCollector_commonKt.checkContext(SafeCollector.common.kt:85)
	at kotlinx.coroutines.flow.internal.SafeCollector.checkContext(SafeCollector.kt:106)
	at kotlinx.coroutines.flow.internal.SafeCollector.emit(SafeCollector.kt:83)
	at kotlinx.coroutines.flow.internal.SafeCollector.emit(SafeCollector.kt:66)
	at io.grpc.kotlin.ClientCalls$rpcImpl$1$1.invokeSuspend(ClientCalls.kt:320)
	at kotlin.coroutines.jvm.internal.BaseContinuationImpl.resumeWith(ContinuationImpl.kt:33)
	at kotlinx.coroutines.DispatchedTask.run(DispatchedTask.kt:106)
	at io.netty.util.concurrent.PromiseTask.runTask(PromiseTask.java:98)
	at io.netty.util.concurrent.PromiseTask.run(PromiseTask.java:106)
	at io.netty.util.concurrent.AbstractEventExecutor.runTask(AbstractEventExecutor.java:174)
	at io.netty.util.concurrent.AbstractEventExecutor.safeExecute(AbstractEventExecutor.java:167)
	at io.netty.util.concurrent.SingleThreadEventExecutor.runAllTasks(SingleThreadEventExecutor.java:470)
	at io.netty.channel.nio.NioEventLoop.run(NioEventLoop.java:569)
	at io.netty.util.concurrent.SingleThreadEventExecutor$4.run(SingleThreadEventExecutor.java:997)
	at io.netty.util.internal.ThreadExecutorMap$2.run(ThreadExecutorMap.java:74)
	at io.ktor.server.netty.EventLoopGroupProxy$Companion.create$lambda$1$lambda$0(NettyApplicationEngine.kt:291)
	at io.netty.util.concurrent.FastThreadLocalRunnable.run(FastThreadLocalRunnable.java:30)
	at java.base/java.lang.Thread.run(Thread.java:833)

Another example: #931 (comment)

@monosoul
Copy link
Contributor Author

KotlinCoroutineTests patch to consistently reproduce the issue:

diff --git a/dd-java-agent/instrumentation/kotlin-coroutines/coroutines-1.5/src/test/kotlin/KotlinCoroutineTests.kt b/dd-java-agent/instrumentation/kotlin-coroutines/coroutines-1.5/src/test/kotlin/KotlinCoroutineTests.kt
--- a/dd-java-agent/instrumentation/kotlin-coroutines/coroutines-1.5/src/test/kotlin/KotlinCoroutineTests.kt	(revision 90f8f57e8b04e9f7dd6f1ac742b3a78e339bd32d)
+++ b/dd-java-agent/instrumentation/kotlin-coroutines/coroutines-1.5/src/test/kotlin/KotlinCoroutineTests.kt	(date 1676106271186)
@@ -6,6 +6,7 @@
 import kotlinx.coroutines.flow.flow
 import kotlinx.coroutines.flow.flowOn
 import kotlinx.coroutines.launch
+import kotlinx.coroutines.withTimeout
 
 @SuppressFBWarnings("NP_PARAMETER_MUST_BE_NONNULL_BUT_MARKED_AS_NULLABLE")
 class KotlinCoroutineTests(dispatcher: CoroutineDispatcher) : CoreKotlinCoroutineTests(dispatcher) {
@@ -15,7 +16,9 @@
     val producer = flow {
       repeat(3) {
         tracedChild("produce_$it")
-        emit(it)
+        withTimeout(100) {
+          emit(it)
+        }
       }
     }.flowOn(jobName("producer"))
 

@github-actions
Copy link
Contributor

🤖 This issue has been addressed in the latest release. See full details in the Release Notes.

Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment
Projects
None yet
Development

No branches or pull requests

2 participants