Connection not closed when parent coroutine timeout

[namjug-kim]

I wanted to give timeout when I used the connection through MutinySessionFactory.
This is a reproduction code for reproducing a problem situation. In fact, Netty-based API servers are used with timeout, and Connection leaks occur.
In this situation, pool hang forever.

@Test
fun testTimeout() {
        println("execute")
        try {
            runBlocking {
                withTimeout(Duration.ofMillis(150)) {
                    try {
                        val withStatelessSession = sessionFactory.withStatelessSession { session ->
                            val createNativeQuery = session.createNativeQuery<Long>("SELECT COUNT(1) FROM User")
                            createNativeQuery.singleResultOrNull
                        }
                        println("result : ${withStatelessSession.awaitSuspending()}")
                    } finally {
                        println("finally called")
                    }
                }
            }
        } catch (e: Exception) {
            e.printStackTrace()
        }
        CountDownLatch(1).await()
}

expected action is to close connection as shown in the log below.

[] 2023-01-03 15:19:58,209 DEBUG [vert.x-eventloop-thread-0] o.h.r.j.i.LogicalConnectionMan: Initiating JDBC connection release from afterTransaction
[] 2023-01-03 15:19:58,210 TRACE [vert.x-eventloop-thread-0] o.h.r.p.i.SqlClientConnection: Connection closed: io.vertx.mysqlclient.impl.MySQLConnectionImpl@39e6b6f7

However, the actual operation did not close Connection.

execute
[] 2023-01-03 15:15:31,600 DEBUG [Test worker @coroutine#1] o.h.r.m.i.MutinySessionFactory: No existing open Mutiny.StatelessSession was found in the current Vert.x context: opening a new instance
[] 2023-01-03 15:15:31,632 DEBUG [vert.x-eventloop-thread-0] i.n.channel.DefaultChannelId: -Dio.netty.processId: 54046 (auto-detected)
[] 2023-01-03 15:15:31,634 DEBUG [vert.x-eventloop-thread-0] i.n.channel.DefaultChannelId: -Dio.netty.machineId: f8:4d:89:ff:fe:69:71:a4 (auto-detected)
[] 2023-01-03 15:15:31,648 DEBUG [vert.x-eventloop-thread-0] i.n.b.PooledByteBufAllocator: -Dio.netty.allocator.numHeapArenas: 20
[] 2023-01-03 15:15:31,648 DEBUG [vert.x-eventloop-thread-0] i.n.b.PooledByteBufAllocator: -Dio.netty.allocator.numDirectArenas: 20
[] 2023-01-03 15:15:31,648 DEBUG [vert.x-eventloop-thread-0] i.n.b.PooledByteBufAllocator: -Dio.netty.allocator.pageSize: 8192
[] 2023-01-03 15:15:31,648 DEBUG [vert.x-eventloop-thread-0] i.n.b.PooledByteBufAllocator: -Dio.netty.allocator.maxOrder: 9
[] 2023-01-03 15:15:31,648 DEBUG [vert.x-eventloop-thread-0] i.n.b.PooledByteBufAllocator: -Dio.netty.allocator.chunkSize: 4194304
[] 2023-01-03 15:15:31,648 DEBUG [vert.x-eventloop-thread-0] i.n.b.PooledByteBufAllocator: -Dio.netty.allocator.smallCacheSize: 256
[] 2023-01-03 15:15:31,648 DEBUG [vert.x-eventloop-thread-0] i.n.b.PooledByteBufAllocator: -Dio.netty.allocator.normalCacheSize: 64
[] 2023-01-03 15:15:31,648 DEBUG [vert.x-eventloop-thread-0] i.n.b.PooledByteBufAllocator: -Dio.netty.allocator.maxCachedBufferCapacity: 32768
[] 2023-01-03 15:15:31,648 DEBUG [vert.x-eventloop-thread-0] i.n.b.PooledByteBufAllocator: -Dio.netty.allocator.cacheTrimInterval: 8192
[] 2023-01-03 15:15:31,648 DEBUG [vert.x-eventloop-thread-0] i.n.b.PooledByteBufAllocator: -Dio.netty.allocator.cacheTrimIntervalMillis: 0
[] 2023-01-03 15:15:31,648 DEBUG [vert.x-eventloop-thread-0] i.n.b.PooledByteBufAllocator: -Dio.netty.allocator.useCacheForAllThreads: false
[] 2023-01-03 15:15:31,648 DEBUG [vert.x-eventloop-thread-0] i.n.b.PooledByteBufAllocator: -Dio.netty.allocator.maxCachedByteBuffersPerChunk: 1023
[] 2023-01-03 15:15:31,653 DEBUG [vert.x-eventloop-thread-0] io.netty.buffer.ByteBufUtil: -Dio.netty.allocator.type: pooled
[] 2023-01-03 15:15:31,653 DEBUG [vert.x-eventloop-thread-0] io.netty.buffer.ByteBufUtil: -Dio.netty.threadLocalDirectBufferSize: 0
[] 2023-01-03 15:15:31,653 DEBUG [vert.x-eventloop-thread-0] io.netty.buffer.ByteBufUtil: -Dio.netty.maxThreadLocalCharBufferSize: 16384
[] 2023-01-03 15:15:31,696 DEBUG [vert.x-eventloop-thread-0] io.netty.util.Recycler: -Dio.netty.recycler.maxCapacityPerThread: 4096
[] 2023-01-03 15:15:31,696 DEBUG [vert.x-eventloop-thread-0] io.netty.util.Recycler: -Dio.netty.recycler.ratio: 8
[] 2023-01-03 15:15:31,696 DEBUG [vert.x-eventloop-thread-0] io.netty.util.Recycler: -Dio.netty.recycler.chunkSize: 32
[] 2023-01-03 15:15:31,696 DEBUG [vert.x-eventloop-thread-0] io.netty.util.Recycler: -Dio.netty.recycler.blocking: false
[] 2023-01-03 15:15:31,699 DEBUG [vert.x-eventloop-thread-0] i.netty.buffer.AbstractByteBuf: -Dio.netty.buffer.checkAccessible: true
[] 2023-01-03 15:15:31,699 DEBUG [vert.x-eventloop-thread-0] i.netty.buffer.AbstractByteBuf: -Dio.netty.buffer.checkBounds: true
[] 2023-01-03 15:15:31,699 DEBUG [vert.x-eventloop-thread-0] i.n.u.ResourceLeakDetectorFact: Loaded default ResourceLeakDetector: io.netty.util.ResourceLeakDetector@78c7da6e
[] 2023-01-03 15:15:31,716 TRACE [vert.x-eventloop-thread-0] o.h.r.p.i.SqlClientConnection: Connection created: io.vertx.mysqlclient.impl.MySQLConnectionImpl@39e6b6f7
[] 2023-01-03 15:15:31,753 DEBUG [vert.x-eventloop-thread-0] o.h.s.i.StatisticsInitiator: Statistics initialized [enabled=false]
finally called
kotlinx.coroutines.TimeoutCancellationException: Timed out waiting for 150 ms
	(Coroutine boundary)
	at com.TestService$name$1$1.invokeSuspend(TestService.kt:47)
	at com.TestService$name$1.invokeSuspend(TestService.kt:41)
	(Coroutine creation stacktrace)
	at kotlin.coroutines.intrinsics.IntrinsicsKt__IntrinsicsJvmKt.createCoroutineUnintercepted(IntrinsicsJvm.kt:122)
	at kotlinx.coroutines.intrinsics.CancellableKt.startCoroutineCancellable(Cancellable.kt:30)
	at kotlinx.coroutines.BuildersKt__BuildersKt.runBlocking$default(Builders.kt:38)
	at kotlinx.coroutines.BuildersKt.runBlocking$default(Unknown Source)
	at com.TestService.name(TestService.kt:40)
	at java.base/jdk.internal.reflect.DirectMethodHandleAccessor.invoke(DirectMethodHandleAccessor.java:104)
	at java.base/java.lang.reflect.Method.invoke(Method.java:578)
	at org.junit.platform.commons.util.ReflectionUtils.invokeMethod(ReflectionUtils.java:725)
	at org.junit.jupiter.engine.execution.MethodInvocation.proceed(MethodInvocation.java:60)
	at org.junit.jupiter.engine.execution.InvocationInterceptorChain$ValidatingInvocation.proceed(InvocationInterceptorChain.java:131)
	at org.junit.jupiter.engine.extension.TimeoutExtension.intercept(TimeoutExtension.java:149)
	at org.junit.jupiter.engine.extension.TimeoutExtension.interceptTestableMethod(TimeoutExtension.java:140)
	at org.junit.jupiter.engine.extension.TimeoutExtension.interceptTestMethod(TimeoutExtension.java:84)
	at org.junit.jupiter.engine.execution.ExecutableInvoker$ReflectiveInterceptorCall.lambda$ofVoidMethod$0(ExecutableInvoker.java:115)
	at org.junit.jupiter.engine.execution.ExecutableInvoker.lambda$invoke$0(ExecutableInvoker.java:105)
	at org.junit.jupiter.engine.execution.InvocationInterceptorChain$InterceptedInvocation.proceed(InvocationInterceptorChain.java:106)
	at org.junit.jupiter.engine.execution.InvocationInterceptorChain.proceed(InvocationInterceptorChain.java:64)
	at org.junit.jupiter.engine.execution.InvocationInterceptorChain.chainAndInvoke(InvocationInterceptorChain.java:45)
	at org.junit.jupiter.engine.execution.InvocationInterceptorChain.invoke(InvocationInterceptorChain.java:37)
	at org.junit.jupiter.engine.execution.ExecutableInvoker.invoke(ExecutableInvoker.java:104)
	at org.junit.jupiter.engine.execution.ExecutableInvoker.invoke(ExecutableInvoker.java:98)
	at org.junit.jupiter.engine.descriptor.TestMethodTestDescriptor.lambda$invokeTestMethod$7(TestMethodTestDescriptor.java:214)
	at org.junit.platform.engine.support.hierarchical.ThrowableCollector.execute(ThrowableCollector.java:73)
	at org.junit.jupiter.engine.descriptor.TestMethodTestDescriptor.invokeTestMethod(TestMethodTestDescriptor.java:210)
	at org.junit.jupiter.engine.descriptor.TestMethodTestDescriptor.execute(TestMethodTestDescriptor.java:135)
	at org.junit.jupiter.engine.descriptor.TestMethodTestDescriptor.execute(TestMethodTestDescriptor.java:66)
	at org.junit.platform.engine.support.hierarchical.NodeTestTask.lambda$executeRecursively$6(NodeTestTask.java:151)
	at org.junit.platform.engine.support.hierarchical.ThrowableCollector.execute(ThrowableCollector.java:73)
	at org.junit.platform.engine.support.hierarchical.NodeTestTask.lambda$executeRecursively$8(NodeTestTask.java:141)
	at org.junit.platform.engine.support.hierarchical.Node.around(Node.java:137)
	at org.junit.platform.engine.support.hierarchical.NodeTestTask.lambda$executeRecursively$9(NodeTestTask.java:139)
	at org.junit.platform.engine.support.hierarchical.ThrowableCollector.execute(ThrowableCollector.java:73)
	at org.junit.platform.engine.support.hierarchical.NodeTestTask.executeRecursively(NodeTestTask.java:138)
	at org.junit.platform.engine.support.hierarchical.NodeTestTask.execute(NodeTestTask.java:95)
	at java.base/java.util.ArrayList.forEach(ArrayList.java:1511)
	at org.junit.platform.engine.support.hierarchical.SameThreadHierarchicalTestExecutorService.invokeAll(SameThreadHierarchicalTestExecutorService.java:41)
	at org.junit.platform.engine.support.hierarchical.NodeTestTask.lambda$executeRecursively$6(NodeTestTask.java:155)
	at org.junit.platform.engine.support.hierarchical.ThrowableCollector.execute(ThrowableCollector.java:73)
	at org.junit.platform.engine.support.hierarchical.NodeTestTask.lambda$executeRecursively$8(NodeTestTask.java:141)
	at org.junit.platform.engine.support.hierarchical.Node.around(Node.java:137)
	at org.junit.platform.engine.support.hierarchical.NodeTestTask.lambda$executeRecursively$9(NodeTestTask.java:139)
	at org.junit.platform.engine.support.hierarchical.ThrowableCollector.execute(ThrowableCollector.java:73)
	at org.junit.platform.engine.support.hierarchical.NodeTestTask.executeRecursively(NodeTestTask.java:138)
	at org.junit.platform.engine.support.hierarchical.NodeTestTask.execute(NodeTestTask.java:95)
	at java.base/java.util.ArrayList.forEach(ArrayList.java:1511)
	at org.junit.platform.engine.support.hierarchical.SameThreadHierarchicalTestExecutorService.invokeAll(SameThreadHierarchicalTestExecutorService.java:41)
	at org.junit.platform.engine.support.hierarchical.NodeTestTask.lambda$executeRecursively$6(NodeTestTask.java:155)
	at org.junit.platform.engine.support.hierarchical.ThrowableCollector.execute(ThrowableCollector.java:73)
	at org.junit.platform.engine.support.hierarchical.NodeTestTask.lambda$executeRecursively$8(NodeTestTask.java:141)
	at org.junit.platform.engine.support.hierarchical.Node.around(Node.java:137)
	at org.junit.platform.engine.support.hierarchical.NodeTestTask.lambda$executeRecursively$9(NodeTestTask.java:139)
	at org.junit.platform.engine.support.hierarchical.ThrowableCollector.execute(ThrowableCollector.java:73)
	at org.junit.platform.engine.support.hierarchical.NodeTestTask.executeRecursively(NodeTestTask.java:138)
	at org.junit.platform.engine.support.hierarchical.NodeTestTask.execute(NodeTestTask.java:95)
	at org.junit.platform.engine.support.hierarchical.SameThreadHierarchicalTestExecutorService.submit(SameThreadHierarchicalTestExecutorService.java:35)
	at org.junit.platform.engine.support.hierarchical.HierarchicalTestExecutor.execute(HierarchicalTestExecutor.java:57)
	at org.junit.platform.engine.support.hierarchical.HierarchicalTestEngine.execute(HierarchicalTestEngine.java:54)
	at org.junit.platform.launcher.core.EngineExecutionOrchestrator.execute(EngineExecutionOrchestrator.java:107)
	at org.junit.platform.launcher.core.EngineExecutionOrchestrator.execute(EngineExecutionOrchestrator.java:88)
	at org.junit.platform.launcher.core.EngineExecutionOrchestrator.lambda$execute$0(EngineExecutionOrchestrator.java:54)
	at org.junit.platform.launcher.core.EngineExecutionOrchestrator.withInterceptedStreams(EngineExecutionOrchestrator.java:67)
	at org.junit.platform.launcher.core.EngineExecutionOrchestrator.execute(EngineExecutionOrchestrator.java:52)
	at org.junit.platform.launcher.core.DefaultLauncher.execute(DefaultLauncher.java:114)
	at org.junit.platform.launcher.core.DefaultLauncher.execute(DefaultLauncher.java:86)
	at org.junit.platform.launcher.core.DefaultLauncherSession$DelegatingLauncher.execute(DefaultLauncherSession.java:86)
	at org.junit.platform.launcher.core.SessionPerRequestLauncher.execute(SessionPerRequestLauncher.java:53)
	at org.gradle.api.internal.tasks.testing.junitplatform.JUnitPlatformTestClassProcessor$CollectAllTestClassesExecutor.processAllTestClasses(JUnitPlatformTestClassProcessor.java:99)
	at org.gradle.api.internal.tasks.testing.junitplatform.JUnitPlatformTestClassProcessor$CollectAllTestClassesExecutor.access$000(JUnitPlatformTestClassProcessor.java:79)
	at org.gradle.api.internal.tasks.testing.junitplatform.JUnitPlatformTestClassProcessor.stop(JUnitPlatformTestClassProcessor.java:75)
	at org.gradle.api.internal.tasks.testing.SuiteTestClassProcessor.stop(SuiteTestClassProcessor.java:61)
	at java.base/jdk.internal.reflect.DirectMethodHandleAccessor.invoke(DirectMethodHandleAccessor.java:104)
	at java.base/java.lang.reflect.Method.invoke(Method.java:578)
	at org.gradle.internal.dispatch.ReflectionDispatch.dispatch(ReflectionDispatch.java:36)
	at org.gradle.internal.dispatch.ReflectionDispatch.dispatch(ReflectionDispatch.java:24)
	at org.gradle.internal.dispatch.ContextClassLoaderDispatch.dispatch(ContextClassLoaderDispatch.java:33)
	at org.gradle.internal.dispatch.ProxyDispatchAdapter$DispatchingInvocationHandler.invoke(ProxyDispatchAdapter.java:94)
	at jdk.proxy1/jdk.proxy1.$Proxy2.stop(Unknown Source)
	at org.gradle.api.internal.tasks.testing.worker.TestWorker$3.run(TestWorker.java:193)
	at org.gradle.api.internal.tasks.testing.worker.TestWorker.executeAndMaintainThreadName(TestWorker.java:129)
	at org.gradle.api.internal.tasks.testing.worker.TestWorker.execute(TestWorker.java:100)
	at org.gradle.api.internal.tasks.testing.worker.TestWorker.execute(TestWorker.java:60)
	at org.gradle.process.internal.worker.child.ActionExecutionWorker.execute(ActionExecutionWorker.java:56)
	at org.gradle.process.internal.worker.child.SystemApplicationClassLoaderWorker.call(SystemApplicationClassLoaderWorker.java:133)
	at org.gradle.process.internal.worker.child.SystemApplicationClassLoaderWorker.call(SystemApplicationClassLoaderWorker.java:71)
	at worker.org.gradle.process.internal.worker.GradleWorkerMain.run(GradleWorkerMain.java:69)
	at worker.org.gradle.process.internal.worker.GradleWorkerMain.main(GradleWorkerMain.java:74)
Caused by: kotlinx.coroutines.TimeoutCancellationException: Timed out waiting for 150 ms
	at kotlinx.coroutines.TimeoutKt.TimeoutCancellationException(Timeout.kt:184)
	at kotlinx.coroutines.TimeoutCoroutine.run(Timeout.kt:154)
	at kotlinx.coroutines.EventLoopImplBase$DelayedRunnableTask.run(EventLoop.common.kt:508)
	at kotlinx.coroutines.EventLoopImplBase.processNextEvent(EventLoop.common.kt:284)
	at kotlinx.coroutines.DefaultExecutor.run(DefaultExecutor.kt:108)
	at java.base/java.lang.Thread.run(Thread.java:1589)
[] 2023-01-03 15:15:31,765 DEBUG [vert.x-eventloop-thread-0] o.h.e.j.spi.SqlStatementLogger: SELECT COUNT(1) FROM User
[] 2023-01-03 15:15:31,787 DEBUG [vert.x-eventloop-thread-0] o.j.l.DelegatingBasicLogger: Result set row: 0
[] 2023-01-03 15:15:31,787 DEBUG [vert.x-eventloop-thread-0] org.hibernate.loader.Loader: Result row: 
[] 2023-01-03 15:15:31,791 DEBUG [vert.x-eventloop-thread-0] o.h.r.j.i.LogicalConnectionMan: Initiating JDBC connection release from afterTransaction

[namjug-kim]

in MutinySessionFactoryImpl#withSession theres no Uni#onCancellation, is this intended?

// MutinySessionFactoryImpl
	private<S extends Mutiny.Closeable, T> Uni<T> withSession(
			Uni<S> sessionUni,
			Function<S, Uni<T>> work,
			Context.Key<S> contextKey) {
		return sessionUni.chain( session -> Uni.createFrom().voidItem()
				.invoke( () -> context.put( contextKey, session ) )
				.chain( () -> work.apply( session ) )
				.eventually( () -> context.remove( contextKey ) )
				.eventually(session::close)
		);
	}

// MutinyStatelessSessionImpl
		Uni<T> executeInTransaction(Function<Mutiny.Transaction, Uni<T>> work) {
			return work.apply( this )
					// in the case of an exception or cancellation
					// we need to rollback the transaction
					.onFailure().call( this::rollback )
					.onCancellation().call( this::rollback )
					// finally, when there was no exception,
					// commit or rollback the transaction
					.call( () -> rollback ? rollback() : commit() );
		}

I saw that Transaction has processing for onCacellation. Is there a best practice to handle cancel event in (withSession, withTransaction)?

[DavideD]

in MutinySessionFactoryImpl#withSession theres no Uni#onCancellation, is this intended?

Mmmh... that's a good point. I haven't thought of it.
I was expecting that using eventually would cover all the basis, but I haven't tested what happens with onCancellation.
We will have to check.

[gavinking]

It seems to me that eventually() should do what you expect here, and if it doesn’t then that’s more like a bug.

[namjug-kim]

val cancellable = sessionFactory
    .withStatelessSession { it.createNativeQuery<Long>("SELECT COUNT(1) FROM User").singleResultOrNull }
    .subscribe()
    .with { println("success : $it") }

val timer = Timer()
timer.schedule(object : TimerTask() {
    override fun run() {
        cancellable.cancel()
    }
}, 150)

CountDownLatch(1).await()

eventually() not invoked in this test..

In uni eventually() theres no process for cancel or terminate

This method is a shortcut for UniOnItemOrFailure.call(BiFunction): onItemOrFailure().call((item, err) -> supplier.get())

[DavideD]

I'm going to create an issue for Vert.x, but in the meantime, we should fix this.

[DavideD]

Sorry, I mean mutiny (not Vert.x)

[DavideD]

@cescoffier, @jponge What do you think?

[namjug-kim]

smallrye/smallrye-mutiny#178

Looking at the issue of discussion about eventually api, the omission of terminate and cancel in eventually seems to be intended.

[namjug-kim]

    private <S extends Mutiny.Closeable, T> Uni<T> withSession(
            Uni<S> sessionUni,
            Function<S, Uni<T>> work,
            Context.Key<S> contextKey) {
        AtomicBoolean connectionCloseCalled = new AtomicBoolean(false);
        return sessionUni.chain(session -> Uni.createFrom().voidItem()
                .invoke(() -> context.put(contextKey, session))
                .chain(() -> work.apply(session))
                .eventually(() -> {
                    if (!connectionCloseCalled.getAndSet(true)) {
                        return Uni.createFrom().voidItem()
                                .invoke(() -> context.remove(contextKey))
                                .chain(session::close);
                    } else {
                        return Uni.createFrom().voidItem();
                    }
                })
                .onCancellation().call(() -> {
                    if (!connectionCloseCalled.getAndSet(true)) {
                        return Uni.createFrom().voidItem()
                                .invoke(() -> context.remove(contextKey))
                                .chain(session::close);
                    } else {
                        return Uni.createFrom().voidItem();
                    }
                })
        );
    }

I tested it by adding the action of closing the session to onCancellation()

I expected it work well. But, when the pool is full and enter the queue, the connection from the queue stuck connection pool.
onCancellation() was executed on the thread that called cancel.

If cancel inside vertx context it works fine for me :)

[gavinking]

Well it might have been intended but from this discussion it doesn’t seem like the current behavior actually achieves what you would want it to do from the user point of view.

[gavinking]

That is to say: it’s supposed to be a finally

[cescoffier]

eventually() is not called on cancellation.
If you want cancellation, you need: onTermination().invoke(....), also called on cancellation.

Note that when this callback is invoked following a cancellation, it will be called on the thread having called cancel().

[cescoffier]

(The question about whether eventually should be called on cancellation is something we need to think about)

[gavinking]

But then Clement, naively, that seems to make eventually() useless and misleading, since it can’t be used to do cleanup, which would seem to be its whole purpose. I would predict that almost everyone who calls that method expects it to be “always called”, and so almost every usage of it is actually broken.

Am I missing something here? What is the use case for a “finally” which sometimes doesn’t get called?